diff --git a/internal/deployer/deployer.go b/internal/deployer/deployer.go index 1fc1925f..3a892404 100644 --- a/internal/deployer/deployer.go +++ b/internal/deployer/deployer.go @@ -11,6 +11,8 @@ import ( ) type Deployer interface { + SetLogger(*slog.Logger) + Deploy(ctx context.Context) error } @@ -67,6 +69,14 @@ type proxyDeployer struct { deployPrivateKey string } +func (d *proxyDeployer) SetLogger(logger *slog.Logger) { + if logger == nil { + panic("logger is nil") + } + + d.logger = logger +} + func (d *proxyDeployer) Deploy(ctx context.Context) error { _, err := d.deployer.Deploy(ctx, d.deployCertificate, d.deployPrivateKey) return err diff --git a/internal/domain/workflow_log.go b/internal/domain/workflow_log.go new file mode 100644 index 00000000..a33c5480 --- /dev/null +++ b/internal/domain/workflow_log.go @@ -0,0 +1,29 @@ +package domain + +import "strings" + +const CollectionNameWorkflowLog = "workflow_logs" + +type WorkflowLog struct { + Meta + WorkflowId string `json:"workflowId" db:"workflowId"` + RunId string `json:"workflorunIdwId" db:"runId"` + NodeId string `json:"nodeId"` + NodeName string `json:"nodeName"` + Level string `json:"level" db:"level"` + Message string `json:"message" db:"message"` + Data map[string]any `json:"data" db:"data"` +} + +type WorkflowLogs []WorkflowLog + +func (r WorkflowLogs) ErrorString() string { + var builder strings.Builder + for _, log := range r { + if log.Level == "ERROR" { + builder.WriteString(log.Message) + builder.WriteString("\n") + } + } + return strings.TrimSpace(builder.String()) +} diff --git a/internal/domain/workflow_run.go b/internal/domain/workflow_run.go index ff5424c5..407afc73 100644 --- a/internal/domain/workflow_run.go +++ b/internal/domain/workflow_run.go @@ -1,7 +1,6 @@ package domain import ( - "strings" "time" ) @@ -14,7 +13,7 @@ type WorkflowRun struct { Trigger WorkflowTriggerType `json:"trigger" db:"trigger"` StartedAt time.Time `json:"startedAt" db:"startedAt"` EndedAt time.Time `json:"endedAt" db:"endedAt"` - Logs []WorkflowRunLog `json:"logs" db:"logs"` + Detail *WorkflowNode `json:"detail" db:"detail"` Error string `json:"error" db:"error"` } @@ -27,39 +26,3 @@ const ( WorkflowRunStatusTypeFailed WorkflowRunStatusType = "failed" WorkflowRunStatusTypeCanceled WorkflowRunStatusType = "canceled" ) - -type WorkflowRunLog struct { - NodeId string `json:"nodeId"` - NodeName string `json:"nodeName"` - Records []WorkflowRunLogRecord `json:"records"` - Error string `json:"error"` -} - -type WorkflowRunLogRecord struct { - Time string `json:"time"` - Level WorkflowRunLogLevel `json:"level"` - Content string `json:"content"` - Error string `json:"error"` -} - -type WorkflowRunLogLevel string - -const ( - WorkflowRunLogLevelDebug WorkflowRunLogLevel = "DEBUG" - WorkflowRunLogLevelInfo WorkflowRunLogLevel = "INFO" - WorkflowRunLogLevelWarn WorkflowRunLogLevel = "WARN" - WorkflowRunLogLevelError WorkflowRunLogLevel = "ERROR" -) - -type WorkflowRunLogs []WorkflowRunLog - -func (r WorkflowRunLogs) ErrorString() string { - var builder strings.Builder - for _, log := range r { - if log.Error != "" { - builder.WriteString(log.Error) - builder.WriteString("\n") - } - } - return builder.String() -} diff --git a/internal/pkg/logging/handler.go b/internal/pkg/logging/handler.go index 5e932d1f..36174b45 100644 --- a/internal/pkg/logging/handler.go +++ b/internal/pkg/logging/handler.go @@ -24,10 +24,14 @@ type HookHandler struct { attrs []slog.Attr } -func NewHookHandler(options HookHandlerOptions) *HookHandler { +func NewHookHandler(opts *HookHandlerOptions) *HookHandler { + if opts == nil { + opts = &HookHandlerOptions{} + } + h := &HookHandler{ mutex: &sync.Mutex{}, - options: &options, + options: opts, } if h.options.WriteFunc == nil { diff --git a/internal/pkg/utils/maps/maps.go b/internal/pkg/utils/maps/maps.go index 4a4417d0..6fd20bc7 100644 --- a/internal/pkg/utils/maps/maps.go +++ b/internal/pkg/utils/maps/maps.go @@ -184,7 +184,7 @@ func GetValueOrDefaultAsBool(dict map[string]any, key string, defaultValue bool) } // 将字典填充到指定类型的结构体。 -// 与 [json.Unmarshal] 类似，但传入的是一个 [map[string]interface{}] 对象而非 JSON 格式的字符串。 +// 与 [json.Unmarshal] 类似，但传入的是一个 [map[string]any] 对象而非 JSON 格式的字符串。 // // 入参： // - dict: 字典。 diff --git a/internal/repository/workflow_log.go b/internal/repository/workflow_log.go new file mode 100644 index 00000000..95bc2e7d --- /dev/null +++ b/internal/repository/workflow_log.go @@ -0,0 +1,110 @@ +package repository + +import ( + "context" + "database/sql" + "errors" + "fmt" + + "github.com/pocketbase/dbx" + "github.com/pocketbase/pocketbase/core" + "github.com/usual2970/certimate/internal/app" + "github.com/usual2970/certimate/internal/domain" +) + +type WorkflowLogRepository struct{} + +func NewWorkflowLogRepository() *WorkflowLogRepository { + return &WorkflowLogRepository{} +} + +func (r *WorkflowLogRepository) ListByWorkflowRunId(ctx context.Context, workflowRunId string) ([]*domain.WorkflowLog, error) { + records, err := app.GetApp().FindRecordsByFilter( + domain.CollectionNameWorkflowLog, + "runId={:runId}", + "-created", + 0, 0, + dbx.Params{"runId": workflowRunId}, + ) + if err != nil { + return nil, err + } + + workflowLogs := make([]*domain.WorkflowLog, 0) + for _, record := range records { + workflowLog, err := r.castRecordToModel(record) + if err != nil { + return nil, err + } + + workflowLogs = append(workflowLogs, workflowLog) + } + + return workflowLogs, nil +} + +func (r *WorkflowLogRepository) Save(ctx context.Context, workflowLog *domain.WorkflowLog) (*domain.WorkflowLog, error) { + collection, err := app.GetApp().FindCollectionByNameOrId(domain.CollectionNameWorkflowLog) + if err != nil { + return workflowLog, err + } + + var record *core.Record + if workflowLog.Id == "" { + record = core.NewRecord(collection) + } else { + record, err = app.GetApp().FindRecordById(collection, workflowLog.Id) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return workflowLog, err + } + record = core.NewRecord(collection) + } + } + + record.Set("workflowId", workflowLog.WorkflowId) + record.Set("runId", workflowLog.RunId) + record.Set("nodeId", workflowLog.NodeId) + record.Set("nodeName", workflowLog.NodeName) + record.Set("level", workflowLog.Level) + record.Set("message", workflowLog.Message) + record.Set("data", workflowLog.Data) + record.Set("created", workflowLog.CreatedAt) + err = app.GetApp().Save(record) + if err != nil { + return workflowLog, err + } + + workflowLog.Id = record.Id + workflowLog.CreatedAt = record.GetDateTime("created").Time() + workflowLog.UpdatedAt = record.GetDateTime("updated").Time() + + return workflowLog, nil +} + +func (r *WorkflowLogRepository) castRecordToModel(record *core.Record) (*domain.WorkflowLog, error) { + if record == nil { + return nil, fmt.Errorf("record is nil") + } + + logdata := make(map[string]any) + if err := record.UnmarshalJSONField("data", &logdata); err != nil { + return nil, err + } + + workflowLog := &domain.WorkflowLog{ + Meta: domain.Meta{ + Id: record.Id, + CreatedAt: record.GetDateTime("created").Time(), + UpdatedAt: record.GetDateTime("updated").Time(), + }, + WorkflowId: record.GetString("workflowId"), + RunId: record.GetString("runId"), + NodeId: record.GetString("nodeId"), + NodeName: record.GetString("nodeName"), + Level: record.GetString("level"), + Message: record.GetString("message"), + Data: logdata, + } + return workflowLog, nil +} diff --git a/internal/repository/workflow_run.go b/internal/repository/workflow_run.go index b1a5234b..aef61ac3 100644 --- a/internal/repository/workflow_run.go +++ b/internal/repository/workflow_run.go @@ -54,7 +54,7 @@ func (r *WorkflowRunRepository) Save(ctx context.Context, workflowRun *domain.Wo record.Set("status", string(workflowRun.Status)) record.Set("startedAt", workflowRun.StartedAt) record.Set("endedAt", workflowRun.EndedAt) - record.Set("logs", workflowRun.Logs) + record.Set("detail", workflowRun.Detail) record.Set("error", workflowRun.Error) err = txApp.Save(record) if err != nil { @@ -101,8 +101,8 @@ func (r *WorkflowRunRepository) castRecordToModel(record *core.Record) (*domain. return nil, fmt.Errorf("record is nil") } - logs := make([]domain.WorkflowRunLog, 0) - if err := record.UnmarshalJSONField("logs", &logs); err != nil { + detail := &domain.WorkflowNode{} + if err := record.UnmarshalJSONField("detail", &detail); err != nil { return nil, err } @@ -117,7 +117,7 @@ func (r *WorkflowRunRepository) castRecordToModel(record *core.Record) (*domain. Trigger: domain.WorkflowTriggerType(record.GetString("trigger")), StartedAt: record.GetDateTime("startedAt").Time(), EndedAt: record.GetDateTime("endedAt").Time(), - Logs: logs, + Detail: detail, Error: record.GetString("error"), } return workflowRun, nil diff --git a/internal/workflow/dispatcher/dispatcher.go b/internal/workflow/dispatcher/dispatcher.go index 53081b6b..8f1adf18 100644 --- a/internal/workflow/dispatcher/dispatcher.go +++ b/internal/workflow/dispatcher/dispatcher.go @@ -51,9 +51,10 @@ type WorkflowDispatcher struct { workflowRepo workflowRepository workflowRunRepo workflowRunRepository + workflowLogRepo workflowLogRepository } -func newWorkflowDispatcher(workflowRepo workflowRepository, workflowRunRepo workflowRunRepository) *WorkflowDispatcher { +func newWorkflowDispatcher(workflowRepo workflowRepository, workflowRunRepo workflowRunRepository, workflowLogRepo workflowLogRepository) *WorkflowDispatcher { dispatcher := &WorkflowDispatcher{ semaphore: make(chan struct{}, maxWorkers), @@ -69,6 +70,7 @@ func newWorkflowDispatcher(workflowRepo workflowRepository, workflowRunRepo work workflowRepo: workflowRepo, workflowRunRepo: workflowRunRepo, + workflowLogRepo: workflowLogRepo, } go func() { @@ -86,139 +88,139 @@ func newWorkflowDispatcher(workflowRepo workflowRepository, workflowRunRepo work return dispatcher } -func (w *WorkflowDispatcher) Dispatch(data *WorkflowWorkerData) { +func (d *WorkflowDispatcher) Dispatch(data *WorkflowWorkerData) { if data == nil { panic("worker data is nil") } - w.enqueueWorker(data) + d.enqueueWorker(data) select { - case w.chWork <- data: + case d.chWork <- data: default: } } -func (w *WorkflowDispatcher) Cancel(runId string) { +func (d *WorkflowDispatcher) Cancel(runId string) { hasWorker := false // 取消正在执行的 WorkflowRun - w.workerMutex.Lock() - if workflowId, ok := w.workerIdMap[runId]; ok { - if worker, ok := w.workers[workflowId]; ok { + d.workerMutex.Lock() + if workflowId, ok := d.workerIdMap[runId]; ok { + if worker, ok := d.workers[workflowId]; ok { hasWorker = true worker.Cancel() - delete(w.workers, workflowId) - delete(w.workerIdMap, runId) + delete(d.workers, workflowId) + delete(d.workerIdMap, runId) } } - w.workerMutex.Unlock() + d.workerMutex.Unlock() // 移除排队中的 WorkflowRun - w.queueMutex.Lock() - w.queue = slices.Filter(w.queue, func(d *WorkflowWorkerData) bool { + d.queueMutex.Lock() + d.queue = slices.Filter(d.queue, func(d *WorkflowWorkerData) bool { return d.RunId != runId }) - w.queueMutex.Unlock() + d.queueMutex.Unlock() // 已挂起，查询 WorkflowRun 并更新其状态为 Canceled if !hasWorker { - if run, err := w.workflowRunRepo.GetById(context.Background(), runId); err == nil { + if run, err := d.workflowRunRepo.GetById(context.Background(), runId); err == nil { if run.Status == domain.WorkflowRunStatusTypePending || run.Status == domain.WorkflowRunStatusTypeRunning { run.Status = domain.WorkflowRunStatusTypeCanceled - w.workflowRunRepo.Save(context.Background(), run) + d.workflowRunRepo.Save(context.Background(), run) } } } } -func (w *WorkflowDispatcher) Shutdown() { +func (d *WorkflowDispatcher) Shutdown() { // 清空排队中的 WorkflowRun - w.queueMutex.Lock() - w.queue = make([]*WorkflowWorkerData, 0) - w.queueMutex.Unlock() + d.queueMutex.Lock() + d.queue = make([]*WorkflowWorkerData, 0) + d.queueMutex.Unlock() // 等待所有正在执行的 WorkflowRun 完成 - w.workerMutex.Lock() - for _, worker := range w.workers { + d.workerMutex.Lock() + for _, worker := range d.workers { worker.Cancel() - delete(w.workers, worker.Data.WorkflowId) - delete(w.workerIdMap, worker.Data.RunId) + delete(d.workers, worker.Data.WorkflowId) + delete(d.workerIdMap, worker.Data.RunId) } - w.workerMutex.Unlock() - w.wg.Wait() + d.workerMutex.Unlock() + d.wg.Wait() } -func (w *WorkflowDispatcher) enqueueWorker(data *WorkflowWorkerData) { - w.queueMutex.Lock() - defer w.queueMutex.Unlock() - w.queue = append(w.queue, data) +func (d *WorkflowDispatcher) enqueueWorker(data *WorkflowWorkerData) { + d.queueMutex.Lock() + defer d.queueMutex.Unlock() + d.queue = append(d.queue, data) } -func (w *WorkflowDispatcher) dequeueWorker() { +func (d *WorkflowDispatcher) dequeueWorker() { for { select { - case w.semaphore <- struct{}{}: + case d.semaphore <- struct{}{}: default: // 达到最大并发数 return } - w.queueMutex.Lock() - if len(w.queue) == 0 { - w.queueMutex.Unlock() - <-w.semaphore + d.queueMutex.Lock() + if len(d.queue) == 0 { + d.queueMutex.Unlock() + <-d.semaphore return } - data := w.queue[0] - w.queue = w.queue[1:] - w.queueMutex.Unlock() + data := d.queue[0] + d.queue = d.queue[1:] + d.queueMutex.Unlock() // 检查是否有相同 WorkflowId 的 WorkflowRun 正在执行 // 如果有，则重新排队，以保证同一个工作流同一时间内只有一个正在执行 // 即不同 WorkflowId 的任务并行化，相同 WorkflowId 的任务串行化 - w.workerMutex.Lock() - if _, exists := w.workers[data.WorkflowId]; exists { - w.queueMutex.Lock() - w.queue = append(w.queue, data) - w.queueMutex.Unlock() - w.workerMutex.Unlock() + d.workerMutex.Lock() + if _, exists := d.workers[data.WorkflowId]; exists { + d.queueMutex.Lock() + d.queue = append(d.queue, data) + d.queueMutex.Unlock() + d.workerMutex.Unlock() - <-w.semaphore + <-d.semaphore continue } ctx, cancel := context.WithCancel(context.Background()) - w.workers[data.WorkflowId] = &workflowWorker{data, cancel} - w.workerIdMap[data.RunId] = data.WorkflowId - w.workerMutex.Unlock() + d.workers[data.WorkflowId] = &workflowWorker{data, cancel} + d.workerIdMap[data.RunId] = data.WorkflowId + d.workerMutex.Unlock() - w.wg.Add(1) - go w.work(ctx, data) + d.wg.Add(1) + go d.work(ctx, data) } } -func (w *WorkflowDispatcher) work(ctx context.Context, data *WorkflowWorkerData) { +func (d *WorkflowDispatcher) work(ctx context.Context, data *WorkflowWorkerData) { defer func() { - <-w.semaphore - w.workerMutex.Lock() - delete(w.workers, data.WorkflowId) - delete(w.workerIdMap, data.RunId) - w.workerMutex.Unlock() + <-d.semaphore + d.workerMutex.Lock() + delete(d.workers, data.WorkflowId) + delete(d.workerIdMap, data.RunId) + d.workerMutex.Unlock() - w.wg.Done() + d.wg.Done() // 尝试取出排队中的其他 WorkflowRun 继续执行 select { - case w.chCandi <- struct{}{}: + case d.chCandi <- struct{}{}: default: } }() // 查询 WorkflowRun - run, err := w.workflowRunRepo.GetById(ctx, data.RunId) + run, err := d.workflowRunRepo.GetById(ctx, data.RunId) if err != nil { if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) { app.GetLogger().Error(fmt.Sprintf("failed to get workflow run #%s", data.RunId), "err", err) @@ -228,13 +230,13 @@ func (w *WorkflowDispatcher) work(ctx context.Context, data *WorkflowWorkerData) return } else if ctx.Err() != nil { run.Status = domain.WorkflowRunStatusTypeCanceled - w.workflowRunRepo.Save(ctx, run) + d.workflowRunRepo.Save(ctx, run) return } // 更新 WorkflowRun 状态为 Running run.Status = domain.WorkflowRunStatusTypeRunning - if _, err := w.workflowRunRepo.Save(ctx, run); err != nil { + if _, err := d.workflowRunRepo.Save(ctx, run); err != nil { if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) { panic(err) } @@ -242,19 +244,17 @@ func (w *WorkflowDispatcher) work(ctx context.Context, data *WorkflowWorkerData) } // 执行工作流 - invoker := newWorkflowInvokerWithData(w.workflowRunRepo, data) + invoker := newWorkflowInvokerWithData(d.workflowLogRepo, data) if runErr := invoker.Invoke(ctx); runErr != nil { if errors.Is(runErr, context.Canceled) { run.Status = domain.WorkflowRunStatusTypeCanceled - run.Logs = invoker.GetLogs() } else { run.Status = domain.WorkflowRunStatusTypeFailed run.EndedAt = time.Now() - run.Logs = invoker.GetLogs() run.Error = runErr.Error() } - if _, err := w.workflowRunRepo.Save(ctx, run); err != nil { + if _, err := d.workflowRunRepo.Save(ctx, run); err != nil { if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) { panic(err) } @@ -265,14 +265,13 @@ func (w *WorkflowDispatcher) work(ctx context.Context, data *WorkflowWorkerData) // 更新 WorkflowRun 状态为 Succeeded/Failed run.EndedAt = time.Now() - run.Logs = invoker.GetLogs() - run.Error = domain.WorkflowRunLogs(invoker.GetLogs()).ErrorString() + run.Error = invoker.GetLogs().ErrorString() if run.Error == "" { run.Status = domain.WorkflowRunStatusTypeSucceeded } else { run.Status = domain.WorkflowRunStatusTypeFailed } - if _, err := w.workflowRunRepo.Save(ctx, run); err != nil { + if _, err := d.workflowRunRepo.Save(ctx, run); err != nil { if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) { panic(err) } diff --git a/internal/workflow/dispatcher/invoker.go b/internal/workflow/dispatcher/invoker.go index d35cca3e..23d70f01 100644 --- a/internal/workflow/dispatcher/invoker.go +++ b/internal/workflow/dispatcher/invoker.go @@ -3,8 +3,10 @@ package dispatcher import ( "context" "errors" + "log/slog" "github.com/usual2970/certimate/internal/domain" + "github.com/usual2970/certimate/internal/pkg/logging" nodes "github.com/usual2970/certimate/internal/workflow/node-processor" ) @@ -12,24 +14,23 @@ type workflowInvoker struct { workflowId string workflowContent *domain.WorkflowNode runId string - runLogs []domain.WorkflowRunLog + logs []domain.WorkflowLog - workflowRunRepo workflowRunRepository + workflowLogRepo workflowLogRepository } -func newWorkflowInvokerWithData(workflowRunRepo workflowRunRepository, data *WorkflowWorkerData) *workflowInvoker { +func newWorkflowInvokerWithData(workflowLogRepo workflowLogRepository, data *WorkflowWorkerData) *workflowInvoker { if data == nil { panic("worker data is nil") } - // TODO: 待优化，日志与执行解耦 return &workflowInvoker{ workflowId: data.WorkflowId, workflowContent: data.WorkflowContent, runId: data.RunId, - runLogs: make([]domain.WorkflowRunLog, 0), + logs: make([]domain.WorkflowLog, 0), - workflowRunRepo: workflowRunRepo, + workflowLogRepo: workflowLogRepo, } } @@ -39,8 +40,8 @@ func (w *workflowInvoker) Invoke(ctx context.Context) error { return w.processNode(ctx, w.workflowContent) } -func (w *workflowInvoker) GetLogs() []domain.WorkflowRunLog { - return w.runLogs +func (w *workflowInvoker) GetLogs() domain.WorkflowLogs { + return w.logs } func (w *workflowInvoker) processNode(ctx context.Context, node *domain.WorkflowNode) error { @@ -68,21 +69,33 @@ func (w *workflowInvoker) processNode(ctx context.Context, node *domain.Workflow if current.Type != domain.WorkflowNodeTypeBranch && current.Type != domain.WorkflowNodeTypeExecuteResultBranch { processor, procErr = nodes.GetProcessor(current) if procErr != nil { - break + panic(procErr) } + processor.SetLogger(slog.New(logging.NewHookHandler(&logging.HookHandlerOptions{ + Level: slog.LevelDebug, + WriteFunc: func(ctx context.Context, record *logging.Record) error { + log := domain.WorkflowLog{} + log.WorkflowId = w.workflowId + log.RunId = w.runId + log.NodeId = current.Id + log.NodeName = current.Name + log.Level = record.Level.String() + log.Message = record.Message + log.Data = record.Data + log.CreatedAt = record.Time + if _, err := w.workflowLogRepo.Save(ctx, &log); err != nil { + return err + } + + w.logs = append(w.logs, log) + return nil + }, + }))) + procErr = processor.Process(ctx) - log := processor.GetLog(ctx) - if log != nil { - w.runLogs = append(w.runLogs, *log) - - // TODO: 待优化，把 /pkg/core/* 包下的输出写入到 DEBUG 级别的日志中 - if run, err := w.workflowRunRepo.GetById(ctx, w.runId); err == nil { - run.Logs = w.runLogs - w.workflowRunRepo.Save(ctx, run) - } - } if procErr != nil { + processor.GetLogger().Error(procErr.Error()) break } } diff --git a/internal/workflow/dispatcher/singleton.go b/internal/workflow/dispatcher/singleton.go index b5834c48..20fdde97 100644 --- a/internal/workflow/dispatcher/singleton.go +++ b/internal/workflow/dispatcher/singleton.go @@ -5,6 +5,7 @@ import ( "sync" "github.com/usual2970/certimate/internal/domain" + "github.com/usual2970/certimate/internal/repository" ) type workflowRepository interface { @@ -17,15 +18,18 @@ type workflowRunRepository interface { Save(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error) } +type workflowLogRepository interface { + Save(ctx context.Context, workflowLog *domain.WorkflowLog) (*domain.WorkflowLog, error) +} + var ( instance *WorkflowDispatcher intanceOnce sync.Once ) -func GetSingletonDispatcher(workflowRepo workflowRepository, workflowRunRepo workflowRunRepository) *WorkflowDispatcher { - // TODO: 待优化构造过程 +func GetSingletonDispatcher() *WorkflowDispatcher { intanceOnce.Do(func() { - instance = newWorkflowDispatcher(workflowRepo, workflowRunRepo) + instance = newWorkflowDispatcher(repository.NewWorkflowRepository(), repository.NewWorkflowRunRepository(), repository.NewWorkflowLogRepository()) }) return instance diff --git a/internal/workflow/node-processor/apply_node.go b/internal/workflow/node-processor/apply_node.go index 26ff6b9e..2fc6c223 100644 --- a/internal/workflow/node-processor/apply_node.go +++ b/internal/workflow/node-processor/apply_node.go @@ -15,7 +15,7 @@ import ( type applyNode struct { node *domain.WorkflowNode - *nodeLogger + *nodeProcessor certRepo certificateRepository outputRepo workflowOutputRepository @@ -23,8 +23,8 @@ type applyNode struct { func NewApplyNode(node *domain.WorkflowNode) *applyNode { return &applyNode{ - node: node, - nodeLogger: newNodeLogger(node), + node: node, + nodeProcessor: newNodeProcessor(node), certRepo: repository.NewCertificateRepository(), outputRepo: repository.NewWorkflowOutputRepository(), @@ -32,40 +32,40 @@ func NewApplyNode(node *domain.WorkflowNode) *applyNode { } func (n *applyNode) Process(ctx context.Context) error { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "进入申请证书节点") + n.logger.Info("ready to apply ...") // 查询上次执行结果 lastOutput, err := n.outputRepo.GetByNodeId(ctx, n.node.Id) if err != nil && !domain.IsRecordNotFoundError(err) { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "查询申请记录失败", err.Error()) return err } // 检测是否可以跳过本次执行 if skippable, skipReason := n.checkCanSkip(ctx, lastOutput); skippable { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, skipReason) + n.logger.Warn(fmt.Sprintf("skip this application, because %s", skipReason)) return nil + } else if skipReason != "" { + n.logger.Info(fmt.Sprintf("continue to apply, because %s", skipReason)) } // 初始化申请器 applicant, err := applicant.NewWithApplyNode(n.node) if err != nil { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "获取申请对象失败", err.Error()) + n.logger.Warn("failed to create applicant provider") return err } // 申请证书 applyResult, err := applicant.Apply() if err != nil { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "申请失败", err.Error()) + n.logger.Warn("failed to apply") return err } - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "申请成功") // 解析证书并生成实体 certX509, err := certs.ParseCertificateFromPEM(applyResult.CertificateFullChain) if err != nil { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "解析证书失败", err.Error()) + n.logger.Warn("failed to parse certificate, may be the CA responded error") return err } certificate := &domain.Certificate{ @@ -89,10 +89,11 @@ func (n *applyNode) Process(ctx context.Context) error { Outputs: n.node.Outputs, } if _, err := n.outputRepo.SaveWithCertificate(ctx, output, certificate); err != nil { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "保存申请记录失败", err.Error()) + n.logger.Warn("failed to save node output") return err } - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "保存申请记录成功") + + n.logger.Info("apply completed") return nil } @@ -103,19 +104,19 @@ func (n *applyNode) checkCanSkip(ctx context.Context, lastOutput *domain.Workflo currentNodeConfig := n.node.GetConfigForApply() lastNodeConfig := lastOutput.Node.GetConfigForApply() if currentNodeConfig.Domains != lastNodeConfig.Domains { - return false, "配置项变化：域名" + return false, "the configuration item 'Domains' changed" } if currentNodeConfig.ContactEmail != lastNodeConfig.ContactEmail { - return false, "配置项变化：联系邮箱" + return false, "the configuration item 'ContactEmail' changed" } if currentNodeConfig.ProviderAccessId != lastNodeConfig.ProviderAccessId { - return false, "配置项变化：DNS 提供商授权" + return false, "the configuration item 'ProviderAccessId' changed" } if !maps.Equal(currentNodeConfig.ProviderConfig, lastNodeConfig.ProviderConfig) { - return false, "配置项变化：DNS 提供商参数" + return false, "the configuration item 'ProviderConfig' changed" } if currentNodeConfig.KeyAlgorithm != lastNodeConfig.KeyAlgorithm { - return false, "配置项变化：数字签名算法" + return false, "the configuration item 'KeyAlgorithm' changed" } lastCertificate, _ := n.certRepo.GetByWorkflowNodeId(ctx, n.node.Id) @@ -123,7 +124,7 @@ func (n *applyNode) checkCanSkip(ctx context.Context, lastOutput *domain.Workflo renewalInterval := time.Duration(currentNodeConfig.SkipBeforeExpiryDays) * time.Hour * 24 expirationTime := time.Until(lastCertificate.ExpireAt) if expirationTime > renewalInterval { - return true, fmt.Sprintf("已申请过证书，且证书尚未临近过期（尚余 %d 天过期，不足 %d 天时续期），跳过此次申请", int(expirationTime.Hours()/24), currentNodeConfig.SkipBeforeExpiryDays) + return true, fmt.Sprintf("the certificate has already been issued (expires in %dD, next renewal in %dD)", int(expirationTime.Hours()/24), currentNodeConfig.SkipBeforeExpiryDays) } } } diff --git a/internal/workflow/node-processor/condition_node.go b/internal/workflow/node-processor/condition_node.go index 499a5004..2bac55fa 100644 --- a/internal/workflow/node-processor/condition_node.go +++ b/internal/workflow/node-processor/condition_node.go @@ -8,13 +8,13 @@ import ( type conditionNode struct { node *domain.WorkflowNode - *nodeLogger + *nodeProcessor } func NewConditionNode(node *domain.WorkflowNode) *conditionNode { return &conditionNode{ - node: node, - nodeLogger: newNodeLogger(node), + node: node, + nodeProcessor: newNodeProcessor(node), } } diff --git a/internal/workflow/node-processor/deploy_node.go b/internal/workflow/node-processor/deploy_node.go index 492f5385..556ca891 100644 --- a/internal/workflow/node-processor/deploy_node.go +++ b/internal/workflow/node-processor/deploy_node.go @@ -3,6 +3,7 @@ package nodeprocessor import ( "context" "fmt" + "log/slog" "strings" "github.com/usual2970/certimate/internal/deployer" @@ -13,7 +14,7 @@ import ( type deployNode struct { node *domain.WorkflowNode - *nodeLogger + *nodeProcessor certRepo certificateRepository outputRepo workflowOutputRepository @@ -21,8 +22,8 @@ type deployNode struct { func NewDeployNode(node *domain.WorkflowNode) *deployNode { return &deployNode{ - node: node, - nodeLogger: newNodeLogger(node), + node: node, + nodeProcessor: newNodeProcessor(node), certRepo: repository.NewCertificateRepository(), outputRepo: repository.NewWorkflowOutputRepository(), @@ -30,12 +31,11 @@ func NewDeployNode(node *domain.WorkflowNode) *deployNode { } func (n *deployNode) Process(ctx context.Context) error { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "开始执行") + n.logger.Info("ready to deploy ...") // 查询上次执行结果 lastOutput, err := n.outputRepo.GetByNodeId(ctx, n.node.Id) if err != nil && !domain.IsRecordNotFoundError(err) { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "查询部署记录失败", err.Error()) return err } @@ -43,20 +43,22 @@ func (n *deployNode) Process(ctx context.Context) error { previousNodeOutputCertificateSource := n.node.GetConfigForDeploy().Certificate previousNodeOutputCertificateSourceSlice := strings.Split(previousNodeOutputCertificateSource, "#") if len(previousNodeOutputCertificateSourceSlice) != 2 { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "证书来源配置错误", previousNodeOutputCertificateSource) - return fmt.Errorf("证书来源配置错误: %s", previousNodeOutputCertificateSource) + n.logger.Warn("invalid certificate source", slog.String("certificate.source", previousNodeOutputCertificateSource)) + return fmt.Errorf("invalid certificate source: %s", previousNodeOutputCertificateSource) } certificate, err := n.certRepo.GetByWorkflowNodeId(ctx, previousNodeOutputCertificateSourceSlice[0]) if err != nil { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "获取证书失败", err.Error()) + n.logger.Warn("invalid certificate source", slog.String("certificate.source", previousNodeOutputCertificateSource)) return err } // 检测是否可以跳过本次执行 if lastOutput != nil && certificate.CreatedAt.Before(lastOutput.UpdatedAt) { if skippable, skipReason := n.checkCanSkip(ctx, lastOutput); skippable { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, skipReason) + n.logger.Warn(fmt.Sprintf("skip this deployment, because %s", skipReason)) return nil + } else if skipReason != "" { + n.logger.Info(fmt.Sprintf("continue to deploy, because %s", skipReason)) } } @@ -66,16 +68,16 @@ func (n *deployNode) Process(ctx context.Context) error { PrivateKey string }{Certificate: certificate.Certificate, PrivateKey: certificate.PrivateKey}) if err != nil { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "获取部署对象失败", err.Error()) + n.logger.Warn("failed to create deployer provider") return err } // 部署证书 + deployer.SetLogger(n.logger) if err := deployer.Deploy(ctx); err != nil { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "部署失败", err.Error()) + n.logger.Warn("failed to deploy") return err } - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "部署成功") // 保存执行结果 output := &domain.WorkflowOutput{ @@ -86,10 +88,11 @@ func (n *deployNode) Process(ctx context.Context) error { Succeeded: true, } if _, err := n.outputRepo.Save(ctx, output); err != nil { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "保存部署记录失败", err.Error()) + n.logger.Warn("failed to save node output") return err } - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "保存部署记录成功") + + n.logger.Info("apply completed") return nil } @@ -100,14 +103,14 @@ func (n *deployNode) checkCanSkip(ctx context.Context, lastOutput *domain.Workfl currentNodeConfig := n.node.GetConfigForDeploy() lastNodeConfig := lastOutput.Node.GetConfigForDeploy() if currentNodeConfig.ProviderAccessId != lastNodeConfig.ProviderAccessId { - return false, "配置项变化：主机提供商授权" + return false, "the configuration item 'ProviderAccessId' changed" } if !maps.Equal(currentNodeConfig.ProviderConfig, lastNodeConfig.ProviderConfig) { - return false, "配置项变化：主机提供商参数" + return false, "the configuration item 'ProviderConfig' changed" } if currentNodeConfig.SkipOnLastSucceeded { - return true, "已部署过证书，跳过此次部署" + return true, "the certificate has already been deployed" } } diff --git a/internal/workflow/node-processor/execute_failure_node.go b/internal/workflow/node-processor/execute_failure_node.go index 2516edb4..59f6a5bd 100644 --- a/internal/workflow/node-processor/execute_failure_node.go +++ b/internal/workflow/node-processor/execute_failure_node.go @@ -8,19 +8,19 @@ import ( type executeFailureNode struct { node *domain.WorkflowNode - *nodeLogger + *nodeProcessor } func NewExecuteFailureNode(node *domain.WorkflowNode) *executeFailureNode { return &executeFailureNode{ - node: node, - nodeLogger: newNodeLogger(node), + node: node, + nodeProcessor: newNodeProcessor(node), } } func (n *executeFailureNode) Process(ctx context.Context) error { // 此类型节点不需要执行任何操作，直接返回 - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "进入执行失败分支") + n.logger.Info("the previous node execution was failed") return nil } diff --git a/internal/workflow/node-processor/execute_success_node.go b/internal/workflow/node-processor/execute_success_node.go index a7833a53..e5b65860 100644 --- a/internal/workflow/node-processor/execute_success_node.go +++ b/internal/workflow/node-processor/execute_success_node.go @@ -8,19 +8,19 @@ import ( type executeSuccessNode struct { node *domain.WorkflowNode - *nodeLogger + *nodeProcessor } func NewExecuteSuccessNode(node *domain.WorkflowNode) *executeSuccessNode { return &executeSuccessNode{ - node: node, - nodeLogger: newNodeLogger(node), + node: node, + nodeProcessor: newNodeProcessor(node), } } func (n *executeSuccessNode) Process(ctx context.Context) error { // 此类型节点不需要执行任何操作，直接返回 - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "进入执行成功分支") + n.logger.Info("the previous node execution was succeeded") return nil } diff --git a/internal/workflow/node-processor/notify_node.go b/internal/workflow/node-processor/notify_node.go index e4c3da2d..1c2b49d8 100644 --- a/internal/workflow/node-processor/notify_node.go +++ b/internal/workflow/node-processor/notify_node.go @@ -2,6 +2,7 @@ package nodeprocessor import ( "context" + "log/slog" "github.com/usual2970/certimate/internal/domain" "github.com/usual2970/certimate/internal/notify" @@ -10,45 +11,44 @@ import ( type notifyNode struct { node *domain.WorkflowNode - *nodeLogger + *nodeProcessor settingsRepo settingsRepository } func NewNotifyNode(node *domain.WorkflowNode) *notifyNode { return ¬ifyNode{ - node: node, - nodeLogger: newNodeLogger(node), + node: node, + nodeProcessor: newNodeProcessor(node), settingsRepo: repository.NewSettingsRepository(), } } func (n *notifyNode) Process(ctx context.Context) error { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "进入推送通知节点") + n.logger.Info("ready to notify ...") nodeConfig := n.node.GetConfigForNotify() // 获取通知配置 settings, err := n.settingsRepo.GetByName(ctx, "notifyChannels") if err != nil { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "获取通知配置失败", err.Error()) return err } // 获取通知渠道 channelConfig, err := settings.GetNotifyChannelConfig(nodeConfig.Channel) if err != nil { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "获取通知渠道配置失败", err.Error()) return err } // 发送通知 if err := notify.SendToChannel(nodeConfig.Subject, nodeConfig.Message, nodeConfig.Channel, channelConfig); err != nil { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "发送通知失败", err.Error()) + n.logger.Warn("failed to notify", slog.String("channel", nodeConfig.Channel)) return err } - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "发送通知成功") + + n.logger.Info("notify completed") return nil } diff --git a/internal/workflow/node-processor/processor.go b/internal/workflow/node-processor/processor.go index 08712280..4523b13a 100644 --- a/internal/workflow/node-processor/processor.go +++ b/internal/workflow/node-processor/processor.go @@ -2,21 +2,34 @@ package nodeprocessor import ( "context" - "errors" - "time" + "fmt" + "io" + "log/slog" "github.com/usual2970/certimate/internal/domain" ) type NodeProcessor interface { - Process(ctx context.Context) error + GetLogger() *slog.Logger + SetLogger(*slog.Logger) - GetLog(ctx context.Context) *domain.WorkflowRunLog - AppendLogRecord(ctx context.Context, level domain.WorkflowRunLogLevel, content string, err ...string) + Process(ctx context.Context) error } -type nodeLogger struct { - log *domain.WorkflowRunLog +type nodeProcessor struct { + logger *slog.Logger +} + +func (n *nodeProcessor) GetLogger() *slog.Logger { + return n.logger +} + +func (n *nodeProcessor) SetLogger(logger *slog.Logger) { + if logger == nil { + panic("logger is nil") + } + + n.logger = logger } type certificateRepository interface { @@ -33,34 +46,12 @@ type settingsRepository interface { GetByName(ctx context.Context, name string) (*domain.Settings, error) } -func newNodeLogger(node *domain.WorkflowNode) *nodeLogger { - return &nodeLogger{ - log: &domain.WorkflowRunLog{ - NodeId: node.Id, - NodeName: node.Name, - Records: make([]domain.WorkflowRunLogRecord, 0), - }, +func newNodeProcessor(node *domain.WorkflowNode) *nodeProcessor { + return &nodeProcessor{ + logger: slog.New(slog.NewTextHandler(io.Discard, nil)), } } -func (l *nodeLogger) GetLog(ctx context.Context) *domain.WorkflowRunLog { - return l.log -} - -func (l *nodeLogger) AppendLogRecord(ctx context.Context, level domain.WorkflowRunLogLevel, content string, err ...string) { - record := domain.WorkflowRunLogRecord{ - Time: time.Now().UTC().Format(time.RFC3339), - Level: level, - Content: content, - } - if len(err) > 0 { - record.Error = err[0] - l.log.Error = err[0] - } - - l.log.Records = append(l.log.Records, record) -} - func GetProcessor(node *domain.WorkflowNode) (NodeProcessor, error) { switch node.Type { case domain.WorkflowNodeTypeStart: @@ -80,7 +71,8 @@ func GetProcessor(node *domain.WorkflowNode) (NodeProcessor, error) { case domain.WorkflowNodeTypeExecuteFailure: return NewExecuteFailureNode(node), nil } - return nil, errors.New("not implemented") + + return nil, fmt.Errorf("supported node type: %s", string(node.Type)) } func getContextWorkflowId(ctx context.Context) string { diff --git a/internal/workflow/node-processor/start_node.go b/internal/workflow/node-processor/start_node.go index 7d04685a..5bbc1c09 100644 --- a/internal/workflow/node-processor/start_node.go +++ b/internal/workflow/node-processor/start_node.go @@ -8,19 +8,19 @@ import ( type startNode struct { node *domain.WorkflowNode - *nodeLogger + *nodeProcessor } func NewStartNode(node *domain.WorkflowNode) *startNode { return &startNode{ - node: node, - nodeLogger: newNodeLogger(node), + node: node, + nodeProcessor: newNodeProcessor(node), } } func (n *startNode) Process(ctx context.Context) error { // 此类型节点不需要执行任何操作，直接返回 - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "进入开始节点") + n.logger.Info("ready to start ...") return nil } diff --git a/internal/workflow/node-processor/upload_node.go b/internal/workflow/node-processor/upload_node.go index 99108481..a3640c2d 100644 --- a/internal/workflow/node-processor/upload_node.go +++ b/internal/workflow/node-processor/upload_node.go @@ -2,18 +2,16 @@ package nodeprocessor import ( "context" - "errors" + "fmt" "strings" - "time" "github.com/usual2970/certimate/internal/domain" - "github.com/usual2970/certimate/internal/pkg/utils/certs" "github.com/usual2970/certimate/internal/repository" ) type uploadNode struct { node *domain.WorkflowNode - *nodeLogger + *nodeProcessor certRepo certificateRepository outputRepo workflowOutputRepository @@ -21,8 +19,8 @@ type uploadNode struct { func NewUploadNode(node *domain.WorkflowNode) *uploadNode { return &uploadNode{ - node: node, - nodeLogger: newNodeLogger(node), + node: node, + nodeProcessor: newNodeProcessor(node), certRepo: repository.NewCertificateRepository(), outputRepo: repository.NewWorkflowOutputRepository(), @@ -30,33 +28,22 @@ func NewUploadNode(node *domain.WorkflowNode) *uploadNode { } func (n *uploadNode) Process(ctx context.Context) error { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "进入上传证书节点") + n.logger.Info("ready to upload ...") nodeConfig := n.node.GetConfigForUpload() // 查询上次执行结果 lastOutput, err := n.outputRepo.GetByNodeId(ctx, n.node.Id) if err != nil && !domain.IsRecordNotFoundError(err) { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "查询申请记录失败", err.Error()) return err } // 检测是否可以跳过本次执行 if skippable, skipReason := n.checkCanSkip(ctx, lastOutput); skippable { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, skipReason) + n.logger.Warn(fmt.Sprintf("skip this upload, because %s", skipReason)) return nil - } - - // 检查证书是否过期 - // 如果证书过期，则直接返回错误 - certX509, err := certs.ParseCertificateFromPEM(nodeConfig.Certificate) - if err != nil { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "解析证书失败") - return err - } - if time.Now().After(certX509.NotAfter) { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelWarn, "证书已过期") - return errors.New("certificate is expired") + } else if skipReason != "" { + n.logger.Info(fmt.Sprintf("continue to upload, because %s", skipReason)) } // 生成证书实体 @@ -75,10 +62,11 @@ func (n *uploadNode) Process(ctx context.Context) error { Outputs: n.node.Outputs, } if _, err := n.outputRepo.SaveWithCertificate(ctx, output, certificate); err != nil { - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "保存上传记录失败", err.Error()) + n.logger.Warn("failed to save node output") return err } - n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "保存上传记录成功") + + n.logger.Info("upload completed") return nil } @@ -89,15 +77,15 @@ func (n *uploadNode) checkCanSkip(ctx context.Context, lastOutput *domain.Workfl currentNodeConfig := n.node.GetConfigForUpload() lastNodeConfig := lastOutput.Node.GetConfigForUpload() if strings.TrimSpace(currentNodeConfig.Certificate) != strings.TrimSpace(lastNodeConfig.Certificate) { - return false, "配置项变化：证书" + return false, "the configuration item 'Certificate' changed" } if strings.TrimSpace(currentNodeConfig.PrivateKey) != strings.TrimSpace(lastNodeConfig.PrivateKey) { - return false, "配置项变化：私钥" + return false, "the configuration item 'PrivateKey' changed" } lastCertificate, _ := n.certRepo.GetByWorkflowNodeId(ctx, n.node.Id) if lastCertificate != nil { - return true, "已上传过证书" + return true, "the certificate has already been uploaded" } } diff --git a/internal/workflow/service.go b/internal/workflow/service.go index d2236a8d..892ee00b 100644 --- a/internal/workflow/service.go +++ b/internal/workflow/service.go @@ -32,7 +32,7 @@ type WorkflowService struct { func NewWorkflowService(workflowRepo workflowRepository, workflowRunRepo workflowRunRepository) *WorkflowService { srv := &WorkflowService{ - dispatcher: dispatcher.GetSingletonDispatcher(workflowRepo, workflowRunRepo), + dispatcher: dispatcher.GetSingletonDispatcher(), workflowRepo: workflowRepo, workflowRunRepo: workflowRunRepo, @@ -83,6 +83,7 @@ func (s *WorkflowService) StartRun(ctx context.Context, req *dtos.WorkflowStartR Status: domain.WorkflowRunStatusTypePending, Trigger: req.RunTrigger, StartedAt: time.Now(), + Detail: workflow.Content, } if resp, err := s.workflowRunRepo.Save(ctx, run); err != nil { return err @@ -91,8 +92,8 @@ func (s *WorkflowService) StartRun(ctx context.Context, req *dtos.WorkflowStartR } s.dispatcher.Dispatch(&dispatcher.WorkflowWorkerData{ - WorkflowId: workflow.Id, - WorkflowContent: workflow.Content, + WorkflowId: run.WorkflowId, + WorkflowContent: run.Detail, RunId: run.Id, }) diff --git a/migrations/1740050400_collections_hotfix.go b/migrations/1740050400_upgrade.go similarity index 83% rename from migrations/1740050400_collections_hotfix.go rename to migrations/1740050400_upgrade.go index be193a17..93bc3f33 100644 --- a/migrations/1740050400_collections_hotfix.go +++ b/migrations/1740050400_upgrade.go @@ -7,11 +7,13 @@ import ( func init() { m.Register(func(app core.App) error { - certimateCollection, err := app.FindCollectionByNameOrId("4szxr9x43tpj6np") - if err != nil { - return err - } else { - // update field + // update collection `certificate` + { + certimateCollection, err := app.FindCollectionByNameOrId("4szxr9x43tpj6np") + if err != nil { + return err + } + if err := certimateCollection.Fields.AddMarshaledJSONAt(4, []byte(`{ "autogeneratePattern": "", "hidden": false, @@ -29,7 +31,6 @@ func init() { return err } - // update field if err := certimateCollection.Fields.AddMarshaledJSONAt(5, []byte(`{ "autogeneratePattern": "", "hidden": false, @@ -47,7 +48,6 @@ func init() { return err } - // update field if err := certimateCollection.Fields.AddMarshaledJSONAt(7, []byte(`{ "autogeneratePattern": "", "hidden": false, @@ -70,11 +70,13 @@ func init() { } } - workflowCollection, err := app.FindCollectionByNameOrId("tovyif5ax6j62ur") - if err != nil { - return err - } else { - // update field + // update collection `workflow` + { + workflowCollection, err := app.FindCollectionByNameOrId("tovyif5ax6j62ur") + if err != nil { + return err + } + if err := workflowCollection.Fields.AddMarshaledJSONAt(6, []byte(`{ "hidden": false, "id": "awlphkfe", @@ -88,7 +90,6 @@ func init() { return err } - // update field if err := workflowCollection.Fields.AddMarshaledJSONAt(7, []byte(`{ "hidden": false, "id": "g9ohkk5o", @@ -107,11 +108,13 @@ func init() { } } - workflowOutputCollection, err := app.FindCollectionByNameOrId("bqnxb95f2cooowp") - if err != nil { - return err - } else { - // update field + // update collection `workflow_output` + { + workflowOutputCollection, err := app.FindCollectionByNameOrId("bqnxb95f2cooowp") + if err != nil { + return err + } + if err := workflowOutputCollection.Fields.AddMarshaledJSONAt(4, []byte(`{ "hidden": false, "id": "c2rm9omj", diff --git a/migrations/1742209200_upgrade.go b/migrations/1742209200_upgrade.go new file mode 100644 index 00000000..8c9ede5f --- /dev/null +++ b/migrations/1742209200_upgrade.go @@ -0,0 +1,320 @@ +package migrations + +import ( + "encoding/json" + "strings" + + "github.com/pocketbase/pocketbase/core" + m "github.com/pocketbase/pocketbase/migrations" + + "github.com/usual2970/certimate/internal/domain" +) + +func init() { + m.Register(func(app core.App) error { + // create collection `workflow_logs` + { + jsonData := `{ + "createRule": null, + "deleteRule": null, + "fields": [ + { + "autogeneratePattern": "[a-z0-9]{15}", + "hidden": false, + "id": "text3208210256", + "max": 15, + "min": 15, + "name": "id", + "pattern": "^[a-z0-9]+$", + "presentable": false, + "primaryKey": true, + "required": true, + "system": true, + "type": "text" + }, + { + "cascadeDelete": true, + "collectionId": "tovyif5ax6j62ur", + "hidden": false, + "id": "relation3371272342", + "maxSelect": 1, + "minSelect": 0, + "name": "workflowId", + "presentable": false, + "required": false, + "system": false, + "type": "relation" + }, + { + "cascadeDelete": true, + "collectionId": "qjp8lygssgwyqyz", + "hidden": false, + "id": "relation821863227", + "maxSelect": 1, + "minSelect": 0, + "name": "runId", + "presentable": false, + "required": false, + "system": false, + "type": "relation" + }, + { + "autogeneratePattern": "", + "hidden": false, + "id": "text157423495", + "max": 0, + "min": 0, + "name": "nodeId", + "pattern": "", + "presentable": false, + "primaryKey": false, + "required": false, + "system": false, + "type": "text" + }, + { + "autogeneratePattern": "", + "hidden": false, + "id": "text3227511481", + "max": 0, + "min": 0, + "name": "nodeName", + "pattern": "", + "presentable": false, + "primaryKey": false, + "required": false, + "system": false, + "type": "text" + }, + { + "autogeneratePattern": "", + "hidden": false, + "id": "text2599078931", + "max": 0, + "min": 0, + "name": "level", + "pattern": "", + "presentable": false, + "primaryKey": false, + "required": false, + "system": false, + "type": "text" + }, + { + "autogeneratePattern": "", + "hidden": false, + "id": "text3065852031", + "max": 0, + "min": 0, + "name": "message", + "pattern": "", + "presentable": false, + "primaryKey": false, + "required": false, + "system": false, + "type": "text" + }, + { + "hidden": false, + "id": "json2918445923", + "maxSize": 0, + "name": "data", + "presentable": false, + "required": false, + "system": false, + "type": "json" + }, + { + "hidden": false, + "id": "autodate2990389176", + "name": "created", + "onCreate": true, + "onUpdate": false, + "presentable": false, + "system": false, + "type": "autodate" + } + ], + "id": "pbc_1682296116", + "indexes": [ + "CREATE INDEX ` + "`" + `idx_IOlpy6XuJ2` + "`" + ` ON ` + "`" + `workflow_logs` + "`" + ` (` + "`" + `workflowId` + "`" + `)", + "CREATE INDEX ` + "`" + `idx_qVlTb2yl7v` + "`" + ` ON ` + "`" + `workflow_logs` + "`" + ` (` + "`" + `runId` + "`" + `)" + ], + "listRule": null, + "name": "workflow_logs", + "system": false, + "type": "base", + "updateRule": null, + "viewRule": null + }` + + collection := &core.Collection{} + if err := json.Unmarshal([]byte(jsonData), &collection); err != nil { + return err + } + + if err := app.Save(collection); err != nil { + return err + } + } + + // migrate data + { + workflowRuns, err := app.FindAllRecords("workflow_run") + if err != nil { + return err + } + + for _, workflowRun := range workflowRuns { + type oldWorkflowRunLogRecord struct { + Time string `json:"time"` + Level string `json:"level"` + Content string `json:"content"` + Error string `json:"error"` + } + type oldWorkflowRunLog struct { + NodeId string `json:"nodeId"` + NodeName string `json:"nodeName"` + Records []oldWorkflowRunLogRecord `json:"records"` + Error string `json:"error"` + } + + logs := make([]oldWorkflowRunLog, 0) + if err := workflowRun.UnmarshalJSONField("logs", &logs); err != nil { + continue + } + + collection, err := app.FindCollectionByNameOrId("workflow_logs") + if err != nil { + return err + } + + for _, log := range logs { + for _, logRecord := range log.Records { + record := core.NewRecord(collection) + record.Set("workflowId", workflowRun.Get("workflowId")) + record.Set("runId", workflowRun.Get("id")) + record.Set("nodeId", log.NodeId) + record.Set("nodeName", log.NodeName) + record.Set("level", logRecord.Level) + record.Set("message", strings.TrimSpace(logRecord.Content+" "+logRecord.Error)) + record.Set("created", log.Records) + if err := app.Save(record); err != nil { + return err + } + } + } + } + } + + // update collection `workflow_run` + { + collection, err := app.FindCollectionByNameOrId("workflow_run") + if err != nil { + return err + } + + if err := collection.Fields.AddMarshaledJSONAt(6, []byte(`{ + "hidden": false, + "id": "json772177811", + "maxSize": 5000000, + "name": "detail", + "presentable": false, + "required": false, + "system": false, + "type": "json" + }`)); err != nil { + return err + } + + if err := app.Save(collection); err != nil { + return err + } + } + + // migrate data + { + workflowRuns, err := app.FindAllRecords("workflow_run") + if err != nil { + return err + } + + workflowOutputs, err := app.FindAllRecords("workflow_output") + if err != nil { + return err + } + + for _, workflowRun := range workflowRuns { + node := &domain.WorkflowNode{} + for _, workflowOutput := range workflowOutputs { + if workflowOutput.GetString("runId") != workflowRun.Get("id") { + continue + } + + if err := workflowOutput.UnmarshalJSONField("node", node); err != nil { + continue + } + + if node.Type != domain.WorkflowNodeTypeApply { + node = &domain.WorkflowNode{} + continue + } + } + + if node.Id == "" { + workflow, _ := app.FindRecordById("workflow", workflowRun.GetString("workflowId")) + if workflow != nil { + workflowRun.Set("detail", workflow.Get("content")) + } else { + workflowRun.Set("detail", make(map[string]any)) + } + } else { + workflow, _ := app.FindRecordById("workflow", workflowRun.GetString("workflowId")) + if workflow != nil { + rootNode := &domain.WorkflowNode{} + if err := workflow.UnmarshalJSONField("content", rootNode); err != nil { + return err + } + + rootNode.Next = node + workflowRun.Set("detail", rootNode) + } else { + rootNode := &domain.WorkflowNode{ + Id: core.GenerateDefaultRandomId(), + Type: domain.WorkflowNodeTypeStart, + Name: "开始", + Config: map[string]any{ + "trigger": "manual", + }, + Next: node, + Validated: true, + } + workflowRun.Set("detail", rootNode) + } + } + + if err := app.Save(workflowRun); err != nil { + return err + } + } + } + + // update collection `workflow_run` + { + collection, err := app.FindCollectionByNameOrId("workflow_run") + if err != nil { + return err + } + + collection.Fields.RemoveByName("logs") + + if err := app.Save(collection); err != nil { + return err + } + } + + return nil + }, func(app core.App) error { + return nil + }) +}