feat: save related runId in certificates or workflow outputs

internal/domain/certificate.go

@@ -27,6 +27,7 @@ type Certificate struct {
 	ACMECertStableUrl string                      `json:"acmeCertStableUrl" db:"acmeCertStableUrl"`
 	WorkflowId        string                      `json:"workflowId" db:"workflowId"`
 	WorkflowNodeId    string                      `json:"workflowNodeId" db:"workflowNodeId"`
+	WorkflowRunId     string                      `json:"workflowRunId" db:"workflowRunId"`
 	WorkflowOutputId  string                      `json:"workflowOutputId" db:"workflowOutputId"`
 	DeletedAt         *time.Time                  `json:"deleted" db:"deleted"`
 }

internal/domain/workflow_output.go

@@ -5,6 +5,7 @@ const CollectionNameWorkflowOutput = "workflow_output"
 type WorkflowOutput struct {
 	Meta
 	WorkflowId string           `json:"workflowId" db:"workflow"`
+	RunId      string           `json:"runId" db:"runId"`
 	NodeId     string           `json:"nodeId" db:"nodeId"`
 	Node       *WorkflowNode    `json:"node" db:"node"`
 	Outputs    []WorkflowNodeIO `json:"outputs" db:"outputs"`

internal/repository/certificate.go

@@ -112,6 +112,7 @@ func (r *CertificateRepository) Save(ctx context.Context, certificate *domain.Ce
 	record.Set("acmeCertUrl", certificate.ACMECertUrl)
 	record.Set("acmeCertStableUrl", certificate.ACMECertStableUrl)
 	record.Set("workflowId", certificate.WorkflowId)
+	record.Set("workflowRunId", certificate.WorkflowRunId)
 	record.Set("workflowNodeId", certificate.WorkflowNodeId)
 	record.Set("workflowOutputId", certificate.WorkflowOutputId)
 	if err := app.GetApp().Save(record); err != nil {
@@ -149,6 +150,7 @@ func (r *CertificateRepository) castRecordToModel(record *core.Record) (*domain.
 		ACMECertUrl:       record.GetString("acmeCertUrl"),
 		ACMECertStableUrl: record.GetString("acmeCertStableUrl"),
 		WorkflowId:        record.GetString("workflowId"),
+		WorkflowRunId:     record.GetString("workflowRunId"),
 		WorkflowNodeId:    record.GetString("workflowNodeId"),
 		WorkflowOutputId:  record.GetString("workflowOutputId"),
 	}

internal/repository/workflow_output.go

@@ -40,7 +40,7 @@ func (r *WorkflowOutputRepository) GetByNodeId(ctx context.Context, workflowNode
 }
 
 func (r *WorkflowOutputRepository) Save(ctx context.Context, workflowOutput *domain.WorkflowOutput) (*domain.WorkflowOutput, error) {
-	record, err := r.saveRecord(ctx, workflowOutput)
+	record, err := r.saveRecord(workflowOutput)
 	if err != nil {
 		return workflowOutput, err
 	}
@@ -52,7 +52,7 @@ func (r *WorkflowOutputRepository) Save(ctx context.Context, workflowOutput *dom
 }
 
 func (r *WorkflowOutputRepository) SaveWithCertificate(ctx context.Context, workflowOutput *domain.WorkflowOutput, certificate *domain.Certificate) (*domain.WorkflowOutput, error) {
-	record, err := r.saveRecord(ctx, workflowOutput)
+	record, err := r.saveRecord(workflowOutput)
 	if err != nil {
 		return workflowOutput, err
 	} else {
@@ -63,6 +63,7 @@ func (r *WorkflowOutputRepository) SaveWithCertificate(ctx context.Context, work
 
 	if certificate != nil {
 		certificate.WorkflowId = workflowOutput.WorkflowId
+		certificate.WorkflowRunId = workflowOutput.RunId
 		certificate.WorkflowNodeId = workflowOutput.NodeId
 		certificate.WorkflowOutputId = workflowOutput.Id
 		certificate, err := NewCertificateRepository().Save(ctx, certificate)
@@ -108,6 +109,7 @@ func (r *WorkflowOutputRepository) castRecordToModel(record *core.Record) (*doma
 			UpdatedAt: record.GetDateTime("updated").Time(),
 		},
 		WorkflowId: record.GetString("workflowId"),
+		RunId:      record.GetString("runId"),
 		NodeId:     record.GetString("nodeId"),
 		Node:       node,
 		Outputs:    outputs,
@@ -116,7 +118,7 @@ func (r *WorkflowOutputRepository) castRecordToModel(record *core.Record) (*doma
 	return workflowOutput, nil
 }
 
-func (r *WorkflowOutputRepository) saveRecord(ctx context.Context, output *domain.WorkflowOutput) (*core.Record, error) {
+func (r *WorkflowOutputRepository) saveRecord(output *domain.WorkflowOutput) (*core.Record, error) {
 	collection, err := app.GetApp().FindCollectionByNameOrId(domain.CollectionNameWorkflowOutput)
 	if err != nil {
 		return nil, err
@@ -132,6 +134,7 @@ func (r *WorkflowOutputRepository) saveRecord(ctx context.Context, output *domai
 		}
 	}
 	record.Set("workflowId", output.WorkflowId)
+	record.Set("runId", output.RunId)
 	record.Set("nodeId", output.NodeId)
 	record.Set("node", output.Node)
 	record.Set("outputs", output.Outputs)

internal/workflow/node-processor/apply_node.go

@@ -14,18 +14,20 @@ import (
 )
 
 type applyNode struct {
-	node       *domain.WorkflowNode
+	node *domain.WorkflowNode
+	*nodeLogger
+
 	certRepo   certificateRepository
 	outputRepo workflowOutputRepository
-	*nodeLogger
 }
 
 func NewApplyNode(node *domain.WorkflowNode) *applyNode {
 	return &applyNode{
 		node:       node,
 		nodeLogger: NewNodeLogger(node),
-		outputRepo: repository.NewWorkflowOutputRepository(),
+
 		certRepo:   repository.NewCertificateRepository(),
+		outputRepo: repository.NewWorkflowOutputRepository(),
 	}
 }
 
@@ -81,6 +83,7 @@ func (n *applyNode) Run(ctx context.Context) error {
 	// TODO: 先保持一个节点始终只有一个输出，后续增加版本控制
 	currentOutput := &domain.WorkflowOutput{
 		WorkflowId: getContextWorkflowId(ctx),
+		RunId:      getContextWorkflowRunId(ctx),
 		NodeId:     n.node.Id,
 		Node:       n.node,
 		Succeeded:  true,

internal/workflow/node-processor/deploy_node.go

@@ -12,18 +12,20 @@ import (
 )
 
 type deployNode struct {
-	node       *domain.WorkflowNode
+	node *domain.WorkflowNode
+	*nodeLogger
+
 	certRepo   certificateRepository
 	outputRepo workflowOutputRepository
-	*nodeLogger
 }
 
 func NewDeployNode(node *domain.WorkflowNode) *deployNode {
 	return &deployNode{
 		node:       node,
 		nodeLogger: NewNodeLogger(node),
-		outputRepo: repository.NewWorkflowOutputRepository(),
+
 		certRepo:   repository.NewCertificateRepository(),
+		outputRepo: repository.NewWorkflowOutputRepository(),
 	}
 }
 
@@ -61,7 +63,7 @@ func (n *deployNode) Run(ctx context.Context) error {
 	}
 
 	// 初始化部署器
-	deploy, err := deployer.NewWithDeployNode(n.node, struct {
+	deployer, err := deployer.NewWithDeployNode(n.node, struct {
 		Certificate string
 		PrivateKey  string
 	}{Certificate: certificate.Certificate, PrivateKey: certificate.PrivateKey})
@@ -71,7 +73,7 @@ func (n *deployNode) Run(ctx context.Context) error {
 	}
 
 	// 部署证书
-	if err := deploy.Deploy(ctx); err != nil {
+	if err := deployer.Deploy(ctx); err != nil {
 		n.AddOutput(ctx, n.node.Name, "部署失败", err.Error())
 		return err
 	}
@@ -80,8 +82,8 @@ func (n *deployNode) Run(ctx context.Context) error {
 	// 保存执行结果
 	// TODO: 先保持一个节点始终只有一个输出，后续增加版本控制
 	currentOutput := &domain.WorkflowOutput{
-		Meta:       domain.Meta{},
 		WorkflowId: getContextWorkflowId(ctx),
+		RunId:      getContextWorkflowRunId(ctx),
 		NodeId:     n.node.Id,
 		Node:       n.node,
 		Succeeded:  true,

internal/workflow/node-processor/notify_node.go

@@ -9,15 +9,17 @@ import (
 )
 
 type notifyNode struct {
-	node         *domain.WorkflowNode
+	node *domain.WorkflowNode
-	settingsRepo settingsRepository
 	*nodeLogger
+
+	settingsRepo settingsRepository
 }
 
 func NewNotifyNode(node *domain.WorkflowNode) *notifyNode {
 	return &notifyNode{
-		node:         node,
+		node:       node,
-		nodeLogger:   NewNodeLogger(node),
+		nodeLogger: NewNodeLogger(node),
+
 		settingsRepo: repository.NewSettingsRepository(),
 	}
 }

internal/workflow/node-processor/processor.go

@@ -10,7 +10,7 @@ import (
 
 type NodeProcessor interface {
 	Run(ctx context.Context) error
-	Log(ctx context.Context) *domain.WorkflowRunLog
+	GetLog(ctx context.Context) *domain.WorkflowRunLog
 	AddOutput(ctx context.Context, title, content string, err ...string)
 }
 
@@ -42,7 +42,7 @@ func NewNodeLogger(node *domain.WorkflowNode) *nodeLogger {
 	}
 }
 
-func (l *nodeLogger) Log(ctx context.Context) *domain.WorkflowRunLog {
+func (l *nodeLogger) GetLog(ctx context.Context) *domain.WorkflowRunLog {
 	return l.log
 }
 
@@ -84,3 +84,7 @@ func GetProcessor(node *domain.WorkflowNode) (NodeProcessor, error) {
 func getContextWorkflowId(ctx context.Context) string {
 	return ctx.Value("workflow_id").(string)
 }
+
+func getContextWorkflowRunId(ctx context.Context) string {
+	return ctx.Value("workflow_run_id").(string)
+}

internal/workflow/node-processor/upload_node.go

@@ -3,6 +3,7 @@ package nodeprocessor
 import (
 	"context"
 	"errors"
+	"strings"
 	"time"
 
 	"github.com/usual2970/certimate/internal/domain"
@@ -11,15 +12,19 @@ import (
 )
 
 type uploadNode struct {
-	node       *domain.WorkflowNode
+	node *domain.WorkflowNode
-	outputRepo workflowOutputRepository
 	*nodeLogger
+
+	certRepo   certificateRepository
+	outputRepo workflowOutputRepository
 }
 
 func NewUploadNode(node *domain.WorkflowNode) *uploadNode {
 	return &uploadNode{
 		node:       node,
 		nodeLogger: NewNodeLogger(node),
+
+		certRepo:   repository.NewCertificateRepository(),
 		outputRepo: repository.NewWorkflowOutputRepository(),
 	}
 }
@@ -38,6 +43,12 @@ func (n *uploadNode) Run(ctx context.Context) error {
 		return err
 	}
 
+	// 检测是否可以跳过本次执行
+	if skippable, skipReason := n.checkCanSkip(ctx, lastOutput); skippable {
+		n.AddOutput(ctx, n.node.Name, skipReason)
+		return nil
+	}
+
 	// 检查证书是否过期
 	// 如果证书过期，则直接返回错误
 	certX509, err := certs.ParseCertificateFromPEM(nodeConfig.Certificate)
@@ -50,7 +61,7 @@ func (n *uploadNode) Run(ctx context.Context) error {
 		return errors.New("certificate is expired")
 	}
 
-	// 生成实体
+	// 生成证书实体
 	certificate := &domain.Certificate{
 		Source: domain.CertificateSourceTypeUpload,
 	}
@@ -60,6 +71,7 @@ func (n *uploadNode) Run(ctx context.Context) error {
 	// TODO: 先保持一个节点始终只有一个输出，后续增加版本控制
 	currentOutput := &domain.WorkflowOutput{
 		WorkflowId: getContextWorkflowId(ctx),
+		RunId:      getContextWorkflowRunId(ctx),
 		NodeId:     n.node.Id,
 		Node:       n.node,
 		Succeeded:  true,
@@ -76,3 +88,24 @@ func (n *uploadNode) Run(ctx context.Context) error {
 
 	return nil
 }
+
+func (n *uploadNode) checkCanSkip(ctx context.Context, lastOutput *domain.WorkflowOutput) (skip bool, reason string) {
+	if lastOutput != nil && lastOutput.Succeeded {
+		// 比较和上次上传时的关键配置（即影响证书上传的）参数是否一致
+		currentNodeConfig := n.node.GetConfigForUpload()
+		lastNodeConfig := lastOutput.Node.GetConfigForUpload()
+		if strings.TrimSpace(currentNodeConfig.Certificate) != strings.TrimSpace(lastNodeConfig.Certificate) {
+			return false, "配置项变化：证书"
+		}
+		if strings.TrimSpace(currentNodeConfig.PrivateKey) != strings.TrimSpace(lastNodeConfig.PrivateKey) {
+			return false, "配置项变化：私钥"
+		}
+
+		lastCertificate, _ := n.certRepo.GetByWorkflowNodeId(ctx, n.node.Id)
+		if lastCertificate != nil {
+			return true, "已上传过证书"
+		}
+	}
+
+	return false, ""
+}

internal/workflow/processor/processor.go

@@ -8,24 +8,27 @@ import (
 )
 
 type workflowProcessor struct {
-	workflow *domain.Workflow
+	workflow        *domain.Workflow
-	logs     []domain.WorkflowRunLog
+	workflowRun     *domain.WorkflowRun
+	workflorRunLogs []domain.WorkflowRunLog
 }
 
-func NewWorkflowProcessor(workflow *domain.Workflow) *workflowProcessor {
+func NewWorkflowProcessor(workflow *domain.Workflow, workflowRun *domain.WorkflowRun) *workflowProcessor {
 	return &workflowProcessor{
-		workflow: workflow,
+		workflow:        workflow,
-		logs:     make([]domain.WorkflowRunLog, 0),
+		workflowRun:     workflowRun,
+		workflorRunLogs: make([]domain.WorkflowRunLog, 0),
 	}
 }
 
 func (w *workflowProcessor) Run(ctx context.Context) error {
-	ctx = setContextWorkflowId(ctx, w.workflow.Id)
+	ctx = context.WithValue(ctx, "workflow_id", w.workflow.Id)
+	ctx = context.WithValue(ctx, "workflow_run_id", w.workflowRun.Id)
 	return w.processNode(ctx, w.workflow.Content)
 }
 
 func (w *workflowProcessor) GetRunLogs() []domain.WorkflowRunLog {
-	return w.logs
+	return w.workflorRunLogs
 }
 
 func (w *workflowProcessor) processNode(ctx context.Context, node *domain.WorkflowNode) error {
@@ -49,9 +52,9 @@ func (w *workflowProcessor) processNode(ctx context.Context, node *domain.Workfl
 				}
 
 				runErr = processor.Run(ctx)
-				log := processor.Log(ctx)
+				log := processor.GetLog(ctx)
 				if log != nil {
-					w.logs = append(w.logs, *log)
+					w.workflorRunLogs = append(w.workflorRunLogs, *log)
 				}
 				if runErr != nil {
 					break
@@ -75,10 +78,6 @@ func (w *workflowProcessor) processNode(ctx context.Context, node *domain.Workfl
 	return nil
 }
 
-func setContextWorkflowId(ctx context.Context, id string) context.Context {
-	return context.WithValue(ctx, "workflow_id", id)
-}
-
 func getBranchByType(branches []domain.WorkflowNode, nodeType domain.WorkflowNodeType) *domain.WorkflowNode {
 	for _, branch := range branches {
 		if branch.Type == nodeType {

internal/workflow/service.go

@@ -141,7 +141,7 @@ func (s *WorkflowService) runWithData(ctx context.Context, runData *workflowRunD
 		run = resp
 	}
 
-	processor := processor.NewWorkflowProcessor(workflow)
+	processor := processor.NewWorkflowProcessor(workflow, run)
 	if runErr := processor.Run(ctx); runErr != nil {
 		run.Status = domain.WorkflowRunStatusTypeFailed
 		run.EndedAt = time.Now()

migrations/1738839725_updated_certificate.go

@@ -0,0 +1,67 @@
+package migrations
+
+import (
+	"encoding/json"
+
+	"github.com/pocketbase/pocketbase/core"
+	m "github.com/pocketbase/pocketbase/migrations"
+)
+
+func init() {
+	m.Register(func(app core.App) error {
+		collection, err := app.FindCollectionByNameOrId("4szxr9x43tpj6np")
+		if err != nil {
+			return err
+		}
+
+		// update collection data
+		if err := json.Unmarshal([]byte(`{
+			"indexes": [
+				"CREATE INDEX ` + "`" + `idx_Jx8TXzDCmw` + "`" + ` ON ` + "`" + `certificate` + "`" + ` (` + "`" + `workflowId` + "`" + `)",
+				"CREATE INDEX ` + "`" + `idx_kcKpgAZapk` + "`" + ` ON ` + "`" + `certificate` + "`" + ` (` + "`" + `workflowNodeId` + "`" + `)",
+				"CREATE INDEX ` + "`" + `idx_2cRXqNDyyp` + "`" + ` ON ` + "`" + `certificate` + "`" + ` (` + "`" + `workflowRunId` + "`" + `)"
+			]
+		}`), &collection); err != nil {
+			return err
+		}
+
+		// add field
+		if err := collection.Fields.AddMarshaledJSONAt(15, []byte(`{
+			"cascadeDelete": false,
+			"collectionId": "qjp8lygssgwyqyz",
+			"hidden": false,
+			"id": "relation3917999135",
+			"maxSelect": 1,
+			"minSelect": 0,
+			"name": "workflowRunId",
+			"presentable": false,
+			"required": false,
+			"system": false,
+			"type": "relation"
+		}`)); err != nil {
+			return err
+		}
+
+		return app.Save(collection)
+	}, func(app core.App) error {
+		collection, err := app.FindCollectionByNameOrId("4szxr9x43tpj6np")
+		if err != nil {
+			return err
+		}
+
+		// update collection data
+		if err := json.Unmarshal([]byte(`{
+			"indexes": [
+				"CREATE INDEX ` + "`" + `idx_Jx8TXzDCmw` + "`" + ` ON ` + "`" + `certificate` + "`" + ` (` + "`" + `workflowId` + "`" + `)",
+				"CREATE INDEX ` + "`" + `idx_kcKpgAZapk` + "`" + ` ON ` + "`" + `certificate` + "`" + ` (` + "`" + `workflowNodeId` + "`" + `)"
+			]
+		}`), &collection); err != nil {
+			return err
+		}
+
+		// remove field
+		collection.Fields.RemoveById("relation3917999135")
+
+		return app.Save(collection)
+	})
+}

migrations/1738840633_updated_workflow_output.go

@@ -0,0 +1,63 @@
+package migrations
+
+import (
+	"encoding/json"
+
+	"github.com/pocketbase/pocketbase/core"
+	m "github.com/pocketbase/pocketbase/migrations"
+)
+
+func init() {
+	m.Register(func(app core.App) error {
+		collection, err := app.FindCollectionByNameOrId("bqnxb95f2cooowp")
+		if err != nil {
+			return err
+		}
+
+		// update collection data
+		if err := json.Unmarshal([]byte(`{
+			"indexes": [
+				"CREATE INDEX ` + "`" + `idx_BYoQPsz4my` + "`" + ` ON ` + "`" + `workflow_output` + "`" + ` (` + "`" + `workflowId` + "`" + `)",
+				"CREATE INDEX ` + "`" + `idx_O9zxLETuxJ` + "`" + ` ON ` + "`" + `workflow_output` + "`" + ` (` + "`" + `runId` + "`" + `)"
+			]
+		}`), &collection); err != nil {
+			return err
+		}
+
+		// add field
+		if err := collection.Fields.AddMarshaledJSONAt(2, []byte(`{
+			"cascadeDelete": false,
+			"collectionId": "qjp8lygssgwyqyz",
+			"hidden": false,
+			"id": "relation821863227",
+			"maxSelect": 1,
+			"minSelect": 0,
+			"name": "runId",
+			"presentable": false,
+			"required": false,
+			"system": false,
+			"type": "relation"
+		}`)); err != nil {
+			return err
+		}
+
+		return app.Save(collection)
+	}, func(app core.App) error {
+		collection, err := app.FindCollectionByNameOrId("bqnxb95f2cooowp")
+		if err != nil {
+			return err
+		}
+
+		// update collection data
+		if err := json.Unmarshal([]byte(`{
+			"indexes": []
+		}`), &collection); err != nil {
+			return err
+		}
+
+		// remove field
+		collection.Fields.RemoveById("relation821863227")
+
+		return app.Save(collection)
+	})
+}