feat(app): wire acp_events_purge runner into jobs scheduler

This commit is contained in:
2026-05-07 17:47:50 +08:00
parent 4c19ce29ca
commit faebfc3135
5 changed files with 40 additions and 0 deletions
+4
View File
@@ -98,6 +98,10 @@ jobs:
enabled: true enabled: true
interval: 24h interval: 24h
completed_retention: 168h completed_retention: 168h
acp_events_purge:
enabled: true
interval: 24h
retention: 168h # 7 天
# 通知通道配置 # 通知通道配置
notify: notify:
+16
View File
@@ -373,6 +373,8 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) {
log) log)
jobReaperRunner := runners.NewJobReaper(jobsRepo, cfg.Jobs.JobReaper.LeaseTimeout, log) jobReaperRunner := runners.NewJobReaper(jobsRepo, cfg.Jobs.JobReaper.LeaseTimeout, log)
jobPurgeRunner := runners.NewJobPurge(jobsRepo, cfg.Jobs.JobPurge.CompletedRetention, log) jobPurgeRunner := runners.NewJobPurge(jobsRepo, cfg.Jobs.JobPurge.CompletedRetention, log)
acpEventsPurgeRunner := runners.NewAcpEventsPurge(
acpEventsPurgeAdapter{repo: acpRepo}, cfg.Jobs.AcpEventsPurge.Retention, log)
deadLetterFn := func(ctx context.Context, job jobs.Job, herr error) { deadLetterFn := func(ctx context.Context, job jobs.Job, herr error) {
errMsg := herr.Error() errMsg := herr.Error()
@@ -443,6 +445,11 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) {
Interval: cfg.Jobs.JobPurge.Interval, Interval: cfg.Jobs.JobPurge.Interval,
CompletedRetention: cfg.Jobs.JobPurge.CompletedRetention, CompletedRetention: cfg.Jobs.JobPurge.CompletedRetention,
}, },
AcpEventsPurge: jobs.AcpEventsPurgeConfig{
Enabled: cfg.Jobs.AcpEventsPurge.Enabled,
Interval: cfg.Jobs.AcpEventsPurge.Interval,
Retention: cfg.Jobs.AcpEventsPurge.Retention,
},
}, jobs.ModuleDeps{ }, jobs.ModuleDeps{
Repo: jobsRepo, Repo: jobsRepo,
Handlers: map[jobs.JobType]jobs.Handler{ Handlers: map[jobs.JobType]jobs.Handler{
@@ -454,6 +461,7 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) {
"attachment_cleanup": attCleanupRunner.Run, "attachment_cleanup": attCleanupRunner.Run,
"job_reaper": jobReaperRunner.Run, "job_reaper": jobReaperRunner.Run,
"job_purge": jobPurgeRunner.Run, "job_purge": jobPurgeRunner.Run,
"acp_events_purge": acpEventsPurgeRunner.Run,
}, },
Clock: clock.Real(), Clock: clock.Real(),
Log: log, Log: log,
@@ -803,3 +811,11 @@ type dispatcherAdapter struct{ d *notify.Dispatcher }
func (a dispatcherAdapter) Dispatch(ctx context.Context, msg notify.Message) error { func (a dispatcherAdapter) Dispatch(ctx context.Context, msg notify.Message) error {
return a.d.Dispatch(ctx, msg) return a.d.Dispatch(ctx, msg)
} }
// acpEventsPurgeAdapter exposes acp.Repository.PurgeEventsBefore as the narrow
// interface required by runners.AcpEventsPurge.
type acpEventsPurgeAdapter struct{ repo acp.Repository }
func (a acpEventsPurgeAdapter) PurgeEventsBefore(ctx context.Context, before time.Time) (int64, error) {
return a.repo.PurgeEventsBefore(ctx, before)
}
+10
View File
@@ -120,6 +120,7 @@ type JobsConfig struct {
AttachmentCleanup AttachmentCleanupCfg `mapstructure:"attachment_cleanup"` AttachmentCleanup AttachmentCleanupCfg `mapstructure:"attachment_cleanup"`
JobReaper JobReaperCfg `mapstructure:"job_reaper"` JobReaper JobReaperCfg `mapstructure:"job_reaper"`
JobPurge JobPurgeCfg `mapstructure:"job_purge"` JobPurge JobPurgeCfg `mapstructure:"job_purge"`
AcpEventsPurge AcpEventsPurgeCfg `mapstructure:"acp_events_purge"`
} }
type RunnerCfg struct { type RunnerCfg struct {
@@ -152,6 +153,12 @@ type JobPurgeCfg struct {
CompletedRetention time.Duration `mapstructure:"completed_retention"` CompletedRetention time.Duration `mapstructure:"completed_retention"`
} }
type AcpEventsPurgeCfg struct {
Enabled bool `mapstructure:"enabled"`
Interval time.Duration `mapstructure:"interval"`
Retention time.Duration `mapstructure:"retention"`
}
// NotifyConfig 是 notify 模块的可选外发通道配置。当前仅 webhook。 // NotifyConfig 是 notify 模块的可选外发通道配置。当前仅 webhook。
type NotifyConfig struct { type NotifyConfig struct {
Webhook WebhookCfg `mapstructure:"webhook"` Webhook WebhookCfg `mapstructure:"webhook"`
@@ -227,6 +234,9 @@ func Load(configFile string) (*Config, error) {
v.SetDefault("jobs.job_purge.enabled", true) v.SetDefault("jobs.job_purge.enabled", true)
v.SetDefault("jobs.job_purge.interval", "24h") v.SetDefault("jobs.job_purge.interval", "24h")
v.SetDefault("jobs.job_purge.completed_retention", "168h") v.SetDefault("jobs.job_purge.completed_retention", "168h")
v.SetDefault("jobs.acp_events_purge.enabled", true)
v.SetDefault("jobs.acp_events_purge.interval", "24h")
v.SetDefault("jobs.acp_events_purge.retention", "168h")
v.SetDefault("notify.webhook.enabled", false) v.SetDefault("notify.webhook.enabled", false)
v.SetDefault("notify.webhook.timeout", "10s") v.SetDefault("notify.webhook.timeout", "10s")
v.SetDefault("acp.enabled", true) v.SetDefault("acp.enabled", true)
+8
View File
@@ -120,6 +120,13 @@ type JobPurgeConfig struct {
CompletedRetention time.Duration CompletedRetention time.Duration
} }
// AcpEventsPurgeConfig extends RunnerConfig with acp_events retention.
type AcpEventsPurgeConfig struct {
Enabled bool
Interval time.Duration
Retention time.Duration
}
// Config is the full jobs.Module configuration. // Config is the full jobs.Module configuration.
type Config struct { type Config struct {
Enabled bool Enabled bool
@@ -130,4 +137,5 @@ type Config struct {
AttachmentCleanup AttachmentCleanupConfig AttachmentCleanup AttachmentCleanupConfig
JobReaper JobReaperConfig JobReaper JobReaperConfig
JobPurge JobPurgeConfig JobPurge JobPurgeConfig
AcpEventsPurge AcpEventsPurgeConfig
} }
+2
View File
@@ -129,6 +129,8 @@ func scheduleEntry(cfg Config, name string) (schedEntry, bool) {
return schedEntry{cfg.JobReaper.Interval, cfg.JobReaper.Enabled}, true return schedEntry{cfg.JobReaper.Interval, cfg.JobReaper.Enabled}, true
case "job_purge": case "job_purge":
return schedEntry{cfg.JobPurge.Interval, cfg.JobPurge.Enabled}, true return schedEntry{cfg.JobPurge.Interval, cfg.JobPurge.Enabled}, true
case "acp_events_purge":
return schedEntry{cfg.AcpEventsPurge.Interval, cfg.AcpEventsPurge.Enabled}, true
} }
return schedEntry{0, false}, false return schedEntry{0, false}, false
} }