opensource/certimate
3
0
Fork 0
mirror of https://github.com/certimate-go/certimate.git synced 2026-08-30 18:01:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: clean code

Browse Source
This commit is contained in:
Fu Diwei
2025-08-21 19:54:50 +08:00
parent a7b4e921d0
commit a23dd0b9e5
8 changed files with 115 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/certificate/service.go
+28 -31
View File
@@ -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<DATETIME('now', '-%d days')", persistenceSettings.ExpiredCertificatesMaxDaysRetention)),
)
if err != nil {
app.GetLogger().Error("failed to delete expired certificates", "err", err)
}
if ret > 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<DATETIME('now', '-%d days')", persistenceSettings.ExpiredCertificatesMaxDaysRetention)),
)
if err != nil {
app.GetLogger().Error("failed to delete expired certificates", "err", err)
return err
}
if ret > 0 {
app.GetLogger().Info(fmt.Sprintf("cleanup %d expired certificates", ret))
}
}
return nil
}
internal/certificate/service_deps.go
+19
View File
@@ -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)
}
internal/statistics/service.go
-4
View File
@@ -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
}
internal/statistics/service_deps.go
+11
View File
@@ -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)
}
internal/workflow/dispatcher/invoker.go
-88
View File
@@ -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
}
internal/workflow/dispatcher/singleton.go
+1
View File
@@ -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)
}
internal/workflow/service.go
+31 -40
View File
@@ -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<DATETIME('now', '-%d days')", persistenceSettings.WorkflowRunsMaxDaysRetention)),
)
if err != nil {
app.GetLogger().Error(fmt.Sprintf("failed to delete workflow history runs: %w", err))
}
if ret > 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<DATETIME('now', '-%d days')", persistenceSettings.WorkflowRunsMaxDaysRetention)),
)
if err != nil {
app.GetLogger().Error(fmt.Sprintf("failed to delete workflow history runs: %w", err))
return err
}
if ret > 0 {
app.GetLogger().Info(fmt.Sprintf("cleanup %d workflow history runs", ret))
}
}
return nil
}
internal/workflow/service_deps.go
+25
View File
@@ -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)
}