udeServer/internal/logic/systemlog/logarchiverestorelogic.go

// Code scaffolded by goctl. Safe to edit.
// goctl 1.10.2

package systemlog

import (
	"context"
	"database/sql"
	"fmt"
	"os"

	"udeServer/internal/apierr"
	"udeServer/internal/audit"
	"udeServer/internal/log"
	udeModel "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 LogArchiveRestoreLogic struct {
	logx.Logger
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

// 归档恢复
func NewLogArchiveRestoreLogic(ctx context.Context, svcCtx *svc.ServiceContext) *LogArchiveRestoreLogic {
	return &LogArchiveRestoreLogic{
		Logger: logx.WithContext(ctx),
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

// restoreResult carries the outcome of one restore goroutine.
type restoreResult struct {
	RecordId int64
	Success  bool
	ErrMsg   string
}

// LogArchiveRestore launches the restore of the selected archive runs in
// background goroutines (bounded to 5 concurrent) and answers immediately;
// progress is tracked through t_log_restore_record.
func (l *LogArchiveRestoreLogic) LogArchiveRestore(req *types.LogArchiveRestoreReq) (*types.LogArchiveRestoreResp, error) {
	if len(req.Ids) == 0 {
		return nil, apierr.BadRequest("ids 不能为空")
	}
	if l.svcCtx.MinioClient == nil {
		return nil, apierr.New(500, "MinIO客户端未加载,无法恢复(检查t_config的file_*配置)")
	}

	log.RecordDebug(l.ctx, "开始日志归档恢复!")

	list, err := l.svcCtx.LogArchiveRecordModel.FindByIds(l.ctx, req.Ids)
	if err != nil {
		log.RecordError(l.ctx, err)
		return nil, apierr.New(500, "查询归档记录失败")
	}
	if len(list) == 0 {
		return nil, apierr.New(404, "归档记录不存在")
	}

	operator := operatorFromContext(l.ctx)

	maxConcurrent := 5
	// 同时可支持5个协程的信号量通道,满位后阻塞
	sem := make(chan struct{}, maxConcurrent)
	// 为每一个归档日志记录一个结果
	resultChan := make(chan restoreResult, len(list))

	for i := range list {
		logArchiveRecord := &list[i]
		// 占位,满位后阻塞
		sem <- struct{}{}
		go func(r *udeModel.TLogArchiveRecord) {
			defer func() {
				<-sem
				if rec := recover(); rec != nil {
					log.RecordError(l.ctx, fmt.Errorf("[%d]的归档恢复Panic!原因:%v", r.Id, rec))
				}
			}()
			// 恢复执行在后台协程,不携带 HTTP 请求上下文
			resultChan <- runRestore(context.Background(), l.svcCtx, r, operator)
		}(logArchiveRecord)
	}

	// 后台汇总结果日志
	go func() {
		var success, failed int
		for i := 0; i < len(list); i++ {
			r := <-resultChan
			if r.Success {
				success++
			} else {
				failed++
				log.RecordDebug(l.ctx, "[%d]的归档恢复失败:%s", r.RecordId, r.ErrMsg)
			}
		}
		log.RecordDebug(l.ctx, "归档恢复任务全部完成,总数:%d,成功:%d,失败:%d", len(list), success, failed)
	}()

	msg := "归档恢复任务已启动,正在后台处理中,请稍候刷新恢复记录查看进度!"
	log.RecordDebug(l.ctx, "%s", msg)
	return &types.LogArchiveRestoreResp{RestoreMsg: msg}, nil
}

// operatorFromContext resolves the operator name (no auth layer -> admin).
func operatorFromContext(ctx context.Context) string {
	name := audit.GetRequestMeta(ctx).UserName
	if name == "" {
		return "admin"
	}
	return name
}

// runRestore executes one archive restore run and books its record row.
func runRestore(localCtx context.Context, svcCtx *svc.ServiceContext, r *udeModel.TLogArchiveRecord, operator string) restoreResult {
	result := restoreResult{RecordId: r.Id}

	// 1.构造恢复记录
	logRestoreRecord := &udeModel.TLogRestoreRecord{
		ArchiveId:      r.Id,
		RestoreStatus:  audit.RestoreProcessing,
		RestoreMessage: sql.NullString{Valid: false},
		RestoredBy:     operator,
	}
	insertResult, err := svcCtx.LogRestoreRecordModel.Insert(localCtx, logRestoreRecord)
	if err != nil {
		log.RecordError(localCtx, err)
		result.Success, result.ErrMsg = false, "新增恢复记录失败:"+err.Error()
		return result
	}
	recordId, err := insertResult.LastInsertId()
	if err != nil {
		log.RecordError(localCtx, err)
		result.Success, result.ErrMsg = false, "获取恢复记录ID失败:"+err.Error()
		return result
	}

	// 状态更新辅助函数
	update := func(status int64, message sql.NullString) {
		if uErr := svcCtx.LogRestoreRecordModel.UpdateResult(localCtx, recordId, status, message); uErr != nil {
			log.RecordError(localCtx, fmt.Errorf("更新[%d]的恢复记录状态失败!原因:%v", recordId, uErr))
		}
	}
	fail := func(errMsg string) {
		result.Success, result.ErrMsg = false, errMsg
		update(audit.RestoreFail, sql.NullString{String: errMsg, Valid: true})
	}

	// 2.下载归档文件
	filePath, err := svcCtx.MinioClient.DownloadFile(localCtx, r.ArchiveFileKey)
	if err != nil || filePath == "" {
		fail(fmt.Sprintf("下载归档文件失败:%v", err))
		return result
	}

	// 3.流式恢复(每批100条语句一个事务)
	err = audit.RestoreSQLStream(localCtx, filePath, 100, func(ctx context.Context, batch []string) error {
		return svcCtx.DB.TransactCtx(localCtx, func(ctx context.Context, session sqlx.Session) error {
			for _, stmt := range batch {
				if _, err := session.ExecCtx(ctx, stmt); err != nil {
					return err
				}
			}
			return nil
		})
	})

	// 4.清理下载的临时文件
	if rErr := os.Remove(filePath); rErr != nil {
		log.RecordDebug(localCtx, "删除临时文件失败(%s):%v", filePath, rErr)
	}

	// 5.更新恢复记录状态
	if err != nil {
		fail("恢复执行失败:" + err.Error())
		return result
	}

	result.Success, result.ErrMsg = true, ""
	update(audit.RestoreSuccess, sql.NullString{Valid: false})
	return result
}