GuanM/certimate
1
0
Fork 0
mirror of https://github.com/usual2970/certimate.git synced 2025-06-07 21:19:51 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: workflow run status & time

Browse Source
This commit is contained in:
Fu Diwei 2025-01-04 22:07:01 +08:00
parent b686579acc
commit 3b9a7fe805
29 changed files with 505 additions and 181 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
internal/app/app.go
View File

@ -1,6 +1,7 @@
package app
import (
"log/slog"
"sync"
"github.com/pocketbase/pocketbase"
@ -19,3 +20,7 @@ func GetApp() *pocketbase.PocketBase {
return instance
}
func GetLogger() *slog.Logger {
return GetApp().Logger()
}

8
internal/certificate/service.go
View File

@ -37,21 +37,21 @@ func (s *certificateService) InitSchedule(ctx context.Context) error {
err := scheduler.Add("certificate", "0 0 * * *", func() {
certs, err := s.repo.ListExpireSoon(context.Background())
if err != nil {
app.GetApp().Logger().Error("failed to get expire soon certificate", "err", err)
app.GetLogger().Error("failed to get expire soon certificate", "err", err)
return
}
msg := buildMsg(certs)
// TODO: 空指针 Bug
if err := notify.SendToAllChannels(msg.Subject, msg.Message); err != nil {
app.GetApp().Logger().Error("failed to send expire soon certificate", "err", err)
app.GetLogger().Error("failed to send expire soon certificate", "err", err)
}
})
if err != nil {
app.GetApp().Logger().Error("failed to add schedule", "err", err)
app.GetLogger().Error("failed to add schedule", "err", err)
return err
}
scheduler.Start()
app.GetApp().Logger().Info("certificate schedule started")
app.GetLogger().Info("certificate schedule started")
return nil
}

36
internal/domain/certificate.go
View File

@ -2,25 +2,25 @@ package domain
import "time"
type Certificate struct {
Meta
Source string `json:"source" db:"source"`
SubjectAltNames string `json:"subjectAltNames" db:"subjectAltNames"`
Certificate string `json:"certificate" db:"certificate"`
PrivateKey string `json:"privateKey" db:"privateKey"`
IssuerCertificate string `json:"issuerCertificate" db:"issuerCertificate"`
EffectAt time.Time `json:"effectAt" db:"effectAt"`
ExpireAt time.Time `json:"expireAt" db:"expireAt"`
AcmeCertUrl string `json:"acmeCertUrl" db:"acmeCertUrl"`
AcmeCertStableUrl string `json:"acmeCertStableUrl" db:"acmeCertStableUrl"`
WorkflowId string `json:"workflowId" db:"workflowId"`
WorkflowNodeId string `json:"workflowNodeId" db:"workflowNodeId"`
WorkflowOutputId string `json:"workflowOutputId" db:"workflowOutputId"`
}
type CertificateSourceType string
const (
CERTIFICATE_SOURCE_WORKFLOW = CertificateSourceType("workflow")
CERTIFICATE_SOURCE_UPLOAD = CertificateSourceType("upload")
CertificateSourceTypeWorkflow = CertificateSourceType("workflow")
CertificateSourceTypeUpload = CertificateSourceType("upload")
)
type Certificate struct {
Meta
Source CertificateSourceType `json:"source" db:"source"`
SubjectAltNames string `json:"subjectAltNames" db:"subjectAltNames"`
Certificate string `json:"certificate" db:"certificate"`
PrivateKey string `json:"privateKey" db:"privateKey"`
IssuerCertificate string `json:"issuerCertificate" db:"issuerCertificate"`
EffectAt time.Time `json:"effectAt" db:"effectAt"`
ExpireAt time.Time `json:"expireAt" db:"expireAt"`
AcmeCertUrl string `json:"acmeCertUrl" db:"acmeCertUrl"`
AcmeCertStableUrl string `json:"acmeCertStableUrl" db:"acmeCertStableUrl"`
WorkflowId string `json:"workflowId" db:"workflowId"`
WorkflowNodeId string `json:"workflowNodeId" db:"workflowNodeId"`
WorkflowOutputId string `json:"workflowOutputId" db:"workflowOutputId"`
}

63
internal/domain/workflow.go
View File

@ -1,48 +1,58 @@
package domain
import (
"time"
"github.com/usual2970/certimate/internal/pkg/utils/maps"
)
const (
WorkflowNodeTypeStart = "start"
WorkflowNodeTypeEnd = "end"
WorkflowNodeTypeApply = "apply"
WorkflowNodeTypeDeploy = "deploy"
WorkflowNodeTypeNotify = "notify"
WorkflowNodeTypeBranch = "branch"
WorkflowNodeTypeCondition = "condition"
)
type WorkflowNodeType string
const (
WorkflowTriggerAuto = "auto"
WorkflowTriggerManual = "manual"
WorkflowNodeTypeStart = WorkflowNodeType("start")
WorkflowNodeTypeEnd = WorkflowNodeType("end")
WorkflowNodeTypeApply = WorkflowNodeType("apply")
WorkflowNodeTypeDeploy = WorkflowNodeType("deploy")
WorkflowNodeTypeNotify = WorkflowNodeType("notify")
WorkflowNodeTypeBranch = WorkflowNodeType("branch")
WorkflowNodeTypeCondition = WorkflowNodeType("condition")
)
type WorkflowTriggerType string
const (
WorkflowTriggerTypeAuto = WorkflowTriggerType("auto")
WorkflowTriggerTypeManual = WorkflowTriggerType("manual")
)
type Workflow struct {
Meta
Name string `json:"name"`
Description string `json:"description"`
Trigger string `json:"trigger"`
TriggerCron string `json:"triggerCron"`
Enabled bool `json:"enabled"`
Content *WorkflowNode `json:"content"`
Draft *WorkflowNode `json:"draft"`
HasDraft bool `json:"hasDraft"`
Name string `json:"name" db:"name"`
Description string `json:"description" db:"description"`
Trigger WorkflowTriggerType `json:"trigger" db:"trigger"`
TriggerCron string `json:"triggerCron" db:"triggerCron"`
Enabled bool `json:"enabled" db:"enabled"`
Content *WorkflowNode `json:"content" db:"content"`
Draft *WorkflowNode `json:"draft" db:"draft"`
HasDraft bool `json:"hasDraft" db:"hasDraft"`
LastRunId string `json:"lastRunId" db:"lastRunId"`
LastRunStatus WorkflowRunStatusType `json:"lastRunStatus" db:"lastRunStatus"`
LastRunTime time.Time `json:"lastRunTime" db:"lastRunTime"`
}
type WorkflowNode struct {
Id string `json:"id"`
Name string `json:"name"`
Next *WorkflowNode `json:"next"`
Id string `json:"id"`
Type WorkflowNodeType `json:"type"`
Name string `json:"name"`
Config map[string]any `json:"config"`
Inputs []WorkflowNodeIO `json:"inputs"`
Outputs []WorkflowNodeIO `json:"outputs"`
Validated bool `json:"validated"`
Type string `json:"type"`
Next *WorkflowNode `json:"next"`
Branches []WorkflowNode `json:"branches"`
Validated bool `json:"validated"`
}
func (n *WorkflowNode) GetConfigString(key string) string {
@ -76,5 +86,6 @@ type WorkflowNodeIOValueSelector struct {
}
type WorkflowRunReq struct {
Id string `json:"id"`
WorkflowId string `json:"workflowId"`
Trigger WorkflowTriggerType `json:"trigger"`
}

23
internal/domain/workflow_run.go
View File

@ -2,15 +2,24 @@ package domain
import "time"
type WorkflowRunStatusType string
const (
WorkflowRunStatusTypePending WorkflowRunStatusType = "pending"
WorkflowRunStatusTypeRunning WorkflowRunStatusType = "running"
WorkflowRunStatusTypeSucceeded WorkflowRunStatusType = "succeeded"
WorkflowRunStatusTypeFailed WorkflowRunStatusType = "failed"
)
type WorkflowRun struct {
Meta
WorkflowId string `json:"workflowId" db:"workflowId"`
Trigger string `json:"trigger" db:"trigger"`
StartedAt time.Time `json:"startedAt" db:"startedAt"`
CompletedAt time.Time `json:"completedAt" db:"completedAt"`
Logs []WorkflowRunLog `json:"logs" db:"logs"`
Succeeded bool `json:"succeeded" db:"succeeded"`
Error string `json:"error" db:"error"`
WorkflowId string `json:"workflowId" db:"workflowId"`
Status WorkflowRunStatusType `json:"status" db:"status"`
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"`
Error string `json:"error" db:"error"`
}
type WorkflowRunLog struct {

8
internal/repository/access.go
View File

@ -24,16 +24,16 @@ func (a *AccessRepository) GetById(ctx context.Context, id string) (*domain.Acce
return nil, err
}
rs := &domain.Access{
access := &domain.Access{
Meta: domain.Meta{
Id: record.GetId(),
CreatedAt: record.GetTime("created"),
UpdatedAt: record.GetTime("updated"),
CreatedAt: record.GetCreated().Time(),
UpdatedAt: record.GetUpdated().Time(),
},
Name: record.GetString("name"),
Provider: record.GetString("provider"),
Config: record.GetString("config"),
Usage: record.GetString("usage"),
}
return rs, nil
return access, nil
}

6
internal/repository/acme_account.go
View File

@ -48,9 +48,9 @@ func (r *AcmeAccountRepository) GetByCAAndEmail(ca, email string) (*domain.AcmeA
return &domain.AcmeAccount{
Meta: domain.Meta{
Id: record.GetString("id"),
CreatedAt: record.GetTime("created"),
UpdatedAt: record.GetTime("updated"),
Id: record.GetId(),
CreatedAt: record.GetCreated().Time(),
UpdatedAt: record.GetUpdated().Time(),
},
CA: record.GetString("ca"),
Email: record.GetString("email"),

2
internal/repository/certificate.go
View File

@ -16,7 +16,7 @@ func NewCertificateRepository() *CertificateRepository {
func (c *CertificateRepository) ListExpireSoon(ctx context.Context) ([]domain.Certificate, error) {
rs := []domain.Certificate{}
if err := app.GetApp().Dao().DB().
NewQuery("select * from certificate where expireAt > datetime('now') and expireAt < datetime('now', '+20 days')").
NewQuery("SELECT * FROM certificate WHERE expireAt > DATETIME('now') AND expireAt < DATETIME('now', '+20 days')").
All(&rs); err != nil {
return nil, err
}

12
internal/repository/settings.go
View File

@ -15,19 +15,19 @@ func NewSettingsRepository() *SettingsRepository {
}
func (s *SettingsRepository) GetByName(ctx context.Context, name string) (*domain.Settings, error) {
resp, err := app.GetApp().Dao().FindFirstRecordByFilter("settings", "name={:name}", dbx.Params{"name": name})
record, err := app.GetApp().Dao().FindFirstRecordByFilter("settings", "name={:name}", dbx.Params{"name": name})
if err != nil {
return nil, err
}
rs := &domain.Settings{
Meta: domain.Meta{
Id: resp.GetString("id"),
CreatedAt: resp.GetTime("created"),
UpdatedAt: resp.GetTime("updated"),
Id: record.GetId(),
CreatedAt: record.GetCreated().Time(),
UpdatedAt: record.GetUpdated().Time(),
},
Name: resp.GetString("name"),
Content: resp.GetString("content"),
Name: record.GetString("name"),
Content: record.GetString("content"),
}
return rs, nil

10
internal/repository/statistics.go
View File

@ -19,7 +19,7 @@ func (r *StatisticsRepository) Get(ctx context.Context) (*domain.Statistics, err
certTotal := struct {
Total int `db:"total"`
}{}
if err := app.GetApp().Dao().DB().NewQuery("select count(*) as total from certificate").One(&certTotal); err != nil {
if err := app.GetApp().Dao().DB().NewQuery("SELECT COUNT(*) AS total FROM certificate").One(&certTotal); err != nil {
return nil, err
}
rs.CertificateTotal = certTotal.Total
@ -29,7 +29,7 @@ func (r *StatisticsRepository) Get(ctx context.Context) (*domain.Statistics, err
Total int `db:"total"`
}{}
if err := app.GetApp().Dao().DB().
NewQuery("select count(*) as total from certificate where expireAt > datetime('now') and expireAt < datetime('now', '+20 days')").
NewQuery("SELECT COUNT(*) AS total FROM certificate WHERE expireAt > DATETIME('now') and expireAt < DATETIME('now', '+20 days')").
One(&certExpireSoonTotal); err != nil {
return nil, err
}
@ -40,7 +40,7 @@ func (r *StatisticsRepository) Get(ctx context.Context) (*domain.Statistics, err
Total int `db:"total"`
}{}
if err := app.GetApp().Dao().DB().
NewQuery("select count(*) as total from certificate where expireAt < datetime('now')").
NewQuery("SELECT COUNT(*) AS total FROM certificate WHERE expireAt < DATETIME('now')").
One(&certExpiredTotal); err != nil {
return nil, err
}
@ -50,7 +50,7 @@ func (r *StatisticsRepository) Get(ctx context.Context) (*domain.Statistics, err
workflowTotal := struct {
Total int `db:"total"`
}{}
if err := app.GetApp().Dao().DB().NewQuery("select count(*) as total from workflow").One(&workflowTotal); err != nil {
if err := app.GetApp().Dao().DB().NewQuery("SELECT COUNT(*) AS total FROM workflow").One(&workflowTotal); err != nil {
return nil, err
}
rs.WorkflowTotal = workflowTotal.Total
@ -59,7 +59,7 @@ func (r *StatisticsRepository) Get(ctx context.Context) (*domain.Statistics, err
workflowEnabledTotal := struct {
Total int `db:"total"`
}{}
if err := app.GetApp().Dao().DB().NewQuery("select count(*) as total from workflow where enabled is TRUE").One(&workflowEnabledTotal); err != nil {
if err := app.GetApp().Dao().DB().NewQuery("SELECT COUNT(*) AS total FROM workflow WHERE enabled IS TRUE").One(&workflowEnabledTotal); err != nil {
return nil, err
}
rs.WorkflowEnabled = workflowEnabledTotal.Total

73
internal/repository/workflow.go
View File

@ -6,6 +6,7 @@ import (
"errors"
"github.com/pocketbase/dbx"
"github.com/pocketbase/pocketbase/daos"
"github.com/pocketbase/pocketbase/models"
"github.com/usual2970/certimate/internal/app"
"github.com/usual2970/certimate/internal/domain"
@ -21,11 +22,12 @@ func (w *WorkflowRepository) ListEnabledAuto(ctx context.Context) ([]domain.Work
records, err := app.GetApp().Dao().FindRecordsByFilter(
"workflow",
"enabled={:enabled} && trigger={:trigger}",
"-created", 1000, 0, dbx.Params{"enabled": true, "trigger": domain.WorkflowTriggerAuto},
"-created", 1000, 0, dbx.Params{"enabled": true, "trigger": domain.WorkflowTriggerTypeAuto},
)
if err != nil {
return nil, err
}
rs := make([]domain.Workflow, 0)
for _, record := range records {
workflow, err := record2Workflow(record)
@ -34,25 +36,50 @@ func (w *WorkflowRepository) ListEnabledAuto(ctx context.Context) ([]domain.Work
}
rs = append(rs, *workflow)
}
return rs, nil
}
func (w *WorkflowRepository) SaveRunLog(ctx context.Context, log *domain.WorkflowRun) error {
func (w *WorkflowRepository) SaveRun(ctx context.Context, run *domain.WorkflowRun) error {
collection, err := app.GetApp().Dao().FindCollectionByNameOrId("workflow_run")
if err != nil {
return err
}
record := models.NewRecord(collection)
record.Set("workflowId", log.WorkflowId)
record.Set("trigger", log.Trigger)
record.Set("startedAt", log.StartedAt)
record.Set("completedAt", log.CompletedAt)
record.Set("logs", log.Logs)
record.Set("succeeded", log.Succeeded)
record.Set("error", log.Error)
err = app.GetApp().Dao().RunInTransaction(func(txDao *daos.Dao) error {
record := models.NewRecord(collection)
record.Set("workflowId", run.WorkflowId)
record.Set("trigger", string(run.Trigger))
record.Set("status", string(run.Status))
record.Set("startedAt", run.StartedAt)
record.Set("endedAt", run.EndedAt)
record.Set("logs", run.Logs)
record.Set("error", run.Error)
err = txDao.SaveRecord(record)
if err != nil {
return err
}
return app.GetApp().Dao().SaveRecord(record)
_, err = txDao.DB().Update(
"workflow",
dbx.Params{
"lastRunId": record.GetId(),
"lastRunStatus": record.GetString("status"),
"lastRunTime": record.GetString("startedAt"),
},
dbx.NewExp("id={:id}", dbx.Params{"id": run.WorkflowId}),
).Execute()
if err != nil {
return err
}
return nil
})
if err != nil {
return err
}
return nil
}
func (w *WorkflowRepository) Get(ctx context.Context, id string) (*domain.Workflow, error) {
@ -81,18 +108,20 @@ func record2Workflow(record *models.Record) (*domain.Workflow, error) {
workflow := &domain.Workflow{
Meta: domain.Meta{
Id: record.GetId(),
CreatedAt: record.GetTime("created"),
UpdatedAt: record.GetTime("updated"),
CreatedAt: record.GetCreated().Time(),
UpdatedAt: record.GetUpdated().Time(),
},
Name: record.GetString("name"),
Description: record.GetString("description"),
Trigger: record.GetString("trigger"),
TriggerCron: record.GetString("triggerCron"),
Enabled: record.GetBool("enabled"),
Content: content,
Draft: draft,
HasDraft: record.GetBool("hasDraft"),
Name: record.GetString("name"),
Description: record.GetString("description"),
Trigger: domain.WorkflowTriggerType(record.GetString("trigger")),
TriggerCron: record.GetString("triggerCron"),
Enabled: record.GetBool("enabled"),
Content: content,
Draft: draft,
HasDraft: record.GetBool("hasDraft"),
LastRunId: record.GetString("lastRunId"),
LastRunStatus: domain.WorkflowRunStatusType(record.GetString("lastRunStatus")),
LastRunTime: record.GetTime("lastRunTime"),
}
return workflow, nil
}

10
internal/repository/workflow_output.go
View File

@ -73,16 +73,16 @@ func (w *WorkflowOutputRepository) GetCertificate(ctx context.Context, nodeId st
rs := &domain.Certificate{
Meta: domain.Meta{
Id: record.GetId(),
CreatedAt: record.GetDateTime("created").Time(),
UpdatedAt: record.GetDateTime("updated").Time(),
CreatedAt: record.GetCreated().Time(),
UpdatedAt: record.GetUpdated().Time(),
},
Source: record.GetString("source"),
Source: domain.CertificateSourceType(record.GetString("source")),
SubjectAltNames: record.GetString("subjectAltNames"),
Certificate: record.GetString("certificate"),
PrivateKey: record.GetString("privateKey"),
IssuerCertificate: record.GetString("issuerCertificate"),
EffectAt: record.GetDateTime("effectAt").Time(),
ExpireAt: record.GetDateTime("expireAt").Time(),
EffectAt: record.GetTime("effectAt"),
ExpireAt: record.GetTime("expireAt"),
AcmeCertUrl: record.GetString("acmeCertUrl"),
AcmeCertStableUrl: record.GetString("acmeCertStableUrl"),
WorkflowId: record.GetString("workflowId"),

15
internal/workflow/event.go
View File

@ -44,27 +44,28 @@ func update(ctx context.Context, record *models.Record) error {
// 是不是自动
// 是不是 enabled
id := record.Id
workflowId := record.Id
enabled := record.GetBool("enabled")
trigger := record.GetString("trigger")
scheduler := app.GetScheduler()
if !enabled || trigger == domain.WorkflowTriggerManual {
scheduler.Remove(id)
if !enabled || trigger == string(domain.WorkflowTriggerTypeManual) {
scheduler.Remove(workflowId)
scheduler.Start()
return nil
}
err := scheduler.Add(id, record.GetString("triggerCron"), func() {
err := scheduler.Add(workflowId, record.GetString("triggerCron"), func() {
NewWorkflowService(repository.NewWorkflowRepository()).Run(ctx, &domain.WorkflowRunReq{
Id: id,
WorkflowId: workflowId,
Trigger: domain.WorkflowTriggerTypeAuto,
})
})
if err != nil {
app.GetApp().Logger().Error("add cron job failed", "err", err)
app.GetLogger().Error("add cron job failed", "err", err)
return fmt.Errorf("add cron job failed: %w", err)
}
app.GetApp().Logger().Error("add cron job failed", "subjectAltNames", record.GetString("subjectAltNames"))
app.GetLogger().Error("add cron job failed", "subjectAltNames", record.GetString("subjectAltNames"))
scheduler.Start()
return nil

2
internal/workflow/node-processor/apply_node.go
View File

@ -98,7 +98,7 @@ func (a *applyNode) Run(ctx context.Context) error {
}
certificateRecord := &domain.Certificate{
Source: string(domain.CERTIFICATE_SOURCE_WORKFLOW),
Source: domain.CertificateSourceTypeWorkflow,
SubjectAltNames: strings.Join(certX509.DNSNames, ";"),
Certificate: certificate.Certificate,
PrivateKey: certificate.PrivateKey,

72
internal/workflow/service.go
View File

@ -3,6 +3,7 @@ package workflow
import (
"context"
"fmt"
"time"
"github.com/usual2970/certimate/internal/app"
"github.com/usual2970/certimate/internal/domain"
@ -11,7 +12,7 @@ import (
type WorkflowRepository interface {
Get(ctx context.Context, id string) (*domain.Workflow, error)
SaveRunLog(ctx context.Context, log *domain.WorkflowRun) error
SaveRun(ctx context.Context, run *domain.WorkflowRun) error
ListEnabledAuto(ctx context.Context) ([]domain.Workflow, error)
}
@ -31,71 +32,80 @@ func (s *WorkflowService) InitSchedule(ctx context.Context) error {
if err != nil {
return err
}
scheduler := app.GetScheduler()
for _, workflow := range workflows {
err := scheduler.Add(workflow.Id, workflow.TriggerCron, func() {
s.Run(ctx, &domain.WorkflowRunReq{
Id: workflow.Id,
WorkflowId: workflow.Id,
Trigger: domain.WorkflowTriggerTypeAuto,
})
})
if err != nil {
app.GetApp().Logger().Error("failed to add schedule", "err", err)
app.GetLogger().Error("failed to add schedule", "err", err)
return err
}
}
scheduler.Start()
app.GetApp().Logger().Info("workflow schedule started")
app.GetLogger().Info("workflow schedule started")
return nil
}
func (s *WorkflowService) Run(ctx context.Context, req *domain.WorkflowRunReq) error {
func (s *WorkflowService) Run(ctx context.Context, options *domain.WorkflowRunReq) error {
// 查询
if req.Id == "" {
if options.WorkflowId == "" {
return domain.ErrInvalidParams
}
workflow, err := s.repo.Get(ctx, req.Id)
workflow, err := s.repo.Get(ctx, options.WorkflowId)
if err != nil {
app.GetApp().Logger().Error("failed to get workflow", "id", req.Id, "err", err)
app.GetLogger().Error("failed to get workflow", "id", options.WorkflowId, "err", err)
return err
}
// 执行
if !workflow.Enabled {
app.GetApp().Logger().Error("workflow is disabled", "id", req.Id)
app.GetLogger().Error("workflow is disabled", "id", options.WorkflowId)
return fmt.Errorf("workflow is disabled")
}
// 执行
run := &domain.WorkflowRun{
WorkflowId: workflow.Id,
Status: domain.WorkflowRunStatusTypeRunning,
Trigger: options.Trigger,
StartedAt: time.Now(),
EndedAt: time.Now(),
}
processor := nodeprocessor.NewWorkflowProcessor(workflow)
if err := processor.Run(ctx); err != nil {
log := &domain.WorkflowRun{
WorkflowId: workflow.Id,
Logs: processor.Log(ctx),
Succeeded: false,
Error: err.Error(),
}
if err := s.repo.SaveRunLog(ctx, log); err != nil {
app.GetApp().Logger().Error("failed to save run log", "err", err)
run.Status = domain.WorkflowRunStatusTypeFailed
run.EndedAt = time.Now()
run.Logs = processor.Log(ctx)
run.Error = err.Error()
if err := s.repo.SaveRun(ctx, run); err != nil {
app.GetLogger().Error("failed to save workflow run", "err", err)
}
return fmt.Errorf("failed to run workflow: %w", err)
}
// 保存执行日志
logs := processor.Log(ctx)
runLogs := domain.WorkflowRunLogs(logs)
runErr := runLogs.FirstError()
succeed := true
if runErr != "" {
succeed = false
runStatus := domain.WorkflowRunStatusTypeSucceeded
runError := domain.WorkflowRunLogs(logs).FirstError()
if runError != "" {
runStatus = domain.WorkflowRunStatusTypeFailed
}
log := &domain.WorkflowRun{
WorkflowId: workflow.Id,
Logs: processor.Log(ctx),
Error: runErr,
Succeeded: succeed,
}
if err := s.repo.SaveRunLog(ctx, log); err != nil {
app.GetApp().Logger().Error("failed to save run log", "err", err)
run.Status = runStatus
run.EndedAt = time.Now()
run.Logs = processor.Log(ctx)
run.Error = runError
if err := s.repo.SaveRun(ctx, run); err != nil {
app.GetLogger().Error("failed to save workflow run", "err", err)
return err
}

106
migrations/1735981441_updated_workflow.go Normal file
View File

@ -0,0 +1,106 @@
package migrations
import (
"encoding/json"
"github.com/pocketbase/dbx"
"github.com/pocketbase/pocketbase/daos"
m "github.com/pocketbase/pocketbase/migrations"
"github.com/pocketbase/pocketbase/models/schema"
)
func init() {
m.Register(func(db dbx.Builder) error {
dao := daos.New(db);
collection, err := dao.FindCollectionByNameOrId("tovyif5ax6j62ur")
if err != nil {
return err
}
// add
new_lastRunId := &schema.SchemaField{}
if err := json.Unmarshal([]byte(`{
"system": false,
"id": "a23wkj9x",
"name": "lastRunId",
"type": "relation",
"required": false,
"presentable": false,
"unique": false,
"options": {
"collectionId": "qjp8lygssgwyqyz",
"cascadeDelete": false,
"minSelect": null,
"maxSelect": 1,
"displayFields": null
}
}`), new_lastRunId); err != nil {
return err
}
collection.Schema.AddField(new_lastRunId)
// add
new_lastRunStatus := &schema.SchemaField{}
if err := json.Unmarshal([]byte(`{
"system": false,
"id": "zivdxh23",
"name": "lastRunStatus",
"type": "select",
"required": false,
"presentable": false,
"unique": false,
"options": {
"maxSelect": 1,
"values": [
"pending",
"running",
"succeeded",
"failed"
]
}
}`), new_lastRunStatus); err != nil {
return err
}
collection.Schema.AddField(new_lastRunStatus)
// add
new_lastRunTime := &schema.SchemaField{}
if err := json.Unmarshal([]byte(`{
"system": false,
"id": "u9bosu36",
"name": "lastRunTime",
"type": "date",
"required": false,
"presentable": false,
"unique": false,
"options": {
"min": "",
"max": ""
}
}`), new_lastRunTime); err != nil {
return err
}
collection.Schema.AddField(new_lastRunTime)
return dao.SaveCollection(collection)
}, func(db dbx.Builder) error {
dao := daos.New(db);
collection, err := dao.FindCollectionByNameOrId("tovyif5ax6j62ur")
if err != nil {
return err
}
// remove
collection.Schema.RemoveField("a23wkj9x")
// remove
collection.Schema.RemoveField("zivdxh23")
// remove
collection.Schema.RemoveField("u9bosu36")
return dao.SaveCollection(collection)
})
}

78
migrations/1735981515_updated_workflow_run.go Normal file
View File

@ -0,0 +1,78 @@
package migrations
import (
"encoding/json"
"github.com/pocketbase/dbx"
"github.com/pocketbase/pocketbase/daos"
m "github.com/pocketbase/pocketbase/migrations"
"github.com/pocketbase/pocketbase/models/schema"
)
func init() {
m.Register(func(db dbx.Builder) error {
dao := daos.New(db);
collection, err := dao.FindCollectionByNameOrId("qjp8lygssgwyqyz")
if err != nil {
return err
}
// remove
collection.Schema.RemoveField("cht6kqw9")
// add
new_status := &schema.SchemaField{}
if err := json.Unmarshal([]byte(`{
"system": false,
"id": "qldmh0tw",
"name": "status",
"type": "select",
"required": false,
"presentable": false,
"unique": false,
"options": {
"maxSelect": 1,
"values": [
"pending",
"running",
"succeeded",
"failed"
]
}
}`), new_status); err != nil {
return err
}
collection.Schema.AddField(new_status)
return dao.SaveCollection(collection)
}, func(db dbx.Builder) error {
dao := daos.New(db);
collection, err := dao.FindCollectionByNameOrId("qjp8lygssgwyqyz")
if err != nil {
return err
}
// add
del_succeeded := &schema.SchemaField{}
if err := json.Unmarshal([]byte(`{
"system": false,
"id": "cht6kqw9",
"name": "succeeded",
"type": "bool",
"required": false,
"presentable": false,
"unique": false,
"options": {}
}`), del_succeeded); err != nil {
return err
}
collection.Schema.AddField(del_succeeded)
// remove
collection.Schema.RemoveField("qldmh0tw")
return dao.SaveCollection(collection)
})
}

4
ui/src/api/workflow.ts
View File

@ -1,5 +1,6 @@
import { ClientResponseError } from "pocketbase";
import { WORKFLOW_TRIGGERS } from "@/domain/workflow";
import { getPocketBase } from "@/repository/pocketbase";
export const run = async (id: string) => {
@ -11,7 +12,8 @@ export const run = async (id: string) => {
"Content-Type": "application/json",
},
body: {
id,
workflowId: id,
trigger: WORKFLOW_TRIGGERS.MANUAL,
},
});

6
ui/src/components/workflow/WorkflowRunDetailDrawer.tsx
View File

@ -4,7 +4,7 @@ import { Alert, Drawer, Typography } from "antd";
import dayjs from "dayjs";
import Show from "@/components/Show";
import { type WorkflowRunModel } from "@/domain/workflowRun";
import { WORKFLOW_RUN_STATUSES, type WorkflowRunModel } from "@/domain/workflowRun";
import { useTriggerElement } from "@/hooks";
export type WorkflowRunDetailDrawerProps = {
@ -32,11 +32,11 @@ const WorkflowRunDetailDrawer = ({ data, loading, trigger, ...props }: WorkflowR
<Drawer destroyOnClose open={open} loading={loading} placement="right" title={`runlog-${data?.id}`} width={640} onClose={() => setOpen(false)}>
<Show when={!!data}>
<Show when={data!.succeeded}>
<Show when={data!.status === WORKFLOW_RUN_STATUSES.SUCCEEDED}>
<Alert showIcon type="success" message={<Typography.Text type="success">{t("workflow_run.props.status.succeeded")}</Typography.Text>} />
</Show>
<Show when={!!data!.error}>
<Show when={data!.status === WORKFLOW_RUN_STATUSES.FAILED}>
<Alert showIcon type="error" message={<Typography.Text type="danger">{t("workflow_run.props.status.failed")}</Typography.Text>} />
</Show>

68
ui/src/components/workflow/WorkflowRuns.tsx
View File

@ -2,14 +2,18 @@ import { useState } from "react";
import { useTranslation } from "react-i18next";
import {
CheckCircleOutlined as CheckCircleOutlinedIcon,
ClockCircleOutlined as ClockCircleOutlinedIcon,
CloseCircleOutlined as CloseCircleOutlinedIcon,
SelectOutlined as SelectOutlinedIcon,
SyncOutlined as SyncOutlinedIcon,
} from "@ant-design/icons";
import { useRequest } from "ahooks";
import { Button, Empty, Space, Table, type TableProps, Typography, notification, theme } from "antd";
import { Button, Empty, Table, type TableProps, Tag, notification } from "antd";
import dayjs from "dayjs";
import { ClientResponseError } from "pocketbase";
import { type WorkflowRunModel } from "@/domain/workflowRun";
import { WORKFLOW_TRIGGERS } from "@/domain/workflow";
import { WORKFLOW_RUN_STATUSES, type WorkflowRunModel } from "@/domain/workflowRun";
import { list as listWorkflowRuns } from "@/repository/workflowRun";
import { getErrMsg } from "@/utils/error";
import WorkflowRunDetailDrawer from "./WorkflowRunDetailDrawer";
@ -23,8 +27,6 @@ export type WorkflowRunsProps = {
const WorkflowRuns = ({ className, style, workflowId }: WorkflowRunsProps) => {
const { t } = useTranslation();
const { token: themeToken } = theme.useToken();
const [notificationApi, NotificationContextHolder] = notification.useNotification();
const tableColumns: TableProps<WorkflowRunModel>["columns"] = [
@ -46,45 +48,67 @@ const WorkflowRuns = ({ className, style, workflowId }: WorkflowRunsProps) => {
title: t("workflow_run.props.status"),
ellipsis: true,
render: (_, record) => {
if (record.succeeded) {
if (record.status === WORKFLOW_RUN_STATUSES.PENDING) {
return <Tag icon={<ClockCircleOutlinedIcon />}>{t("workflow_run.props.status.pending")}</Tag>;
} else if (record.status === WORKFLOW_RUN_STATUSES.RUNNING) {
return (
<Space>
<CheckCircleOutlinedIcon style={{ color: themeToken.colorSuccess }} />
<Typography.Text type="success">{t("workflow_run.props.status.succeeded")}</Typography.Text>
</Space>
<Tag icon={<SyncOutlinedIcon spin />} color="processing">
{t("workflow_run.props.status.running")}
</Tag>
);
} else {
} else if (record.status === WORKFLOW_RUN_STATUSES.SUCCEEDED) {
return (
<Space>
<CloseCircleOutlinedIcon style={{ color: themeToken.colorError }} />
<Typography.Text type="danger">{t("workflow_run.props.status.failed")}</Typography.Text>
</Space>
<Tag icon={<CheckCircleOutlinedIcon />} color="success">
{t("workflow_run.props.status.succeeded")}
</Tag>
);
} else if (record.status === WORKFLOW_RUN_STATUSES.FAILED) {
return (
<Tag icon={<CloseCircleOutlinedIcon />} color="error">
{t("workflow_run.props.status.failed")}
</Tag>
);
}
return <></>;
},
},
{
key: "trigger",
title: t("workflow_run.props.trigger"),
ellipsis: true,
render: () => {
return "TODO";
render: (_, record) => {
if (record.trigger === WORKFLOW_TRIGGERS.AUTO) {
return t("workflow_run.props.trigger.auto");
} else if (record.trigger === WORKFLOW_TRIGGERS.MANUAL) {
return t("workflow_run.props.trigger.manual");
}
return <></>;
},
},
{
key: "startedAt",
title: t("workflow_run.props.started_at"),
ellipsis: true,
render: () => {
return "TODO";
render: (_, record) => {
if (record.startedAt) {
return dayjs(record.startedAt).format("YYYY-MM-DD HH:mm:ss");
}
return <></>;
},
},
{
key: "completedAt",
title: t("workflow_run.props.completed_at"),
key: "endedAt",
title: t("workflow_run.props.ended_at"),
ellipsis: true,
render: () => {
return "TODO";
render: (_, record) => {
if (record.endedAt) {
return dayjs(record.endedAt).format("YYYY-MM-DD HH:mm:ss");
}
return <></>;
},
},
{

3
ui/src/domain/workflow.ts
View File

@ -13,6 +13,9 @@ export interface WorkflowModel extends BaseModel {
content?: WorkflowNode;
draft?: WorkflowNode;
hasDraft?: boolean;
lastRunId?: string;
lastRunStatus?: string;
lastRunTime?: string;
}
export const WORKFLOW_TRIGGERS = Object.freeze({

14
ui/src/domain/workflowRun.ts
View File

@ -1,8 +1,11 @@
export interface WorkflowRunModel extends BaseModel {
workflowId: string;
status: string;
trigger: string;
startedAt: ISO8601String;
endedAt: ISO8601String;
logs: WorkflowRunLog[];
error: string;
succeeded: boolean;
}
export type WorkflowRunLog = {
@ -18,3 +21,12 @@ export type WorkflowRunLogOutput = {
content: string;
error: string;
};
export const WORKFLOW_RUN_STATUSES = Object.freeze({
PENDING: "pending",
RUNNING: "running",
SUCCEEDED: "succeeded",
FAILED: "failed",
} as const);
export type WorkflorRunStatusType = (typeof WORKFLOW_RUN_STATUSES)[keyof typeof WORKFLOW_RUN_STATUSES];

2
ui/src/i18n/locales/en/nls.workflow.json
View File

@ -16,7 +16,7 @@
"workflow.props.trigger": "Trigger",
"workflow.props.trigger.auto": "Auto",
"workflow.props.trigger.manual": "Manual",
"workflow.props.latest_execution_status": "Latest execution status",
"workflow.props.last_run_at": "Last run at",
"workflow.props.state": "State",
"workflow.props.state.filter.enabled": "Enabled",
"workflow.props.state.filter.disabled": "Disabled",

6
ui/src/i18n/locales/en/nls.workflow.runs.json
View File

@ -1,9 +1,13 @@
{
"workflow_run.props.id": "ID",
"workflow_run.props.status": "Status",
"workflow_run.props.status.pending": "Pending",
"workflow_run.props.status.running": "Running",
"workflow_run.props.status.succeeded": "Succeeded",
"workflow_run.props.status.failed": "Failed",
"workflow_run.props.trigger": "Trigger",
"workflow_run.props.trigger.auto": "Timing",
"workflow_run.props.trigger.manual": "Manual",
"workflow_run.props.started_at": "Started at",
"workflow_run.props.completed_at": "Completed at"
"workflow_run.props.ended_at": "Ended at"
}

2
ui/src/i18n/locales/zh/nls.workflow.json
View File

@ -16,7 +16,7 @@
"workflow.props.trigger": "触发方式",
"workflow.props.trigger.auto": "自动",
"workflow.props.trigger.manual": "手动",
"workflow.props.latest_execution_status": "最近执行状态",
"workflow.props.last_run_at": "最近执行时间",
"workflow.props.state": "启用状态",
"workflow.props.state.filter.enabled": "启用",
"workflow.props.state.filter.disabled": "未启用",

8
ui/src/i18n/locales/zh/nls.workflow.runs.json
View File

@ -1,9 +1,13 @@
{
"workflow_run.props.id": "ID",
"workflow_run.props.status": "状态",
"workflow_run.props.status.pending": "等待执行",
"workflow_run.props.status.running": "执行中",
"workflow_run.props.status.succeeded": "成功",
"workflow_run.props.status.failed": "失败",
"workflow_run.props.trigger": "触发方式",
"workflow_run.props.trigger": "执行方式",
"workflow_run.props.trigger.auto": "定时执行",
"workflow_run.props.trigger.manual": "手动执行",
"workflow_run.props.started_at": "开始时间",
"workflow_run.props.completed_at": "完成时间"
"workflow_run.props.ended_at": "完成时间"
}

4
ui/src/pages/certificates/CertificateList.tsx
View File

@ -125,7 +125,7 @@ const CertificateList = () => {
}
}}
>
{record.expand?.workflowId?.name ?? `#${workflowId}`}
{record.expand?.workflowId?.name ?? <span className="font-mono">{t(`#${workflowId}`)}</span>}
</Typography.Link>
</Space>
);
@ -174,7 +174,7 @@ const CertificateList = () => {
icon={<DeleteOutlinedIcon />}
variant="text"
onClick={() => {
alert("TODO");
alert("TODO: 暂时不支持删除证书");
}}
/>
</Tooltip>

2
ui/src/pages/workflows/WorkflowDetail.tsx
View File

@ -248,7 +248,7 @@ const WorkflowDetail = () => {
<Show when={tabValue === "orchestration"}>
<div className="relative">
<div className="py-12 lg:pr-36 xl:pr-48">
<WorkflowElements disabled={isRunning} />
<WorkflowElements />
</div>
<div className="absolute right-0 top-0 z-[1]">
<Space>

38
ui/src/pages/workflows/WorkflowList.tsx
View File

@ -1,10 +1,18 @@
import { useState } from "react";
import { useTranslation } from "react-i18next";
import { useNavigate, useSearchParams } from "react-router-dom";
import { DeleteOutlined as DeleteOutlinedIcon, EditOutlined as EditOutlinedIcon, PlusOutlined as PlusOutlinedIcon } from "@ant-design/icons";
import {
CheckCircleOutlined as CheckCircleOutlinedIcon,
CloseCircleOutlined as CloseCircleOutlinedIcon,
DeleteOutlined as DeleteOutlinedIcon,
EditOutlined as EditOutlinedIcon,
PlusOutlined as PlusOutlinedIcon,
} from "@ant-design/icons";
import { PageHeader } from "@ant-design/pro-components";
import { useRequest } from "ahooks";
import {
Badge,
Button,
Divider,
Empty,
@ -26,6 +34,7 @@ import dayjs from "dayjs";
import { ClientResponseError } from "pocketbase";
import { WORKFLOW_TRIGGERS, type WorkflowModel, isAllNodesValidated } from "@/domain/workflow";
import { WORKFLOW_RUN_STATUSES } from "@/domain/workflowRun";
import { list as listWorkflow, remove as removeWorkflow, save as saveWorkflow } from "@/repository/workflow";
import { getErrMsg } from "@/utils/error";
@ -146,11 +155,28 @@ const WorkflowList = () => {
},
},
{
key: "lastExecutedAt",
title: t("workflow.props.latest_execution_status"),
render: () => {
// TODO: 最近执行状态
return <>TODO</>;
key: "lastRun",
title: t("workflow.props.last_run_at"),
render: (_, record) => {
if (record.lastRunId) {
if (record.lastRunStatus === WORKFLOW_RUN_STATUSES.SUCCEEDED) {
return (
<Space className="max-w-full" direction="vertical" size={4}>
<Badge status="success" count={<CheckCircleOutlinedIcon style={{ color: themeToken.colorSuccess }} />} />
<Typography.Text type="secondary">{dayjs(record.lastRunTime!).format("YYYY-MM-DD HH:mm:ss")}</Typography.Text>
</Space>
);
} else if (record.lastRunStatus === WORKFLOW_RUN_STATUSES.FAILED) {
return (
<Space>
<Badge status="error" count={<CloseCircleOutlinedIcon style={{ color: themeToken.colorError }} />} />
<Typography.Text>{dayjs(record.lastRunTime!).format("YYYY-MM-DD HH:mm:ss")}</Typography.Text>
</Space>
);
}
}
return <></>;
},
},
{