udeServer/internal/logic/createservicelogic.go

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

package logic

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"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:    nullString(req.Port),
			Iden:    nullString(req.Iden),
			KeyIden: nullString(req.KeyIden),
			KeyAlgo: nullString(req.KeyAlgo),
			Status:  nullString(req.Status),
		}
		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 挂载。
	//    外挂程序未启动/网络不通时挂载会失败:服务数据已落库不回滚,仅把错误抛给前端提示。
	if err := l.notifyFuseMount(created, dirList); err != nil {
		l.Errorf("service created(id=%d), but fuse mount notify failed: %v", svcId, err)
		return toServiceResp(created), fmt.Errorf("服务已创建,但通知外挂程序挂载失败: %v", err)
	}

	return toServiceResp(created), nil
}

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

// notifyFuseMount 落库成功后回调外挂程序(svc.Ip:svc.Port)的 /api/fuseMount 接口,
// 由外挂程序对保护目录执行 fuse 挂载。请求体为 JSON:
//
//	POST http://{ip}:{port}/api/fuseMount
//	{ "iden": "...", "keyIden": "...", "keyAlgo": "...", "directories": ["/data/fuse", ...] }
func (l *CreateServiceLogic) notifyFuseMount(svc *model.TService, dirs []string) error {
	ip, port := svc.Ip.String, svc.Port.String
	if ip == "" || port == "" {
		// 没配外挂服务地址:只落库,跳过挂载通知
		l.Infof("service %d has empty ip/port, skip fuse mount notify", svc.Id)
		return nil
	}
	if len(dirs) == 0 {
		return nil // 没有保护目录,无需通知挂载
	}

	url := fmt.Sprintf("http://%s:%s/api/fuseMount", ip, port)
	payload := map[string]interface{}{
		"iden":        svc.Iden.String,
		"keyIden":     svc.KeyIden.String,
		"keyAlgo":     svc.KeyAlgo.String,
		"directories": dirs,
	}
	body, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("序列化挂载请求失败: %w", err)
	}

	resp, err := fuseMountClient.Post(url, "application/json", bytes.NewReader(body))
	if err != nil {
		return fmt.Errorf("请求 %s 失败(外挂程序未启动?): %w", url, err)
	}
	defer resp.Body.Close()

	if resp.StatusCode >= http.StatusBadRequest {
		snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 256))
		return fmt.Errorf("外挂程序返回 HTTP %d: %s", resp.StatusCode, string(snippet))
	}
	return nil
}