diff --git a/config.example.yaml b/config.example.yaml index 57e2d94..a4e1610 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -98,6 +98,10 @@ jobs: enabled: true interval: 24h completed_retention: 168h + acp_events_purge: + enabled: true + interval: 24h + retention: 168h # 7 天 # 通知通道配置 notify: diff --git a/internal/app/app.go b/internal/app/app.go index 1b4d1ef..944ca2e 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -373,6 +373,8 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) { log) jobReaperRunner := runners.NewJobReaper(jobsRepo, cfg.Jobs.JobReaper.LeaseTimeout, 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) { 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, 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{ Repo: jobsRepo, 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, "job_reaper": jobReaperRunner.Run, "job_purge": jobPurgeRunner.Run, + "acp_events_purge": acpEventsPurgeRunner.Run, }, Clock: clock.Real(), Log: log, @@ -803,3 +811,11 @@ type dispatcherAdapter struct{ d *notify.Dispatcher } func (a dispatcherAdapter) Dispatch(ctx context.Context, msg notify.Message) error { 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) +} diff --git a/internal/config/config.go b/internal/config/config.go index 7c59fa6..e63440a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -120,6 +120,7 @@ type JobsConfig struct { AttachmentCleanup AttachmentCleanupCfg `mapstructure:"attachment_cleanup"` JobReaper JobReaperCfg `mapstructure:"job_reaper"` JobPurge JobPurgeCfg `mapstructure:"job_purge"` + AcpEventsPurge AcpEventsPurgeCfg `mapstructure:"acp_events_purge"` } type RunnerCfg struct { @@ -152,6 +153,12 @@ type JobPurgeCfg struct { 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。 type NotifyConfig struct { 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.interval", "24h") 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.timeout", "10s") v.SetDefault("acp.enabled", true) diff --git a/internal/jobs/domain.go b/internal/jobs/domain.go index 193b0ed..d9bbe79 100644 --- a/internal/jobs/domain.go +++ b/internal/jobs/domain.go @@ -120,6 +120,13 @@ type JobPurgeConfig struct { 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. type Config struct { Enabled bool @@ -130,4 +137,5 @@ type Config struct { AttachmentCleanup AttachmentCleanupConfig JobReaper JobReaperConfig JobPurge JobPurgeConfig + AcpEventsPurge AcpEventsPurgeConfig } diff --git a/internal/jobs/module.go b/internal/jobs/module.go index 34d1dab..f3c5c1d 100644 --- a/internal/jobs/module.go +++ b/internal/jobs/module.go @@ -129,6 +129,8 @@ func scheduleEntry(cfg Config, name string) (schedEntry, bool) { return schedEntry{cfg.JobReaper.Interval, cfg.JobReaper.Enabled}, true case "job_purge": 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 }