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/audit"

	"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)的 /api/migrate 接口。
// 刷库语义:外挂读取保护目录存量明文 → 按服务密钥加密写入 <保护目录>-cipher
// → 校验通过后清空保护目录原明文(数据已转移到密文目录,磁盘上只留密文一份),
// 完成加密语义闭环,同时为挂载的空目录校验扫清障碍。
// 调用时机:创建流程中、挂载之前(先刷后挂编排,见 CreateService 第4步)。
// 实际下发逻辑抽成包级 dispatchFuseBrush,与手动补刷(BrushServiceLogic)共用。
func (l *CreateServiceLogic) notifyFuseBrush(svc *model.TService, dirs []string) error {
	return dispatchFuseBrush(l.ctx, l.svcCtx, svc, dirs)
}

// doFuseRequest 发送发往外挂程序的 HTTP 请求,统一携带逐服务的 X-TOKEN 鉴权头。
// X-TOKEN 的值 = 服务表 t_service 的 iden(接入标识)字段;外挂程序拿它与
// 自己环境变量 TOKEN 配的值恒定时间比对(见 FuseDemo 的 NewAuthInterceptor),
// 因此每台外挂机的 TOKEN 要配成"接在该机上的服务"的 iden 值。
// client 按场景传:挂载/卸载/目录浏览用 fuseMountClient(10s),刷库用 fuseBrushClient(10min)。
// 之所以不用 client.Post/client.Get 快捷方式:它们构造的请求没法逐个加 Header,
// 必须先 http.NewRequest 再 client.Do。
func doFuseRequest(client *http.Client, iden, method, url string, body io.Reader, contentType string) (*http.Response, error) {
	req, err := http.NewRequest(method, url, body)
	if err != nil {
		return nil, err
	}
	if contentType != "" {
		req.Header.Set("Content-Type", contentType)
	}
	req.Header.Set("X-TOKEN", iden)
	return client.Do(req)
}

// fuseBrushClient 刷库专用HTTP客户端:刷库是重操作(全量加密可能数分钟),
// 不能复用挂载的10s短超时,单独放宽到10分钟。
var fuseBrushClient = &http.Client{Timeout: 10 * time.Minute}

// dispatchFuseBrush 下发刷库到外挂(包级函数,创建流程与手动补刷共用同一实现)。
// 逐保护目录调用 POST /api/migrate,请求体:
//
//	{ "backing": "<目录>-cipher", "mount": ["<目录>"], "overwrite": true, "ikey": "<密钥内容>" }
//
// ikey 取自 t_key.content(按服务 keyIden 查询),与挂载下发同源,保证刷库密钥=挂载密钥。
// 外挂约定:部分文件失败也返回 HTTP 200,成败靠响应体的 failed 计数判断。
// 注:创建流程的自动刷库无需挂载中校验——创建时序是先刷后挂,刷库时必然未挂载;
// 手动补刷的挂载中拦截由调用方 BrushServiceLogic 负责。
func dispatchFuseBrush(ctx context.Context, svcCtx *svc.ServiceContext, svc *model.TService, dirs []string) error {
	if len(dirs) == 0 {
		return fmt.Errorf("没配保护目录,无库可刷")
	}
	ip, port := svc.Ip.String, svc.Port // port 是 int64
	if ip == "" || port == 0 {
		return fmt.Errorf("外挂服务 IP/端口为空,无法下发刷库")
	}
	keyIden := svc.KeyIden.String
	if keyIden == "" {
		return fmt.Errorf("服务未配置密钥标识(keyIden),无法下发刷库")
	}
	key, err := svcCtx.KeyModel.FindOneByIden(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/migrate", ip, port)
	for _, dir := range dirs {
		dirLinuxPath := path.Clean(dir)
		payload := map[string]interface{}{
			"backing":   dirLinuxPath + "-cipher", // 密文输出目录
			"mount":     []string{dirLinuxPath},   // 明文源目录(接口定义是数组,单目录传单元素)
			"overwrite": true,                     // 补刷时目标已存在则覆盖,保证刷库可重复执行
			"ikey":      ikey,
		}
		body, err := json.Marshal(payload)
		if err != nil {
			return fmt.Errorf("序列化刷库请求失败: %w", err)
		}
		resp, err := doFuseRequest(fuseBrushClient, svc.Iden.String, http.MethodPost, baseURL, bytes.NewReader(body), "application/json")
		if err != nil {
			return fmt.Errorf("请求 %s 失败(外挂程序未启动?): %w", baseURL, err)
		}
		// 先把响应体整体读出再Close,保证Keep-Alive连接可复用
		respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
		resp.Body.Close()
		if resp.StatusCode >= http.StatusBadRequest {
			return fmt.Errorf("刷库目录 %s 失败,外挂程序返回 HTTP %d: %s", dir, resp.StatusCode, string(respBody))
		}
		var result struct { // 匿名结构体:只用一次的临时JSON模型,不必单独定义类型
			Total  int `json:"total"`
			Failed int `json:"failed"`
		}
		if err := json.Unmarshal(respBody, &result); err != nil {
			return fmt.Errorf("解析刷库响应失败(目录 %s): %w", dir, err)
		}
		if result.Failed > 0 {
			return fmt.Errorf("刷库目录 %s: %d/%d 个文件失败", dir, result.Failed, result.Total)
		}
	}
	return nil
}

// 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)
		}

		// 发起 POST(DoSimple 记录 type=2 接口日志,父 Span 指向本次入站请求)
		if _, callErr := audit.DoSimple(l.svcCtx.SpanRecorder, l.ctx, "FuseMount", baseURL, body,
			func() (map[string]interface{}, int64, error) {
				resp, rerr := fuseMountClient.Post(baseURL, "application/json", bytes.NewReader(body))
				if rerr != nil {
					return nil, 0, fmt.Errorf("请求 %s 失败(外挂程序未启动?): %w", baseURL, rerr)
				}
				// 读到 EOF 保证底层 TCP 连接池可以复用 Keep-Alive
				raw, _ := io.ReadAll(resp.Body)
				resp.Body.Close()
				status := int64(resp.StatusCode)
				if status >= http.StatusBadRequest {
					return nil, status, fmt.Errorf("挂载目录 %s 失败,外挂程序返回 HTTP %d: %s", mount, status, string(raw))
				}
				return map[string]interface{}{"status": status, "body": string(raw)}, status, nil
			}); callErr != nil {
			return callErr
		}
		l.Infof("fuse mount ok: %s -> %s", mount, mount+"-cipher")
	}
	return nil
}