udeServer/internal/logic/createservicelogic.go

// udeServer: CreateService 走事务 —— 同一事务内插 t_service + N 条 t_directory,
// 任意一条 INSERT 失败,全部回滚,不留脏数据。

package logic

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"path"
	"strings"
	"time"

	"udeServer/internal/model"
	"udeServer/internal/svc"
	"udeServer/internal/types"

	"github.com/zeromicro/go-zero/core/logx"
	"github.com/zeromicro/go-zero/core/stores/sqlx"
)

type CreateServiceLogic struct {
	logx.Logger
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

func NewCreateServiceLogic(ctx context.Context, svcCtx *svc.ServiceContext) *CreateServiceLogic {
	return &CreateServiceLogic{
		Logger: logx.WithContext(ctx),
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

func (l *CreateServiceLogic) CreateService(req *types.CreateServiceReq) (*types.ServiceResp, error) {
	var svcId int64
	var dirList []string // 落库成功的保护目录(trim 后非空),事务提交后用于通知外挂程序挂载
	err := l.svcCtx.DB.Transact(func(session sqlx.Session) error {
		// 把 model 绑到事务 session 上,后续所有写操作都走同一个连接/事务
		svcModel := l.svcCtx.ServiceModel.WithSession(session)
		dirModel := l.svcCtx.DirectoryModel.WithSession(session)

		// 1. 插服务
		svcData := &model.TService{
			Name:        nullString(req.Name),
			Ip:          nullString(req.Ip),
			Port:        req.Port, // 端口 int 类型,int64 直传(0=未配置)
			Iden:        nullString(req.Iden),
			KeyIden:     nullString(req.KeyIden),
			KeyAlgo:     nullString(req.KeyAlgo),
			Status:      nullString(req.Status),
			MountStatus: nullString("unmounted"), // 挂载状态初始值:未挂载(表列 NOT NULL,必须显式传值,否则 NULL 会报错)
			Brush:       req.Brush,               // 一键刷库配置原样落库(0=否默认,1=创建后自动下发刷库)
		}
		svcRes, err := svcModel.Insert(l.ctx, svcData)
		if err != nil {
			return fmt.Errorf("insert t_service: %w", err)
		}
		svcId, err = svcRes.LastInsertId()
		if err != nil {
			return fmt.Errorf("get last insert id: %w", err)
		}

		// 2. 插保护目录(与上面的服务同事务,任一条失败整体回滚)
		for _, dir := range req.Directories {
			path := strings.TrimSpace(dir)
			if path == "" {
				continue
			}
			dirList = append(dirList, path)
			_, err := dirModel.Insert(l.ctx, &model.TDirectory{
				ServiceId: nullInt64(svcId),
				Directory: nullString(path),
			})
			if err != nil {
				return fmt.Errorf("insert t_directory[%s]: %w", path, err)
			}
		}
		return nil
	})
	if err != nil {
		l.Errorf("create service tx failed: %v", err)
		return nil, err
	}

	// 事务提交后再回查完整实体(CreateAt 由 DB 默认值生成)
	created, err := findService(l.ctx, l.svcCtx.ServiceModel, svcId)
	if err != nil {
		return nil, err
	}
	// 3. 落库成功后,通知外挂程序(req.Ip:req.Port)执行 fuse 挂载。
	//    挂载失败不阻塞服务创建:响应体带 mountWarning,前端弹黄色警告而非红色错误。
	resp := toServiceResp(created)
	// 4. 一键刷库(brush=1):挂载【之前】先下发。
	//    刷库语义 = 外挂把保护目录存量明文加密进 <目录>-cipher 并清空原明文,
	//    目录变空后才能通过外挂的空目录挂载校验(先刷后挂的编排顺序)。
	//    刷库失败软处理:聚合进警告后继续走挂载(空目录场景挂载不受影响)。
	var warnings []string
	if req.Brush == 1 {
		if err := l.notifyFuseBrush(created, dirList); err != nil {
			l.Errorf("service created(id=%d), but brush dispatch failed: %v", svcId, err)
			warnings = append(warnings, fmt.Sprintf("一键刷库: %v", err))
		}
	}

	// 5. 挂载结果落库 + 响应回显:
	//    成功 → mount_status=mounted;失败 → mount_status=failed(同时带 mountWarning 黄条)。
	//    状态落库失败只记日志,不影响主流程(服务创建本身已成功)。
	//    注:「没配置保护目录」目前也走 failed 分支,语义细分留给后续迭代。
	if err := l.notifyFuseMount(created, dirList); err != nil {
		l.Errorf("service created(id=%d), but fuse mount notify failed: %v", svcId, err)
		warnings = append(warnings, fmt.Sprintf("服务已创建,但通知外挂程序挂载失败: %v", err))
		resp.MountStatus = "failed"
		if uerr := l.svcCtx.ServiceModel.UpdateMountStatus(l.ctx, svcId, "failed"); uerr != nil {
			l.Errorf("update mount_status=failed failed(id=%d): %v", svcId, uerr)
		}
	} else {
		resp.MountStatus = "mounted"
		if uerr := l.svcCtx.ServiceModel.UpdateMountStatus(l.ctx, svcId, "mounted"); uerr != nil {
			l.Errorf("update mount_status=mounted failed(id=%d): %v", svcId, uerr)
		}
	}
	if len(warnings) > 0 {
		// strings.Join 把刷库/挂载的失败信息拼成一条,分号分隔,前端一条黄条展示
		resp.MountWarning = strings.Join(warnings, "; ")
	}

	return resp, nil
}

// notifyFuseBrush 下发一键刷库到外挂程序(svc.Ip:svc.Port)。
// 刷库语义:外挂读取保护目录存量明文 → 按服务密钥加密写入 <保护目录>-cipher
// → 清空保护目录原明文(数据已转移到密文目录),为挂载的空目录校验扫清障碍。
// 调用时机:创建流程中、挂载之前(目录非空时挂载会被空目录校验拒绝)。
// TODO: 外挂程序侧的刷库接口(建议 POST /api/brush,请求体 {mount,ikey} 与挂载同源)
// 开发完成后,在此填入真实下发实现(HTTP 客户端复用包级 fuseMountClient)。
// 当前返回固定 error,由调用方软处理成黄色警告,绝不阻塞服务创建。
func (l *CreateServiceLogic) notifyFuseBrush(svc *model.TService, dirs []string) error {
	if len(dirs) == 0 {
		return nil // 没配保护目录,无库可刷
	}
	if svc.Ip.String == "" || svc.Port == 0 { // port 已是 int64,0 视为未配置
		return fmt.Errorf("外挂服务 IP/端口为空,无法下发刷库")
	}
	// TODO: 逐目录调用外挂程序刷库接口(与 notifyFuseMount 同款循环模式)
	return fmt.Errorf("外挂程序刷库接口尚未实装,本轮未执行刷库(存量数据场景挂载会被空目录校验拒绝)")
}

// fuseMountClient 复用的 HTTP 客户端,10s 超时防止外挂程序无响应拖死创建接口
var fuseMountClient = &http.Client{Timeout: 10 * time.Second}

// notifyFuseMount 落库成功后回调外挂程序(svc.Ip:svc.Port)的 /api/fuseMount 接口,
// 由外挂程序对保护目录执行 fuse 挂载。一个目录一次调用,请求体:
//
//	POST http://{ip}:{port}/api/fuseMount
//	{ "backing": "<保护目录>-cipher", "mount": "<保护目录>", "ikey": "<密钥内容>" }
//
// ikey 取自 t_key.content(按服务配置的密钥标识 keyIden 查询)。
func (l *CreateServiceLogic) notifyFuseMount(svc *model.TService, dirs []string) error {
	ip, port := svc.Ip.String, svc.Port // port 已是 int64
	if ip == "" || port == 0 {
		return fmt.Errorf("外挂服务 IP/端口为空,无法下发挂载")
	}
	if len(dirs) == 0 {
		return fmt.Errorf("没有配置保护目录,无需通知挂载")
	}

	// 密钥内容:按服务配置的密钥标识(keyIden)查 t_key
	keyIden := svc.KeyIden.String
	if keyIden == "" {
		return fmt.Errorf("服务未配置密钥标识(keyIden),无法下发挂载")
	}
	key, err := l.svcCtx.KeyModel.FindOneByIden(l.ctx, keyIden)
	if err != nil {
		return fmt.Errorf("查询密钥(%s)失败: %w", keyIden, err)
	}
	ikey := key.Content.String
	if ikey == "" {
		return fmt.Errorf("密钥(%s)内容为空,无法下发挂载", keyIden)
	}

	baseURL := fmt.Sprintf("http://%s:%d/api/fuseMount", ip, port) // %d:port 是 int64,数字占位符
	for _, mount := range dirs {
		dirLinuxPath := path.Clean(mount)
		cipherDir := dirLinuxPath + "-cipher"
		payload := map[string]interface{}{
			"backing": cipherDir,    // 密文存储目录
			"mount":   dirLinuxPath, // 挂载点
			"ikey":    ikey,
		}
		body, err := json.Marshal(payload)
		if err != nil {
			return fmt.Errorf("序列化挂载请求失败: %w", err)
		}

		resp, err := fuseMountClient.Post(baseURL, "application/json", bytes.NewReader(body))
		if err != nil {
			return fmt.Errorf("请求 %s 失败(外挂程序未启动?): %w", baseURL, err)
		}
		if resp.StatusCode >= http.StatusBadRequest {
			snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 256))
			resp.Body.Close()
			return fmt.Errorf("挂载目录 %s 失败,外挂程序返回 HTTP %d: %s", mount, resp.StatusCode, string(snippet))
		}
		// 2. 成功处理分支：必须将剩余的 body 数据读入黑洞 (io.Discard)，以保证底层 TCP 连接池可以复用 Keep-Alive
		_, _ = io.Copy(io.Discard, resp.Body)
		resp.Body.Close()
		l.Infof("fuse mount ok: %s -> %s", mount, mount+"-cipher")
	}
	return nil
}