feat(auth): prune spent one-time codes on a schedule (RUK-296)

Makefile

@@ -280,6 +280,7 @@ mocks:
 	$(GOBIN)/mockgen -typed -destination ./internal/pkg/generated/mocks/goque_processors/invitationrotateprocessor/processor.go -source ./internal/goque_processors/invitationrotateprocessor/processor.go
 	$(GOBIN)/mockgen -typed -destination ./internal/pkg/generated/mocks/goque_processors/invitationpruneprocessor/processor.go -source ./internal/goque_processors/invitationpruneprocessor/processor.go
 	$(GOBIN)/mockgen -typed -destination ./internal/pkg/generated/mocks/goque_processors/licenseheartbeatprocessor/processor.go -source ./internal/goque_processors/licenseheartbeatprocessor/processor.go
+	$(GOBIN)/mockgen -typed -destination ./internal/pkg/generated/mocks/goque_processors/otppruneprocessor/processor.go -source ./internal/goque_processors/otppruneprocessor/processor.go
 	$(GOBIN)/mockgen -typed -destination ./internal/pkg/generated/mocks/services/usersummary/service.go -source ./internal/services/usersummary/service.go
 	$(GOBIN)/mockgen -typed -destination ./internal/pkg/generated/mocks/services/license/service.go -source ./internal/services/license/service.go
 	$(GOBIN)/mockgen -typed -destination ./internal/pkg/generated/mocks/server/middlewares/auth.go -source ./internal/server/middlewares/auth.go

deployment/maintmode/dev/app.config.yaml

@@ -193,6 +193,21 @@ task_processor:
     # 365 days by created_at. Terminal rows older than this are deleted in batches.
     retention: 8760h
     batch_limit: 1000
+  # otp_prune deletes spent one-time codes: rows with kind='otp' whose expires_at
+  # is older than retention. Consumed and merely-expired codes leave through the
+  # same threshold; password rows are never touched.
+  #
+  # Retention is short (unlike the year the other sweeps keep) because the row
+  # holds a code digest and a session nonce while the code itself lives minutes.
+  # Worst-case residency is retention plus one cron period, so ~48h here, not 24h.
+  #
+  # cron_spec must be set here: like audit_prune and invitation_prune, an empty
+  # or malformed value fails NewTaskProcessors and aborts startup. The 03:15
+  # offset keeps this sweep out of the minute the three other daily jobs share.
+  otp_prune:
+    cron_spec: "15 3 * * *"
+    retention: 24h
+    batch_limit: 1000
 
 # crypto addresses the master key (KEK) that wraps the data-encryption keys
 # protecting integration secrets at rest. A KEK is addressed by URI:

deployment/maintmode/local/app.config.yaml

@@ -185,6 +185,21 @@ task_processor:
     cron_spec: "0 3 * * *"
     retention: 8760h
     batch_limit: 1000
+  # otp_prune deletes spent one-time codes: rows with kind='otp' whose expires_at
+  # is older than retention. Consumed and merely-expired codes leave through the
+  # same threshold; password rows are never touched.
+  #
+  # Retention is short (unlike the year the other sweeps keep) because the row
+  # holds a code digest and a session nonce while the code itself lives minutes.
+  # Worst-case residency is retention plus one cron period, so ~48h here, not 24h.
+  #
+  # cron_spec must be set here: like audit_prune and invitation_prune, an empty
+  # or malformed value fails NewTaskProcessors and aborts startup. The 03:15
+  # offset keeps this sweep out of the minute the three other daily jobs share.
+  otp_prune:
+    cron_spec: "15 3 * * *"
+    retention: 24h
+    batch_limit: 1000
 
 # crypto addresses the master key (KEK) that wraps the data-encryption keys
 # protecting integration secrets at rest. A KEK is addressed by URI:

deployment/maintmode/prod/app.config.yaml

@@ -227,6 +227,21 @@ task_processor:
     cron_spec: "0 3 * * *"
     retention: 8760h
     batch_limit: 1000
+  # otp_prune deletes spent one-time codes: rows with kind='otp' whose expires_at
+  # is older than retention. Consumed and merely-expired codes leave through the
+  # same threshold; password rows are never touched.
+  #
+  # Retention is short (unlike the year the other sweeps keep) because the row
+  # holds a code digest and a session nonce while the code itself lives minutes.
+  # Worst-case residency is retention plus one cron period, so ~48h here, not 24h.
+  #
+  # cron_spec must be set here: like audit_prune and invitation_prune, an empty
+  # or malformed value fails NewTaskProcessors and aborts startup. The 03:15
+  # offset keeps this sweep out of the minute the three other daily jobs share.
+  otp_prune:
+    cron_spec: "15 3 * * *"
+    retention: 24h
+    batch_limit: 1000
 
 # crypto addresses the master key (KEK) that wraps the data-encryption keys
 # protecting integration secrets at rest. A KEK is addressed by URI:

deployment/maintmode/test/app.config.yaml

@@ -193,6 +193,21 @@ task_processor:
     cron_spec: "0 3 * * *"
     retention: 8760h
     batch_limit: 1000
+  # otp_prune deletes spent one-time codes: rows with kind='otp' whose expires_at
+  # is older than retention. Consumed and merely-expired codes leave through the
+  # same threshold; password rows are never touched.
+  #
+  # Retention is short (unlike the year the other sweeps keep) because the row
+  # holds a code digest and a session nonce while the code itself lives minutes.
+  # Worst-case residency is retention plus one cron period, so ~48h here, not 24h.
+  #
+  # cron_spec must be set here: like audit_prune and invitation_prune, an empty
+  # or malformed value fails NewTaskProcessors and aborts startup. The 03:15
+  # offset keeps this sweep out of the minute the three other daily jobs share.
+  otp_prune:
+    cron_spec: "15 3 * * *"
+    retention: 24h
+    batch_limit: 1000
 
 # crypto addresses the master key (KEK) that wraps the data-encryption keys
 # protecting integration secrets at rest. A KEK is addressed by URI:

internal/app/bootstrap/processors.go

@@ -15,6 +15,7 @@ import (
 	"github.com/ruko1202/maintmode/internal/goque_processors/invitationrotateprocessor"
 	"github.com/ruko1202/maintmode/internal/goque_processors/licenseheartbeatprocessor"
 	"github.com/ruko1202/maintmode/internal/goque_processors/otpemailprocessor"
+	"github.com/ruko1202/maintmode/internal/goque_processors/otppruneprocessor"
 	"github.com/ruko1202/maintmode/internal/goque_processors/reminderprocessor"
 	"github.com/ruko1202/maintmode/internal/pkg/secrets"
 	"github.com/ruko1202/maintmode/internal/services/otp"
@@ -25,8 +26,10 @@ import (
 
 // NewTaskProcessors builds the single goque worker for the maintmode process and
 // registers every task type the merged process owns: maint.reminder,
-// maint.auto.cancel (+ cron), invitation.email, audit.write and audit.prune
-// (+ cron), plus license.heartbeat (+ cron) when SaaS license mode is enabled.
+// maint.auto.cancel (+ cron), invitation.email, otp.email, audit.write,
+// audit.prune (+ cron), the invitation rotate/prune pair (+ crons) and
+// otp.prune (+ cron), plus license.heartbeat (+ cron) when SaaS license mode is
+// enabled.
 // Everything is registered on one registrar, and verify() runs once at the end to
 // assert the registered set matches entity.ExpectedProcessorTaskTypes for this
 // process's toggles.
@@ -184,13 +187,53 @@ func NewTaskProcessors(
 		reg.RegisterPeriodicJob(heartbeatJob)
 	}
 
+	if err := registerOTPPrune(reg, cfg, services); err != nil {
+		return nil, err
+	}
+
 	if err := reg.verify(entity.ExpectedProcessorTaskTypes(licenseEnabled)); err != nil {
 		return nil, err
 	}
 
 	return goq, nil
 }
 
+// registerOTPPrune registers the one-time-code retention sweep: a daily cron job
+// enqueues one task carrying the retention window and batch limit from config,
+// and the processor deletes auth_credentials rows with kind='otp' whose
+// expires_at is older than that window, in bounded batches. Consumed codes age
+// out through the same predicate; password rows are never eligible.
+//
+// One worker, because the sweep is a single drained DELETE loop that must not run
+// concurrently with itself, and a day-bucketed external id, so multi-replica
+// ticks collapse to one enqueue per day — the same shape as the other sweeps.
+//
+// The cron spec comes straight from config with no code-side default: a missing
+// or malformed value fails here and aborts startup, exactly as it does for
+// audit.prune and invitation.prune. All four deployment stands set it.
+func registerOTPPrune(reg *processorRegistrar, cfg config.TaskProcessorConfig, services *Services) error {
+	pruneCfg := cfg.OTPPrune
+
+	reg.RegisterProcessor(
+		entity.ProcessorTaskOTPPrune,
+		otppruneprocessor.NewTaskProcessor(services.OTP),
+		messagingProcessorOpts(cfg.Messaging, 1)...,
+	)
+
+	pruneJob, err := goque.NewCronJob(
+		entity.ProcessorTaskOTPPruneCron,
+		pruneCfg.CronSpec,
+		time.UTC,
+		otppruneprocessor.NewTaskFactory(pruneCfg.Retention, pruneCfg.BatchLimit),
+	)
+	if err != nil {
+		return fmt.Errorf("failed to build otp-prune cron job: %w", err)
+	}
+	reg.RegisterPeriodicJob(pruneJob)
+
+	return nil
+}
+
 // registerInvitationRotation registers the invitation lifecycle's daily
 // retention pair on reg:
 //

internal/config/app_config.go

@@ -322,6 +322,7 @@ type TaskProcessorConfig struct {
 	AuditPrune       TaskProcessorAuditPruneConfig       `mapstructure:"audit_prune"`
 	InvitationRotate TaskProcessorInvitationRotateConfig `mapstructure:"invitation_rotate"`
 	InvitationPrune  TaskProcessorInvitationPruneConfig  `mapstructure:"invitation_prune"`
+	OTPPrune         TaskProcessorOTPPruneConfig         `mapstructure:"otp_prune"`
 }
 
 // CryptoConfig addresses the master keys (KEKs) that wrap the data-encryption
@@ -463,6 +464,24 @@ type TaskProcessorInvitationPruneConfig struct {
 	BatchLimit int64 `mapstructure:"batch_limit"`
 }
 
+// TaskProcessorOTPPruneConfig tunes the one-time-code retention sweep that
+// deletes spent OTP credentials (see services/otp.Service.Prune).
+type TaskProcessorOTPPruneConfig struct {
+	// CronSpec is the 5-field schedule for the producer job. Unlike the sibling
+	// sweeps this one has a code-side fallback, so leaving it empty degrades to
+	// the default schedule rather than aborting startup. The task is
+	// day-bucketed, so firing more often than daily still yields one prune a day.
+	CronSpec string `mapstructure:"cron_spec"`
+	// Retention is the age threshold: an OTP whose expires_at is older than
+	// now-Retention is deleted. Short by design (24h), because the row holds a
+	// code digest and a session nonce and the code itself lives only minutes.
+	// Worst-case residency is Retention plus one cron period.
+	Retention time.Duration `mapstructure:"retention"`
+	// BatchLimit bounds how many rows one DELETE statement removes; the sweep loops
+	// batches until the table is drained for the cutoff.
+	BatchLimit int64 `mapstructure:"batch_limit"`
+}
+
 type JWTVerifierConfig struct {
 	// This struct is shared by two verifiers that use DIFFERENT issuer fields.
 	// Set the one your consumer reads; validateIssuerConfig enforces both at
@@ -633,6 +652,10 @@ func initConfig(appName string) *AppConfig {
 		log.Panicf("invalid config for service %s: %s", appName, err)
 	}
 
+	if err := cfg.validateOTPRetention(); err != nil {
+		log.Panicf("invalid config for service %s: %s", appName, err)
+	}
+
 	if err := cfg.validateValkeyConfig(); err != nil {
 		log.Panicf("invalid config for service %s: %s", appName, err)
 	}
@@ -851,6 +874,27 @@ func (c *AppConfig) validateInvitationRetention() error {
 	return nil
 }
 
+// validateOTPRetention rejects a negative otp-prune retention at startup, for
+// the same reason as its invitation twin: a negative value would push the prune
+// cutoff into the future, and it is always an operator typo rather than an
+// intent. The service clamps it defensively too — that clamp, not this check, is
+// what actually keeps a live code from being deleted — but a bad value in config
+// should still be loud. Zero is allowed and means "unset".
+//
+// Deliberately no comparison against the OTP TTL. A code is only eligible once
+// its expires_at is already past, so every positive retention is safe however
+// small, and reaching otp.TTL from here would be an import cycle anyway
+// (services/otp imports this package).
+func (c *AppConfig) validateOTPRetention() error {
+	if c.TaskProcessor.OTPPrune.Retention < 0 {
+		return fmt.Errorf(
+			"task_processor.otp_prune.retention must not be negative, got %s",
+			c.TaskProcessor.OTPPrune.Retention,
+		)
+	}
+	return nil
+}
+
 // validateIssuerConfig rejects a verifier config that would silently stop
 // checking the token issuer.
 //

internal/entity/goque_processor_owner_test.go

@@ -26,6 +26,8 @@ var allDeclaredTaskTypes = []string{
 	ProcessorTaskInvitationPrune,
 	ProcessorTaskInvitationPruneCron,
 	ProcessorTaskOTPEmailSend,
+	ProcessorTaskOTPPrune,
+	ProcessorTaskOTPPruneCron,
 }
 
 // disabledTaskTypes is every declared type whose processor is intentionally not