diff --git a/internal/certificate/service.go b/internal/certificate/service.go index c844b17f9..b27219795 100644 --- a/internal/certificate/service.go +++ b/internal/certificate/service.go @@ -17,16 +17,6 @@ import ( xcert "github.com/certimate-go/certimate/pkg/utils/cert" ) -type certificateRepository interface { - ListExpireSoon(ctx context.Context) ([]*domain.Certificate, error) - GetById(ctx context.Context, id string) (*domain.Certificate, error) - DeleteWhere(ctx context.Context, exprs ...dbx.Expression) (int, error) -} - -type settingsRepository interface { - GetByName(ctx context.Context, name string) (*domain.Settings, error) -} - type CertificateService struct { certificateRepo certificateRepository settingsRepo settingsRepository @@ -41,27 +31,8 @@ func NewCertificateService(certificateRepo certificateRepository, settingsRepo s func (s *CertificateService) InitSchedule(ctx context.Context) error { // 每日清理过期证书 - app.GetScheduler().MustAdd("certificateExpiredCleanup", "0 0 * * *", func() { - settings, err := s.settingsRepo.GetByName(ctx, "persistence") - if err != nil { - app.GetLogger().Error("failed to get persistence settings", "err", err) - return - } - - persistenceSettings, _ := settings.UnmarshalContentAsPersistence() - if persistenceSettings != nil && persistenceSettings.ExpiredCertificatesMaxDaysRetention != 0 { - ret, err := s.certificateRepo.DeleteWhere( - context.Background(), - dbx.NewExp(fmt.Sprintf("validityNotAfter 0 { - app.GetLogger().Info(fmt.Sprintf("cleanup %d expired certificates", ret)) - } - } + app.GetScheduler().MustAdd("cleanupCertificateExpired", "0 0 * * *", func() { + s.cleanupExpiredCertificates(context.Background()) }) return nil @@ -218,3 +189,29 @@ func (s *CertificateService) ValidatePrivateKey(ctx context.Context, req *dtos.C IsValid: true, }, nil } + +func (s *CertificateService) cleanupExpiredCertificates(ctx context.Context) error { + settings, err := s.settingsRepo.GetByName(ctx, "persistence") + if err != nil { + app.GetLogger().Error("failed to get persistence settings", "err", err) + return err + } + + persistenceSettings, _ := settings.UnmarshalContentAsPersistence() + if persistenceSettings != nil && persistenceSettings.ExpiredCertificatesMaxDaysRetention != 0 { + ret, err := s.certificateRepo.DeleteWhere( + context.Background(), + dbx.NewExp(fmt.Sprintf("validityNotAfter 0 { + app.GetLogger().Info(fmt.Sprintf("cleanup %d expired certificates", ret)) + } + } + + return nil +} diff --git a/internal/certificate/service_deps.go b/internal/certificate/service_deps.go new file mode 100644 index 000000000..7b22dbbad --- /dev/null +++ b/internal/certificate/service_deps.go @@ -0,0 +1,19 @@ +package certificate + +import ( + "context" + + "github.com/pocketbase/dbx" + + "github.com/certimate-go/certimate/internal/domain" +) + +type certificateRepository interface { + ListExpireSoon(ctx context.Context) ([]*domain.Certificate, error) + GetById(ctx context.Context, id string) (*domain.Certificate, error) + DeleteWhere(ctx context.Context, exprs ...dbx.Expression) (int, error) +} + +type settingsRepository interface { + GetByName(ctx context.Context, name string) (*domain.Settings, error) +} diff --git a/internal/statistics/service.go b/internal/statistics/service.go index ea041a15a..b9d2b37ac 100644 --- a/internal/statistics/service.go +++ b/internal/statistics/service.go @@ -6,10 +6,6 @@ import ( "github.com/certimate-go/certimate/internal/domain" ) -type statisticsRepository interface { - Get(ctx context.Context) (*domain.Statistics, error) -} - type StatisticsService struct { statRepo statisticsRepository } diff --git a/internal/statistics/service_deps.go b/internal/statistics/service_deps.go new file mode 100644 index 000000000..2c53fa2d5 --- /dev/null +++ b/internal/statistics/service_deps.go @@ -0,0 +1,11 @@ +package statistics + +import ( + "context" + + "github.com/certimate-go/certimate/internal/domain" +) + +type statisticsRepository interface { + Get(ctx context.Context) (*domain.Statistics, error) +} diff --git a/internal/workflow/dispatcher/invoker.go b/internal/workflow/dispatcher/invoker.go index b85440fe9..d838b4996 100644 --- a/internal/workflow/dispatcher/invoker.go +++ b/internal/workflow/dispatcher/invoker.go @@ -85,91 +85,3 @@ func (w *workflowInvoker) Invoke(ctx context.Context) error { func (w *workflowInvoker) GetLogs() domain.WorkflowLogs { return w.logs } - -func (w *workflowInvoker) processNode(ctx context.Context, graph *domain.WorkflowGraph) error { - // current := graph - // for current != nil { - // select { - // case <-ctx.Done(): - // return ctx.Err() - // default: - // } - - // if current.Type == domain.WorkflowNodeTypeCondition || current.Type == domain.WorkflowNodeTypeTryCatch { - // for _, branch := range current.Branches { - // if err := w.processNode(ctx, &branch); err != nil { - // // 并行分支的某一分支发生错误时，忽略此错误，继续执行其他分支 - // if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) { - // continue - // } - // return err - // } - // } - // } - - // var processor nodes.NodeProcessor - // var procErr error - // for { - // if current.Type != domain.WorkflowNodeTypeCondition && current.Type != domain.WorkflowNodeTypeTryCatch { - // processor, procErr = nodes.GetProcessor(current) - // if procErr != nil { - // 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.Timestamp = record.Time.UnixMilli() - // log.Level = int32(record.Level) - // 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) - // if procErr != nil { - // if current.Type != domain.WorkflowNodeTypeBranchBlock { - // processor.GetLogger().Error(procErr.Error()) - // } - // break - // } - - // nodeOutputs := processor.GetOutputs() - // if len(nodeOutputs) > 0 { - // ctx = nodes.AddNodeOutput(ctx, current.Id, nodeOutputs) - // } - // } - - // break - // } - - // // TODO: 优化可读性 - // if procErr != nil && current.Type == domain.WorkflowNodeTypeBranchBlock { - // current = nil - // procErr = nil - // return nil - // } else if procErr != nil && current.Next != nil && current.Next.Type != domain.WorkflowNodeTypeTryCatch { - // return procErr - // } else if procErr != nil && current.Next != nil && current.Next.Type == domain.WorkflowNodeTypeTryCatch { - // current = w.getBranchByType(current.Next.Branches, domain.WorkflowNodeTypeCatchBlock) - // } else if procErr == nil && current.Next != nil && current.Next.Type == domain.WorkflowNodeTypeTryCatch { - // current = w.getBranchByType(current.Next.Branches, domain.WorkflowNodeTypeTryBlock) - // } else { - // current = current.Next - // } - // } - - return nil -} diff --git a/internal/workflow/dispatcher/singleton.go b/internal/workflow/dispatcher/singleton.go index d87b57b1d..8af0921da 100644 --- a/internal/workflow/dispatcher/singleton.go +++ b/internal/workflow/dispatcher/singleton.go @@ -15,6 +15,7 @@ type workflowRepository interface { type workflowRunRepository interface { GetById(ctx context.Context, id string) (*domain.WorkflowRun, error) + Save(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error) SaveWithCascading(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error) } diff --git a/internal/workflow/service.go b/internal/workflow/service.go index 9b476ea9c..62286aedc 100644 --- a/internal/workflow/service.go +++ b/internal/workflow/service.go @@ -14,22 +14,6 @@ import ( "github.com/certimate-go/certimate/internal/workflow/dispatcher" ) -type workflowRepository interface { - ListEnabledScheduled(ctx context.Context) ([]*domain.Workflow, error) - GetById(ctx context.Context, id string) (*domain.Workflow, error) - Save(ctx context.Context, workflow *domain.Workflow) (*domain.Workflow, error) -} - -type workflowRunRepository interface { - GetById(ctx context.Context, id string) (*domain.WorkflowRun, error) - SaveWithCascading(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error) - DeleteWhere(ctx context.Context, exprs ...dbx.Expression) (int, error) -} - -type settingsRepository interface { - GetByName(ctx context.Context, name string) (*domain.Settings, error) -} - type WorkflowService struct { dispatcher *dispatcher.WorkflowDispatcher @@ -51,29 +35,8 @@ func NewWorkflowService(workflowRepo workflowRepository, workflowRunRepo workflo func (s *WorkflowService) InitSchedule(ctx context.Context) error { // 每日清理工作流执行历史 - app.GetScheduler().MustAdd("workflowHistoryRunsCleanup", "0 0 * * *", func() { - settings, err := s.settingsRepo.GetByName(ctx, "persistence") - if err != nil { - app.GetLogger().Error(fmt.Sprintf("failed to get persistence settings: %w", err)) - return - } - - persistenceSettings, _ := settings.UnmarshalContentAsPersistence() - if persistenceSettings != nil && persistenceSettings.WorkflowRunsMaxDaysRetention != 0 { - ret, err := s.workflowRunRepo.DeleteWhere( - context.Background(), - dbx.NewExp(fmt.Sprintf("status!='%s'", string(domain.WorkflowRunStatusTypePending))), - dbx.NewExp(fmt.Sprintf("status!='%s'", string(domain.WorkflowRunStatusTypeProcessing))), - dbx.NewExp(fmt.Sprintf("endedAt 0 { - app.GetLogger().Info(fmt.Sprintf("cleanup %d workflow history runs", ret)) - } - } + app.GetScheduler().MustAdd("cleanupWorkflowHistoryRuns", "0 0 * * *", func() { + s.cleanupHistoryRuns(context.Background()) }) // 工作流后台任务 @@ -87,7 +50,7 @@ func (s *WorkflowService) InitSchedule(ctx context.Context) error { var errs []error err := app.GetScheduler().Add(fmt.Sprintf("workflow#%s", workflow.Id), workflow.TriggerCron, func() { - s.StartRun(ctx, &dtos.WorkflowStartRunReq{ + s.StartRun(context.Background(), &dtos.WorkflowStartRunReq{ WorkflowId: workflow.Id, RunTrigger: domain.WorkflowTriggerTypeScheduled, }) @@ -161,3 +124,31 @@ func (s *WorkflowService) CancelRun(ctx context.Context, req *dtos.WorkflowCance func (s *WorkflowService) Shutdown(ctx context.Context) { s.dispatcher.Shutdown() } + +func (s *WorkflowService) cleanupHistoryRuns(ctx context.Context) error { + settings, err := s.settingsRepo.GetByName(ctx, "persistence") + if err != nil { + app.GetLogger().Error(fmt.Sprintf("failed to get persistence settings: %w", err)) + return err + } + + persistenceSettings, _ := settings.UnmarshalContentAsPersistence() + if persistenceSettings != nil && persistenceSettings.WorkflowRunsMaxDaysRetention != 0 { + ret, err := s.workflowRunRepo.DeleteWhere( + context.Background(), + dbx.NewExp(fmt.Sprintf("status!='%s'", string(domain.WorkflowRunStatusTypePending))), + dbx.NewExp(fmt.Sprintf("status!='%s'", string(domain.WorkflowRunStatusTypeProcessing))), + dbx.NewExp(fmt.Sprintf("endedAt 0 { + app.GetLogger().Info(fmt.Sprintf("cleanup %d workflow history runs", ret)) + } + } + + return nil +} diff --git a/internal/workflow/service_deps.go b/internal/workflow/service_deps.go new file mode 100644 index 000000000..b95eda059 --- /dev/null +++ b/internal/workflow/service_deps.go @@ -0,0 +1,25 @@ +package workflow + +import ( + "context" + + "github.com/pocketbase/dbx" + + "github.com/certimate-go/certimate/internal/domain" +) + +type workflowRepository interface { + ListEnabledScheduled(ctx context.Context) ([]*domain.Workflow, error) + GetById(ctx context.Context, id string) (*domain.Workflow, error) + Save(ctx context.Context, workflow *domain.Workflow) (*domain.Workflow, error) +} + +type workflowRunRepository interface { + GetById(ctx context.Context, id string) (*domain.WorkflowRun, error) + SaveWithCascading(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error) + DeleteWhere(ctx context.Context, exprs ...dbx.Expression) (int, error) +} + +type settingsRepository interface { + GetByName(ctx context.Context, name string) (*domain.Settings, error) +}