Implement complete workflow execution process

internal/repository/workflow_output.go

@@ -18,13 +18,17 @@ func NewWorkflowOutputRepository() *WorkflowOutputRepository {
 }
 
 func (w *WorkflowOutputRepository) Get(ctx context.Context, nodeId string) (*domain.WorkflowOutput, error) {
-	record, err := app.GetApp().Dao().FindFirstRecordByFilter("workflow_output", "nodeId={:nodeId}", dbx.Params{"nodeId": nodeId})
+	records, err := app.GetApp().Dao().FindRecordsByFilter("workflow_output", "nodeId={:nodeId}", "-created", 1, 0, dbx.Params{"nodeId": nodeId})
 	if err != nil {
 		if errors.Is(err, sql.ErrNoRows) {
 			return nil, domain.ErrRecordNotFound
 		}
 		return nil, err
 	}
+	if len(records) == 0 {
+		return nil, domain.ErrRecordNotFound
+	}
+	record := records[0]
 
 	node := &domain.WorkflowNode{}
 	if err := record.UnmarshalJSONField("node", node); err != nil {
@@ -46,11 +50,46 @@ func (w *WorkflowOutputRepository) Get(ctx context.Context, nodeId string) (*dom
 		NodeId:   record.GetString("nodeId"),
 		Node:     node,
 		Output:   output,
+		Succeed:  record.GetBool("succeed"),
 	}
 
 	return rs, nil
 }
 
+func (w *WorkflowOutputRepository) GetCertificate(ctx context.Context, nodeId string) (*domain.Certificate, error) {
+	records, err := app.GetApp().Dao().FindRecordsByFilter("certificate", "nodeId={:nodeId}", "-created", 1, 0, dbx.Params{"nodeId": nodeId})
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, domain.ErrRecordNotFound
+		}
+		return nil, err
+	}
+	if len(records) == 0 {
+		return nil, domain.ErrRecordNotFound
+	}
+
+	record := records[0]
+
+	rs := &domain.Certificate{
+		Meta: domain.Meta{
+			Id:      record.GetId(),
+			Created: record.GetTime("created"),
+			Updated: record.GetTime("updated"),
+		},
+		Certificate:       record.GetString("certificate"),
+		PrivateKey:        record.GetString("privateKey"),
+		IssuerCertificate: record.GetString("issuerCertificate"),
+		SAN:               record.GetString("san"),
+		Output:            record.GetString("output"),
+		ExpireAt:          record.GetTime("expireAt"),
+		CertUrl:           record.GetString("certUrl"),
+		CertStableUrl:     record.GetString("certStableUrl"),
+		Workflow:          record.GetString("workflow"),
+		NodeId:            record.GetString("nodeId"),
+	}
+	return rs, nil
+}
+
 // 保存节点输出
 func (w *WorkflowOutputRepository) Save(ctx context.Context, output *domain.WorkflowOutput, certificate *domain.Certificate, cb func(id string) error) error {
 	var record *models.Record
@@ -72,6 +111,7 @@ func (w *WorkflowOutputRepository) Save(ctx context.Context, output *domain.Work
 	record.Set("nodeId", output.NodeId)
 	record.Set("node", output.Node)
 	record.Set("output", output.Output)
+	record.Set("succeed", output.Succeed)
 
 	if err := app.GetApp().Dao().SaveRecord(record); err != nil {
 		return err
@@ -92,14 +132,31 @@ func (w *WorkflowOutputRepository) Save(ctx context.Context, output *domain.Work
 		certRecord.Set("privateKey", certificate.PrivateKey)
 		certRecord.Set("issuerCertificate", certificate.IssuerCertificate)
 		certRecord.Set("san", certificate.SAN)
-		certRecord.Set("workflowOutput", certificate.Output)
+		certRecord.Set("output", certificate.Output)
 		certRecord.Set("expireAt", certificate.ExpireAt)
 		certRecord.Set("certUrl", certificate.CertUrl)
 		certRecord.Set("certStableUrl", certificate.CertStableUrl)
+		certRecord.Set("workflow", certificate.Workflow)
+		certRecord.Set("nodeId", certificate.NodeId)
 
 		if err := app.GetApp().Dao().SaveRecord(certRecord); err != nil {
 			return err
 		}
+
+		// 更新 certificate
+		for i, item := range output.Output {
+			if item.Name == "certificate" {
+				output.Output[i].Value = certRecord.GetId()
+				break
+			}
+		}
+
+		record.Set("output", output.Output)
+
+		if err := app.GetApp().Dao().SaveRecord(record); err != nil {
+			return err
+		}
+
 	}
 	return nil
 }