Implement complete workflow execution process

internal/deployer/deployer.go

@@ -16,7 +16,7 @@ import (
 	"github.com/usual2970/certimate/internal/applicant"
 	"github.com/usual2970/certimate/internal/domain"
 	"github.com/usual2970/certimate/internal/pkg/utils/x509"
-	"github.com/usual2970/certimate/internal/utils/app"
+	"github.com/usual2970/certimate/internal/repository"
 )
 
 const (
@@ -49,7 +49,7 @@ type DeployerOption struct {
 	DomainId     string                `json:"domainId"`
 	Domain       string                `json:"domain"`
 	Access       string                `json:"access"`
-	AccessRecord *models.Record        `json:"-"`
+	AccessRecord *domain.Access        `json:"-"`
 	DeployConfig domain.DeployConfig   `json:"deployConfig"`
 	Certificate  applicant.Certificate `json:"certificate"`
 	Variables    map[string]string     `json:"variables"`
@@ -90,16 +90,21 @@ func Gets(record *models.Record, cert *applicant.Certificate) ([]Deployer, error
 	return rs, nil
 }
 
+func GetWithTypeAndOption(deployType string, option *DeployerOption) (Deployer, error) {
+	return getWithTypeAndOption(deployType, option)
+}
+
 func getWithDeployConfig(record *models.Record, cert *applicant.Certificate, deployConfig domain.DeployConfig) (Deployer, error) {
-	access, err := app.GetApp().Dao().FindRecordById("access", deployConfig.Access)
+	accessRepo := repository.NewAccessRepository()
+	access, err := accessRepo.GetById(context.Background(), deployConfig.Access)
 	if err != nil {
-		return nil, fmt.Errorf("access record not found: %w", err)
+		return nil, fmt.Errorf("获取access失败:%w", err)
 	}
 
 	option := &DeployerOption{
 		DomainId:     record.Id,
 		Domain:       record.GetString("domain"),
-		Access:       access.GetString("config"),
+		Access:       access.Config,
 		AccessRecord: access,
 		DeployConfig: deployConfig,
 	}
@@ -112,7 +117,11 @@ func getWithDeployConfig(record *models.Record, cert *applicant.Certificate, dep
 		}
 	}
 
-	switch deployConfig.Type {
+	return getWithTypeAndOption(deployConfig.Type, option)
+}
+
+func getWithTypeAndOption(deployType string, option *DeployerOption) (Deployer, error) {
+	switch deployType {
 	case targetAliyunOSS:
 		return NewAliyunOSSDeployer(option)
 	case targetAliyunCDN:

internal/domain/access.go

@@ -11,6 +11,16 @@ type Access struct {
 	Usage      string    `json:"usage"`
 }
 
+// 兼容一下原 pocketbase 的 record
+func (a *Access) GetString(key string) string {
+	switch key {
+	case "name":
+		return a.Name
+	default:
+		return ""
+	}
+}
+
 type AliyunAccess struct {
 	AccessKeyId     string `json:"accessKeyId"`
 	AccessKeySecret string `json:"accessKeySecret"`

internal/domain/certificate.go

@@ -11,7 +11,9 @@ type Certificate struct {
 	CertUrl           string    `json:"certUrl"`
 	CertStableUrl     string    `json:"certStableUrl"`
 	Output            string    `json:"output"`
+	Workflow          string    `json:"workflow"`
 	ExpireAt          time.Time `json:"ExpireAt"`
+	NodeId            string    `json:"nodeId"`
 }
 
 type MetaData struct {

internal/domain/workflow.go

@@ -9,7 +9,7 @@ const (
 	WorkflowNodeTypeStart     = "start"
 	WorkflowNodeTypeEnd       = "end"
 	WorkflowNodeTypeApply     = "apply"
-	WorkflowNodeTypeDeply     = "deploy"
+	WorkflowNodeTypeDeploy    = "deploy"
 	WorkflowNodeTypeNotify    = "notify"
 	WorkflowNodeTypeBranch    = "branch"
 	WorkflowNodeTypeCondition = "condition"

internal/repository/access.go

@@ -2,8 +2,11 @@ package repository
 
 import (
 	"context"
+	"database/sql"
+	"errors"
 
 	"github.com/usual2970/certimate/internal/domain"
+	"github.com/usual2970/certimate/internal/utils/app"
 )
 
 type AccessRepository struct{}
@@ -13,5 +16,24 @@ func NewAccessRepository() *AccessRepository {
 }
 
 func (a *AccessRepository) GetById(ctx context.Context, id string) (*domain.Access, error) {
-	return nil, nil
+	record, err := app.GetApp().Dao().FindRecordById("access", id)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, domain.ErrRecordNotFound
+		}
+		return nil, err
+	}
+
+	rs := &domain.Access{
+		Meta: domain.Meta{
+			Id:      record.GetId(),
+			Created: record.GetTime("created"),
+			Updated: record.GetTime("updated"),
+		},
+		Name:       record.GetString("name"),
+		Config:     record.GetString("config"),
+		ConfigType: record.GetString("configType"),
+		Usage:      record.GetString("usage"),
+	}
+	return rs, nil
 }

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
 }

internal/routes/routes.go

@@ -4,16 +4,22 @@ import (
 	"github.com/usual2970/certimate/internal/notify"
 	"github.com/usual2970/certimate/internal/repository"
 	"github.com/usual2970/certimate/internal/rest"
+	"github.com/usual2970/certimate/internal/workflow"
 
 	"github.com/labstack/echo/v5"
 	"github.com/pocketbase/pocketbase/apis"
 )
 
 func Register(e *echo.Echo) {
+	group := e.Group("/api", apis.RequireAdminAuth())
+
 	notifyRepo := repository.NewSettingRepository()
 	notifySvc := notify.NewNotifyService(notifyRepo)
 
-	group := e.Group("/api", apis.RequireAdminAuth())
+	workflowRepo := repository.NewWorkflowRepository()
+	workflowSvc := workflow.NewWorkflowService(workflowRepo)
+
+	rest.NewWorkflowHandler(group, workflowSvc)
 
 	rest.NewNotifyHandler(group, notifySvc)
 }

internal/workflow/node-processor/apply_node.go

@@ -7,7 +7,6 @@ import (
 	"github.com/usual2970/certimate/internal/domain"
 	"github.com/usual2970/certimate/internal/pkg/utils/x509"
 	"github.com/usual2970/certimate/internal/repository"
-	"github.com/usual2970/certimate/internal/utils/xtime"
 )
 
 type applyNode struct {
@@ -28,39 +27,42 @@ type WorkflowOutputRepository interface {
 	// 查询节点输出
 	Get(ctx context.Context, nodeId string) (*domain.WorkflowOutput, error)
 
+	// 查询申请节点的证书
+	GetCertificate(ctx context.Context, nodeId string) (*domain.Certificate, error)
+
 	// 保存节点输出
 	Save(ctx context.Context, output *domain.WorkflowOutput, certificate *domain.Certificate, cb func(id string) error) error
 }
 
 // 申请节点根据申请类型执行不同的操作
 func (a *applyNode) Run(ctx context.Context) error {
-	a.AddOutput(ctx, xtime.BeijingTimeStr(), a.node.Name, "开始执行")
+	a.AddOutput(ctx, a.node.Name, "开始执行")
 	// 查询是否申请过，已申请过则直接返回（先保持和 v0.2 一致）
 	output, err := a.outputRepo.Get(ctx, a.node.Id)
 	if err != nil && !domain.IsRecordNotFound(err) {
-		a.AddOutput(ctx, xtime.BeijingTimeStr(), a.node.Name, "查询申请记录失败", err.Error())
+		a.AddOutput(ctx, a.node.Name, "查询申请记录失败", err.Error())
 		return err
 	}
 
 	if output != nil && output.Succeed {
-		a.AddOutput(ctx, xtime.BeijingTimeStr(), a.node.Name, "已申请过")
+		a.AddOutput(ctx, a.node.Name, "已申请过")
 		return nil
 	}
 
 	// 获取Applicant
 	apply, err := applicant.GetWithApplyNode(a.node)
 	if err != nil {
-		a.AddOutput(ctx, xtime.BeijingTimeStr(), a.node.Name, "获取申请对象失败", err.Error())
+		a.AddOutput(ctx, a.node.Name, "获取申请对象失败", err.Error())
 		return err
 	}
 
 	// 申请
 	certificate, err := apply.Apply()
 	if err != nil {
-		a.AddOutput(ctx, xtime.BeijingTimeStr(), a.node.Name, "申请失败", err.Error())
+		a.AddOutput(ctx, a.node.Name, "申请失败", err.Error())
 		return err
 	}
-	a.AddOutput(ctx, xtime.BeijingTimeStr(), a.node.Name, "申请成功")
+	a.AddOutput(ctx, a.node.Name, "申请成功")
 
 	// 记录申请结果
 	output = &domain.WorkflowOutput{
@@ -68,11 +70,12 @@ func (a *applyNode) Run(ctx context.Context) error {
 		NodeId:   a.node.Id,
 		Node:     a.node,
 		Succeed:  true,
+		Output:   a.node.Output,
 	}
 
 	cert, err := x509.ParseCertificateFromPEM(certificate.Certificate)
 	if err != nil {
-		a.AddOutput(ctx, xtime.BeijingTimeStr(), a.node.Name, "解析证书失败", err.Error())
+		a.AddOutput(ctx, a.node.Name, "解析证书失败", err.Error())
 		return err
 	}
 
@@ -84,20 +87,22 @@ func (a *applyNode) Run(ctx context.Context) error {
 		CertUrl:           certificate.CertUrl,
 		CertStableUrl:     certificate.CertStableUrl,
 		ExpireAt:          cert.NotAfter,
+		Workflow:          GetWorkflowId(ctx),
+		NodeId:            a.node.Id,
 	}
 
 	if err := a.outputRepo.Save(ctx, output, certificateRecord, func(id string) error {
 		if certificateRecord != nil {
-			certificateRecord.Id = id
+			certificateRecord.Output = id
 		}
 
 		return nil
 	}); err != nil {
-		a.AddOutput(ctx, xtime.BeijingTimeStr(), a.node.Name, "保存申请记录失败", err.Error())
+		a.AddOutput(ctx, a.node.Name, "保存申请记录失败", err.Error())
 		return err
 	}
 
-	a.AddOutput(ctx, xtime.BeijingTimeStr(), a.node.Name, "保存申请记录成功")
+	a.AddOutput(ctx, a.node.Name, "保存申请记录成功")
 
 	return nil
 }

internal/workflow/node-processor/deploy_node.go

@@ -1 +1,104 @@
-package nodeprocessor
+package nodeprocessor
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/usual2970/certimate/internal/deployer"
+	"github.com/usual2970/certimate/internal/domain"
+	"github.com/usual2970/certimate/internal/repository"
+)
+
+type deployNode struct {
+	node       *domain.WorkflowNode
+	outputRepo WorkflowOutputRepository
+	*Logger
+}
+
+func NewDeployNode(node *domain.WorkflowNode) *deployNode {
+	return &deployNode{
+		node:       node,
+		Logger:     NewLogger(node),
+		outputRepo: repository.NewWorkflowOutputRepository(),
+	}
+}
+
+func (d *deployNode) Run(ctx context.Context) error {
+	d.AddOutput(ctx, d.node.Name, "开始执行")
+	// 检查是否部署过（部署过则直接返回，和 v0.2 暂时保持一致）
+	output, err := d.outputRepo.Get(ctx, d.node.Id)
+	if err != nil && !domain.IsRecordNotFound(err) {
+		d.AddOutput(ctx, d.node.Name, "查询部署记录失败", err.Error())
+		return err
+	}
+	if output != nil && output.Succeed {
+		d.AddOutput(ctx, d.node.Name, "已部署过")
+		return nil
+	}
+	// 获取部署对象
+	// 获取证书
+	certSource := d.node.GetConfigString("certificate")
+
+	certSourceSlice := strings.Split(certSource, "#")
+	if len(certSourceSlice) != 2 {
+		d.AddOutput(ctx, d.node.Name, "证书来源配置错误", certSource)
+		return fmt.Errorf("证书来源配置错误: %s", certSource)
+	}
+
+	cert, err := d.outputRepo.GetCertificate(ctx, certSourceSlice[0])
+	if err != nil {
+		d.AddOutput(ctx, d.node.Name, "获取证书失败", err.Error())
+		return err
+	}
+
+	accessRepo := repository.NewAccessRepository()
+	access, err := accessRepo.GetById(context.Background(), d.node.GetConfigString("access"))
+	if err != nil {
+		d.AddOutput(ctx, d.node.Name, "获取授权配置失败", err.Error())
+		return err
+	}
+	option := &deployer.DeployerOption{
+		DomainId:     d.node.Id,
+		Domain:       cert.SAN,
+		Access:       access.Config,
+		AccessRecord: access,
+		DeployConfig: domain.DeployConfig{
+			Id:     d.node.Id,
+			Access: access.Id,
+			Type:   d.node.GetConfigString("providerType"),
+			Config: d.node.Config,
+		},
+	}
+
+	deploy, err := deployer.GetWithTypeAndOption(d.node.GetConfigString("providerType"), option)
+	if err != nil {
+		d.AddOutput(ctx, d.node.Name, "获取部署对象失败", err.Error())
+		return err
+	}
+
+	// 部署
+	if err := deploy.Deploy(ctx); err != nil {
+		d.AddOutput(ctx, d.node.Name, "部署失败", err.Error())
+		return err
+	}
+
+	d.AddOutput(ctx, d.node.Name, "部署成功")
+
+	// 记录部署结果
+	output = &domain.WorkflowOutput{
+		Workflow: GetWorkflowId(ctx),
+		NodeId:   d.node.Id,
+		Node:     d.node,
+		Succeed:  true,
+	}
+
+	if err := d.outputRepo.Save(ctx, output, nil, nil); err != nil {
+		d.AddOutput(ctx, d.node.Name, "保存部署记录失败", err.Error())
+		return err
+	}
+
+	d.AddOutput(ctx, d.node.Name, "保存部署记录成功")
+
+	return nil
+}

internal/workflow/node-processor/notify_node.go

@@ -1 +1,55 @@
-package nodeprocessor
+package nodeprocessor
+
+import (
+	"context"
+
+	"github.com/usual2970/certimate/internal/domain"
+	"github.com/usual2970/certimate/internal/notify"
+	"github.com/usual2970/certimate/internal/repository"
+)
+
+type SettingRepository interface {
+	GetByName(ctx context.Context, name string) (*domain.Setting, error)
+}
+type notifyNode struct {
+	node        *domain.WorkflowNode
+	settingRepo SettingRepository
+	*Logger
+}
+
+func NewNotifyNode(node *domain.WorkflowNode) *notifyNode {
+	return &notifyNode{
+		node:        node,
+		Logger:      NewLogger(node),
+		settingRepo: repository.NewSettingRepository(),
+	}
+}
+
+func (n *notifyNode) Run(ctx context.Context) error {
+	n.AddOutput(ctx, n.node.Name, "开始执行")
+
+	// 获取通知配置
+	setting, err := n.settingRepo.GetByName(ctx, "notifyChannels")
+	if err != nil {
+		n.AddOutput(ctx, n.node.Name, "获取通知配置失败", err.Error())
+		return err
+	}
+
+	channelConfig, err := setting.GetChannelContent(n.node.GetConfigString("channel"))
+	if err != nil {
+		n.AddOutput(ctx, n.node.Name, "获取通知渠道配置失败", err.Error())
+		return err
+	}
+
+	if err := notify.SendToChannel(n.node.GetConfigString("title"),
+		n.node.GetConfigString("content"),
+		n.node.GetConfigString("channel"),
+		channelConfig,
+	); err != nil {
+		n.AddOutput(ctx, n.node.Name, "发送通知失败", err.Error())
+		return err
+	}
+
+	n.AddOutput(ctx, n.node.Name, "发送通知成功")
+	return nil
+}

internal/workflow/node-processor/processor.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 
 	"github.com/usual2970/certimate/internal/domain"
+	"github.com/usual2970/certimate/internal/utils/xtime"
 )
 
 type RunLog struct {
@@ -23,7 +24,7 @@ type RunLogOutput struct {
 type NodeProcessor interface {
 	Run(ctx context.Context) error
 	Log(ctx context.Context) *RunLog
-	AddOutput(ctx context.Context, time, title, content string, err ...string)
+	AddOutput(ctx context.Context, title, content string, err ...string)
 }
 
 type Logger struct {
@@ -43,9 +44,9 @@ func (l *Logger) Log(ctx context.Context) *RunLog {
 	return l.log
 }
 
-func (l *Logger) AddOutput(ctx context.Context, time, title, content string, err ...string) {
+func (l *Logger) AddOutput(ctx context.Context, title, content string, err ...string) {
 	output := RunLogOutput{
-		Time:    time,
+		Time:    xtime.BeijingTimeStr(),
 		Title:   title,
 		Content: content,
 	}
@@ -64,6 +65,10 @@ func GetProcessor(node *domain.WorkflowNode) (NodeProcessor, error) {
 		return NewConditionNode(node), nil
 	case domain.WorkflowNodeTypeApply:
 		return NewApplyNode(node), nil
+	case domain.WorkflowNodeTypeDeploy:
+		return NewDeployNode(node), nil
+	case domain.WorkflowNodeTypeNotify:
+		return NewNotifyNode(node), nil
 	}
 	return nil, errors.New("not implemented")
 }

internal/workflow/node-processor/workflow_processor.go

@@ -50,6 +50,7 @@ func (w *workflowProcessor) runNode(ctx context.Context, node *domain.WorkflowNo
 				return err
 			}
 		}
+		current = current.Next
 
 	}
 	return nil

internal/workflow/service.go

@@ -47,6 +47,7 @@ func (s *WorkflowService) Run(ctx context.Context, req *domain.WorkflowRunReq) e
 	}
 
 	// 保存执行日志
+	
 
 	return nil
 }

ui/src/domain/version.ts

@@ -1 +1 @@
-export const version = "Certimate v0.2.20";
+export const version = "Certimate v0.2.21";