diff --git a/Makefile b/Makefile index 63a0be87..7c547cf8 100644 --- a/Makefile +++ b/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 diff --git a/deployment/maintmode/dev/app.config.yaml b/deployment/maintmode/dev/app.config.yaml index 70336cfe..7e34cb51 100644 --- a/deployment/maintmode/dev/app.config.yaml +++ b/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: diff --git a/deployment/maintmode/local/app.config.yaml b/deployment/maintmode/local/app.config.yaml index b3a358eb..dccf9dcf 100644 --- a/deployment/maintmode/local/app.config.yaml +++ b/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: diff --git a/deployment/maintmode/prod/app.config.yaml b/deployment/maintmode/prod/app.config.yaml index c7e2ec8d..c7e2ca27 100644 --- a/deployment/maintmode/prod/app.config.yaml +++ b/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: diff --git a/deployment/maintmode/test/app.config.yaml b/deployment/maintmode/test/app.config.yaml index 1307bd93..e5d97bba 100644 --- a/deployment/maintmode/test/app.config.yaml +++ b/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: diff --git a/docs/runbooks/otp-signin.md b/docs/runbooks/otp-signin.md deleted file mode 100644 index c049c3be..00000000 --- a/docs/runbooks/otp-signin.md +++ /dev/null @@ -1,108 +0,0 @@ -# Runbook: email one-time-code sign-in - -Covers `POST /api/v1/login/otp/request` and `POST /api/v1/login/otp/verify`. - -Both endpoints answer every failure identically on purpose, so a user's report -("it says the code is wrong") rarely narrows anything down. The audit trail and -the logs are where the distinctions live. - -## Limiter degradation - -Three tiers guard these routes, and they degrade differently when Valkey is -unreachable: - -| Tier | Key | On a Valkey outage | -| --- | --- | --- | -| Per IP | client address | per-replica bucket, so the effective cap is `N × limit` across `N` replicas | -| Per address | the email in the body | same, per replica | -| Instance-wide | a constant | **stops applying entirely** | - -The instance-wide tier is the deliberate exception. Its key is a constant, so a -per-replica bucket would put every caller into one in-memory bucket and let a -single attacker hold the whole sign-in surface at 429. Losing the anti-sweep -control during an outage beats converting the outage into a total sign-in -outage; the per-address tier, which protects an individual account, keeps -working throughout. - -The practical consequence: **during a Valkey outage this deployment is more -exposed to a distributed sweep than usual.** If an outage coincides with a spike -in failed verifications, treat the two as related. - -## Reading the fallback signal - -The alert is `RateLimiterValkeyFallback`, on `ratelimit_valkey_failures_total`. - -That counter carries **no tier label**, so it cannot tell you which limiter fell -back. Since one Valkey outage is the common cause for all three, read a firing -alert as *all three degraded at once* — per-IP and per-address to `N × limit`, -instance-wide to nothing. - -A sustained non-zero rate means Valkey, not an attack. Check Valkey's own health -first; the limiters recover on their own once it answers. - -`otp_attempt_claim_errors_total` is separate and more serious. It counts -failures to claim a guess against a code, and the claim *is* the per-code -attempt ceiling. A sustained non-zero rate means the brute-force ceiling is not -being enforced — verification fails closed while this happens, so users see -errors rather than a silent hole, but the cause is a database problem and -nothing else will say so. - -## Clearing the burnt-code barrier - -Spending every guess on a code does **not** free the active-code slot: no new -code is issued until the burnt one expires (at most `auth.otp_ttl`, 5 minutes by -default). Without that rule, "five attempts" would mean "five attempts per code, -unlimited codes". - -The refusal is deliberately invisible — the same `202` and the same response -shape as a successful request — and deliberately unaudited, since no secret was -presented and auditing it would let anyone write unbounded rows by replaying the -endpoint. So a barred user has no self-service diagnosis, and this is the only -way to confirm it. - -Find a barred address: - -```sql -SELECT c.id, u.email, c.attempts, c.expires_at -FROM auth_credentials c -JOIN users u ON u.id = c.user_id -WHERE c.kind = 'otp' - AND c.consumed_at IS NULL - AND c.expires_at > now() - AND c.attempts >= 5 -- auth.otp_max_attempts - AND u.email = 'user@example.com'; -``` - -Clear it, **keyed on the id from that query**: - -```sql -UPDATE auth_credentials SET consumed_at = now() WHERE id = ''; -``` - -The predicate is not optional. Running the `SET` without a `WHERE` retires every -live code on the instance: nobody is signed in by it, and everyone waiting on a -code has to request a new one. - -Usually the right answer is to wait. The barrier is at most one `otp_ttl`, and -clearing it by hand for a user who is being targeted removes the control while -the attempt is in progress. - -## Reading a 429 storm - -A burst of 429s on `/login/otp` no longer means "one caller is hammering us". -With an instance-wide tier, one attacker exhausting the global budget returns -429 to **every** caller, including legitimate ones. Nothing in the metrics -distinguishes that from a local burst or from a Valkey fallback. - -To tell them apart: - -- Are the 429s spread across many client addresses? That points at the global - tier, so a sweep is in progress and everyone is being refused. -- Concentrated on one address? Per-IP or per-address, and the ordinary cause. -- Is `RateLimiterValkeyFallback` firing? Then caps are per-replica and the - instance-wide tier is off — the 429s are not coming from it. - -Any alert added on this route later must exclude 429. The per-IP limiter carries -no route component, so `login/oauth`, `login/password` and `users/invitations` -share one budget with these routes and produce 429s here for traffic that never -touched them. diff --git a/internal/app/bootstrap/processors.go b/internal/app/bootstrap/processors.go index c78469bf..d56fb317 100644 --- a/internal/app/bootstrap/processors.go +++ b/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,6 +187,10 @@ 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 } @@ -191,6 +198,42 @@ func NewTaskProcessors( 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: // diff --git a/internal/config/app_config.go b/internal/config/app_config.go index 78f20750..f8e96557 100644 --- a/internal/config/app_config.go +++ b/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. // diff --git a/internal/entity/goque_processor_owner_test.go b/internal/entity/goque_processor_owner_test.go index add4f445..4a2c7732 100644 --- a/internal/entity/goque_processor_owner_test.go +++ b/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 diff --git a/internal/entity/goque_processors_task.go b/internal/entity/goque_processors_task.go index da8cf88c..93468860 100644 --- a/internal/entity/goque_processors_task.go +++ b/internal/entity/goque_processors_task.go @@ -81,6 +81,18 @@ const ( // generic sender processor could not decode it. The body does not exist at // enqueue time -- it is rendered by this type's own processor. ProcessorTaskOTPEmailSend = "otp.email" + // ProcessorTaskOTPPrune is the goque task type produced by the OTP-retention + // periodic job. Its processor deletes auth_credentials rows with kind='otp' + // whose expires_at is older than the retention window, in bounded batches. + // + // Age is measured on expires_at, not created_at, because + // auth_credentials_otp_expiry_idx is the only index on the table and + // migrations were out of scope. A consumed code carries an expiry too and + // ages out through the same sweep, so consumed_at never enters the predicate. + // password rows are never eligible. The payload carries the retention window + // and batch limit (from config). + ProcessorTaskOTPPrune = "otp.prune" + ProcessorTaskOTPPruneCron = "otp.prune.cron" ) // ActiveProcessorTaskTypes is the set of goque task types the process must @@ -108,6 +120,8 @@ var ActiveProcessorTaskTypes = map[string]struct{}{ ProcessorTaskInvitationPrune: {}, ProcessorTaskInvitationPruneCron: {}, ProcessorTaskOTPEmailSend: {}, + ProcessorTaskOTPPrune: {}, + ProcessorTaskOTPPruneCron: {}, } // ExpectedProcessorTaskTypes returns the exact task-type set the process must @@ -214,6 +228,16 @@ type ProcessorTaskPayloadInvitationPrune struct { BatchLimit int64 `json:"batch_limit"` } +// ProcessorTaskPayloadOTPPrune is the payload of an OTP-retention sweep task. +// Same shape and same reason as the invitation one: the cron job stamps the +// tunables from config so the processor stays config-free. Retention is the age +// past expires_at at which a one-time code is deleted, BatchLimit bounds how +// many rows one DELETE removes. +type ProcessorTaskPayloadOTPPrune struct { + Retention time.Duration `json:"retention"` + BatchLimit int64 `json:"batch_limit"` +} + // ProcessorTaskPayloadAuditWrite is the payload of an audit-write task. // It is the rendered, point-in-time snapshot of one audit event: // the publisher fills every persisted field at dispatch time and the processor diff --git a/internal/entity/goque_processors_task_test.go b/internal/entity/goque_processors_task_test.go index 60da1c34..e460369f 100644 --- a/internal/entity/goque_processors_task_test.go +++ b/internal/entity/goque_processors_task_test.go @@ -111,3 +111,30 @@ func TestProcessorTaskPayloadAuditWrite_ToAuditEntry(t *testing.T) { require.Equal(t, payload.Details, entry.Details) require.Equal(t, payload.Metadata, entry.Metadata) } + +func TestOTPPruneTaskTypes_RegisteredInGuard(t *testing.T) { + // Same contract as the invitation pair above: both the task and its .cron + // producer must sit in the active set, or NewTaskProcessors registers a + // processor the guard does not expect and startup fails. + for _, taskType := range []string{ + ProcessorTaskOTPPrune, + ProcessorTaskOTPPruneCron, + } { + _, ok := ActiveProcessorTaskTypes[taskType] + require.Truef(t, ok, "task type %q must be in ActiveProcessorTaskTypes", taskType) + } +} + +func TestProcessorTaskPayloadOTPPrune_JSONRoundTrip(t *testing.T) { + want := ProcessorTaskPayloadOTPPrune{ + Retention: 24 * time.Hour, + BatchLimit: 1000, + } + + raw, err := json.Marshal(want) + require.NoError(t, err) + + var got ProcessorTaskPayloadOTPPrune + require.NoError(t, json.Unmarshal(raw, &got)) + require.Equal(t, want, got) +} diff --git a/internal/goque_processors/otppruneprocessor/external_id.go b/internal/goque_processors/otppruneprocessor/external_id.go new file mode 100644 index 00000000..37050e54 --- /dev/null +++ b/internal/goque_processors/otppruneprocessor/external_id.go @@ -0,0 +1,21 @@ +package otppruneprocessor + +import ( + "fmt" + "time" +) + +// otpPruneExternalID derives the deterministic, day-bucketed external id for an +// otp.prune task. Truncating to the day makes every replica that ticks within +// the same day produce the same id, so the goque (type, external_id) unique +// constraint dedupes them to a single enqueued task per day. Retention is a +// daily-granularity concern, so one prune per day is the intended cadence even +// if the cron schedule fires more often. +// +// The bucket is computed from wall-clock fire time, so replicas with clock skew +// straddling a day boundary can land in adjacent buckets and enqueue twice. That +// is harmless: the prune is idempotent (a second run on the same cutoff finds +// nothing left to delete) and bounded. +func otpPruneExternalID(now time.Time) string { + return fmt.Sprintf("otp-prune-%s", now.UTC().Format("2006-01-02")) +} diff --git a/internal/goque_processors/otppruneprocessor/processor.go b/internal/goque_processors/otppruneprocessor/processor.go new file mode 100644 index 00000000..8c2e6837 --- /dev/null +++ b/internal/goque_processors/otppruneprocessor/processor.go @@ -0,0 +1,71 @@ +// Package otppruneprocessor handles otp.prune goque tasks. The task is produced +// once per cron tick by a periodic job and carries the retention tunables +// (window + batch limit) in its payload; at process time it deletes one-time +// codes whose expires_at is older than the retention window, in bounded batches. +package otppruneprocessor + +import ( + "cmp" + "context" + "time" + + "github.com/ruko1202/goque" + "github.com/ruko1202/xlog" + + "github.com/ruko1202/maintmode/internal/entity" + "github.com/ruko1202/maintmode/internal/utils/xtime" +) + +const ( + // Both defaults must match the ones the service falls back to and the values + // shipped in deployment config, so an unset payload field behaves the same + // wherever it is resolved. 24h is short on purpose: the row holds a code + // digest and a session nonce, and the code itself lives only minutes. + defaultRetention = 24 * time.Hour + defaultBatchLimit = 1000 +) + +// Pruner deletes one-time codes older than the retention window. Defined +// consumer-side so the processor can be tested with a mock. +type Pruner interface { + Prune(ctx context.Context, retention time.Duration, batchLimit int64) error +} + +// NewTaskProcessor returns the goque TaskProcessor for ProcessorTaskOTPPrune. It +// reads the retention window and batch limit from the task payload and runs the +// prune. +func NewTaskProcessor(pruner Pruner) goque.TaskProcessor { + return goque.NewTypedTaskProcessor( + goque.TypedTaskProcessorFunc[entity.ProcessorTaskPayloadOTPPrune]( + func(ctx context.Context, task *goque.TypedTask[entity.ProcessorTaskPayloadOTPPrune]) error { + ctx, span := xlog.WithOperationSpan(ctx, "service.OTP.PruneProcessor.ProcessTask") + defer span.End() + + return pruner.Prune(ctx, task.Payload.Retention, task.Payload.BatchLimit) + }, + ), + goque.WithCancelTaskWhenPayloadDecodeError[entity.ProcessorTaskPayloadOTPPrune](), + ) +} + +// NewTaskFactory returns the goque PeriodicJobFactory that produces one otp.prune +// task per cron tick, stamping the configured retention window and batch limit +// into the payload. +// +// The external id is bucketed to the day so that, with several replicas all +// ticking the same schedule, only the first insert for a given day succeeds; the +// others collide on the (type, external_id) unique key and return +// goque.ErrDuplicateTask -- the desired at-most-once-per-day fan-in. On a single +// replica this never fires. +func NewTaskFactory(retention time.Duration, batchLimit int64) goque.PeriodicJobFactory { + return func(_ context.Context) (*goque.Task, error) { + return goque.NewTaskWithPayloadAndExternalID( + entity.ProcessorTaskOTPPrune, + entity.ProcessorTaskPayloadOTPPrune{ + Retention: cmp.Or(retention, defaultRetention), + BatchLimit: cmp.Or(batchLimit, defaultBatchLimit), + }, + otpPruneExternalID(xtime.UTCNow()), + ) + } +} diff --git a/internal/goque_processors/otppruneprocessor/processor_test.go b/internal/goque_processors/otppruneprocessor/processor_test.go new file mode 100644 index 00000000..3a133a0d --- /dev/null +++ b/internal/goque_processors/otppruneprocessor/processor_test.go @@ -0,0 +1,136 @@ +package otppruneprocessor + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/ruko1202/goque" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + + "github.com/ruko1202/maintmode/internal/entity" + mock_otppruneprocessor "github.com/ruko1202/maintmode/internal/pkg/generated/mocks/goque_processors/otppruneprocessor" +) + +func newPruneTask(t *testing.T, retention time.Duration, batchLimit int64) *goque.Task { + t.Helper() + + task, err := goque.NewTaskWithPayloadAndExternalID( + entity.ProcessorTaskOTPPrune, + entity.ProcessorTaskPayloadOTPPrune{Retention: retention, BatchLimit: batchLimit}, + "test-external-id", + ) + require.NoError(t, err) + + return task +} + +func TestProcessTask_DelegatesWithPayloadTunables(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + pruner := mock_otppruneprocessor.NewMockPruner(ctrl) + + var gotRetention time.Duration + var gotLimit int64 + pruner.EXPECT(). + Prune(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, retention time.Duration, batchLimit int64) error { + gotRetention, gotLimit = retention, batchLimit + return nil + }) + + err := NewTaskProcessor(pruner).ProcessTask(context.Background(), newPruneTask(t, 24*time.Hour, 1000)) + require.NoError(t, err) + require.Equal(t, 24*time.Hour, gotRetention) + require.Equal(t, int64(1000), gotLimit) +} + +func TestProcessTask_PropagatesError(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + pruner := mock_otppruneprocessor.NewMockPruner(ctrl) + + wantErr := errors.New("prune failed") + pruner.EXPECT().Prune(gomock.Any(), gomock.Any(), gomock.Any()).Return(wantErr) + + err := NewTaskProcessor(pruner).ProcessTask(context.Background(), newPruneTask(t, time.Hour, 10)) + require.ErrorIs(t, err, wantErr) +} + +// TestNewTaskFactory_StampsTunablesAndDayBucket asserts the tunables the factory +// stamps survive the JSON round-trip and reach the pruner unchanged. +// +// The task type is asserted because the factory chooses which constant to stamp, +// and stamping the wrong one -- the .cron producer type, say -- would enqueue +// tasks that no registered processor drains, losing the work silently. That is +// verified by mutation, not assumed. +// +// Nothing else about the task struct is asserted: NotNil after a nil error, or a +// non-empty external id, is goque echoing its own arguments back. The external +// id's actual value is pinned by TestOTPPruneExternalID_DayBucketed. +func TestNewTaskFactory_StampsTunablesAndDayBucket(t *testing.T) { + t.Parallel() + + task, err := NewTaskFactory(48*time.Hour, 500)(context.Background()) + require.NoError(t, err) + require.Equal(t, entity.ProcessorTaskOTPPrune, task.Type) + + ctrl := gomock.NewController(t) + pruner := mock_otppruneprocessor.NewMockPruner(ctrl) + + var gotRetention time.Duration + var gotLimit int64 + pruner.EXPECT(). + Prune(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, retention time.Duration, batchLimit int64) error { + gotRetention, gotLimit = retention, batchLimit + return nil + }) + + require.NoError(t, NewTaskProcessor(pruner).ProcessTask(context.Background(), task)) + require.Equal(t, 48*time.Hour, gotRetention) + require.Equal(t, int64(500), gotLimit) +} + +// TestNewTaskFactory_DefaultsZeroTunables checks the cmp.Or fallbacks, and pins +// that the retention default is the 24h the service and the deployment config +// also use -- three values that must not drift apart. +func TestNewTaskFactory_DefaultsZeroTunables(t *testing.T) { + t.Parallel() + + task, err := NewTaskFactory(0, 0)(context.Background()) + require.NoError(t, err) + + ctrl := gomock.NewController(t) + pruner := mock_otppruneprocessor.NewMockPruner(ctrl) + + var gotRetention time.Duration + var gotLimit int64 + pruner.EXPECT(). + Prune(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, retention time.Duration, batchLimit int64) error { + gotRetention, gotLimit = retention, batchLimit + return nil + }) + + require.NoError(t, NewTaskProcessor(pruner).ProcessTask(context.Background(), task)) + require.Equal(t, defaultRetention, gotRetention) + require.Equal(t, 24*time.Hour, gotRetention) + require.Equal(t, int64(defaultBatchLimit), gotLimit) +} + +func TestOTPPruneExternalID_DayBucketed(t *testing.T) { + t.Parallel() + + base := time.Date(2026, 9, 5, 3, 15, 0, 0, time.UTC) + require.Equal(t, "otp-prune-2026-09-05", otpPruneExternalID(base)) + + // Any other moment in the same UTC day must collapse to the same id -- that + // is what dedupes replicas ticking the same schedule. + require.Equal(t, otpPruneExternalID(base), otpPruneExternalID(base.Add(20*time.Hour))) + require.NotEqual(t, otpPruneExternalID(base), otpPruneExternalID(base.Add(24*time.Hour))) +} diff --git a/internal/pkg/generated/mocks/goque_processors/otppruneprocessor/processor.go b/internal/pkg/generated/mocks/goque_processors/otppruneprocessor/processor.go new file mode 100644 index 00000000..f41f79f8 --- /dev/null +++ b/internal/pkg/generated/mocks/goque_processors/otppruneprocessor/processor.go @@ -0,0 +1,80 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: ./internal/goque_processors/otppruneprocessor/processor.go +// +// Generated by this command: +// +// mockgen -typed -destination ./internal/pkg/generated/mocks/goque_processors/otppruneprocessor/processor.go -source ./internal/goque_processors/otppruneprocessor/processor.go +// + +// Package mock_otppruneprocessor is a generated GoMock package. +package mock_otppruneprocessor + +import ( + context "context" + reflect "reflect" + time "time" + + gomock "go.uber.org/mock/gomock" +) + +// MockPruner is a mock of Pruner interface. +type MockPruner struct { + ctrl *gomock.Controller + recorder *MockPrunerMockRecorder + isgomock struct{} +} + +// MockPrunerMockRecorder is the mock recorder for MockPruner. +type MockPrunerMockRecorder struct { + mock *MockPruner +} + +// NewMockPruner creates a new mock instance. +func NewMockPruner(ctrl *gomock.Controller) *MockPruner { + mock := &MockPruner{ctrl: ctrl} + mock.recorder = &MockPrunerMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockPruner) EXPECT() *MockPrunerMockRecorder { + return m.recorder +} + +// Prune mocks base method. +func (m *MockPruner) Prune(ctx context.Context, retention time.Duration, batchLimit int64) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Prune", ctx, retention, batchLimit) + ret0, _ := ret[0].(error) + return ret0 +} + +// Prune indicates an expected call of Prune. +func (mr *MockPrunerMockRecorder) Prune(ctx, retention, batchLimit any) *MockPrunerPruneCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Prune", reflect.TypeOf((*MockPruner)(nil).Prune), ctx, retention, batchLimit) + return &MockPrunerPruneCall{Call: call} +} + +// MockPrunerPruneCall wrap *gomock.Call +type MockPrunerPruneCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockPrunerPruneCall) Return(arg0 error) *MockPrunerPruneCall { + c.Call = c.Call.Return(arg0) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockPrunerPruneCall) Do(f func(context.Context, time.Duration, int64) error) *MockPrunerPruneCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockPrunerPruneCall) DoAndReturn(f func(context.Context, time.Duration, int64) error) *MockPrunerPruneCall { + c.Call = c.Call.DoAndReturn(f) + return c +} diff --git a/internal/pkg/generated/mocks/services/otp/service.go b/internal/pkg/generated/mocks/services/otp/service.go new file mode 100644 index 00000000..0011504d --- /dev/null +++ b/internal/pkg/generated/mocks/services/otp/service.go @@ -0,0 +1,531 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: ./internal/services/otp/service.go +// +// Generated by this command: +// +// mockgen -typed -destination ./internal/pkg/generated/mocks/services/otp/service.go -source ./internal/services/otp/service.go +// + +// Package mock_otp is a generated GoMock package. +package mock_otp + +import ( + context "context" + reflect "reflect" + time "time" + + uuid "github.com/google/uuid" + entity "github.com/ruko1202/maintmode/internal/entity" + gomock "go.uber.org/mock/gomock" +) + +// MockStore is a mock of Store interface. +type MockStore struct { + ctrl *gomock.Controller + recorder *MockStoreMockRecorder + isgomock struct{} +} + +// MockStoreMockRecorder is the mock recorder for MockStore. +type MockStoreMockRecorder struct { + mock *MockStore +} + +// NewMockStore creates a new mock instance. +func NewMockStore(ctrl *gomock.Controller) *MockStore { + mock := &MockStore{ctrl: ctrl} + mock.recorder = &MockStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockStore) EXPECT() *MockStoreMockRecorder { + return m.recorder +} + +// ClaimOTPAttempt mocks base method. +func (m *MockStore) ClaimOTPAttempt(ctx context.Context, id uuid.UUID, maxAttempts int16) (bool, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ClaimOTPAttempt", ctx, id, maxAttempts) + ret0, _ := ret[0].(bool) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ClaimOTPAttempt indicates an expected call of ClaimOTPAttempt. +func (mr *MockStoreMockRecorder) ClaimOTPAttempt(ctx, id, maxAttempts any) *MockStoreClaimOTPAttemptCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ClaimOTPAttempt", reflect.TypeOf((*MockStore)(nil).ClaimOTPAttempt), ctx, id, maxAttempts) + return &MockStoreClaimOTPAttemptCall{Call: call} +} + +// MockStoreClaimOTPAttemptCall wrap *gomock.Call +type MockStoreClaimOTPAttemptCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockStoreClaimOTPAttemptCall) Return(arg0 bool, arg1 error) *MockStoreClaimOTPAttemptCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockStoreClaimOTPAttemptCall) Do(f func(context.Context, uuid.UUID, int16) (bool, error)) *MockStoreClaimOTPAttemptCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockStoreClaimOTPAttemptCall) DoAndReturn(f func(context.Context, uuid.UUID, int16) (bool, error)) *MockStoreClaimOTPAttemptCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// ConsumeOTP mocks base method. +func (m *MockStore) ConsumeOTP(ctx context.Context, id uuid.UUID) (bool, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ConsumeOTP", ctx, id) + ret0, _ := ret[0].(bool) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ConsumeOTP indicates an expected call of ConsumeOTP. +func (mr *MockStoreMockRecorder) ConsumeOTP(ctx, id any) *MockStoreConsumeOTPCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ConsumeOTP", reflect.TypeOf((*MockStore)(nil).ConsumeOTP), ctx, id) + return &MockStoreConsumeOTPCall{Call: call} +} + +// MockStoreConsumeOTPCall wrap *gomock.Call +type MockStoreConsumeOTPCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockStoreConsumeOTPCall) Return(arg0 bool, arg1 error) *MockStoreConsumeOTPCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockStoreConsumeOTPCall) Do(f func(context.Context, uuid.UUID) (bool, error)) *MockStoreConsumeOTPCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockStoreConsumeOTPCall) DoAndReturn(f func(context.Context, uuid.UUID) (bool, error)) *MockStoreConsumeOTPCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// Create mocks base method. +func (m *MockStore) Create(ctx context.Context, cred *entity.AuthCredential) (*entity.AuthCredential, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, cred) + ret0, _ := ret[0].(*entity.AuthCredential) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Create indicates an expected call of Create. +func (mr *MockStoreMockRecorder) Create(ctx, cred any) *MockStoreCreateCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockStore)(nil).Create), ctx, cred) + return &MockStoreCreateCall{Call: call} +} + +// MockStoreCreateCall wrap *gomock.Call +type MockStoreCreateCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockStoreCreateCall) Return(arg0 *entity.AuthCredential, arg1 error) *MockStoreCreateCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockStoreCreateCall) Do(f func(context.Context, *entity.AuthCredential) (*entity.AuthCredential, error)) *MockStoreCreateCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockStoreCreateCall) DoAndReturn(f func(context.Context, *entity.AuthCredential) (*entity.AuthCredential, error)) *MockStoreCreateCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// GetUnconsumedOTPByUserID mocks base method. +func (m *MockStore) GetUnconsumedOTPByUserID(ctx context.Context, userID uuid.UUID) (*entity.AuthCredential, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetUnconsumedOTPByUserID", ctx, userID) + ret0, _ := ret[0].(*entity.AuthCredential) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetUnconsumedOTPByUserID indicates an expected call of GetUnconsumedOTPByUserID. +func (mr *MockStoreMockRecorder) GetUnconsumedOTPByUserID(ctx, userID any) *MockStoreGetUnconsumedOTPByUserIDCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetUnconsumedOTPByUserID", reflect.TypeOf((*MockStore)(nil).GetUnconsumedOTPByUserID), ctx, userID) + return &MockStoreGetUnconsumedOTPByUserIDCall{Call: call} +} + +// MockStoreGetUnconsumedOTPByUserIDCall wrap *gomock.Call +type MockStoreGetUnconsumedOTPByUserIDCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockStoreGetUnconsumedOTPByUserIDCall) Return(arg0 *entity.AuthCredential, arg1 error) *MockStoreGetUnconsumedOTPByUserIDCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockStoreGetUnconsumedOTPByUserIDCall) Do(f func(context.Context, uuid.UUID) (*entity.AuthCredential, error)) *MockStoreGetUnconsumedOTPByUserIDCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockStoreGetUnconsumedOTPByUserIDCall) DoAndReturn(f func(context.Context, uuid.UUID) (*entity.AuthCredential, error)) *MockStoreGetUnconsumedOTPByUserIDCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// GetUnconsumedOTPByUserIDForUpdate mocks base method. +func (m *MockStore) GetUnconsumedOTPByUserIDForUpdate(ctx context.Context, userID uuid.UUID) (*entity.AuthCredential, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetUnconsumedOTPByUserIDForUpdate", ctx, userID) + ret0, _ := ret[0].(*entity.AuthCredential) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetUnconsumedOTPByUserIDForUpdate indicates an expected call of GetUnconsumedOTPByUserIDForUpdate. +func (mr *MockStoreMockRecorder) GetUnconsumedOTPByUserIDForUpdate(ctx, userID any) *MockStoreGetUnconsumedOTPByUserIDForUpdateCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetUnconsumedOTPByUserIDForUpdate", reflect.TypeOf((*MockStore)(nil).GetUnconsumedOTPByUserIDForUpdate), ctx, userID) + return &MockStoreGetUnconsumedOTPByUserIDForUpdateCall{Call: call} +} + +// MockStoreGetUnconsumedOTPByUserIDForUpdateCall wrap *gomock.Call +type MockStoreGetUnconsumedOTPByUserIDForUpdateCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockStoreGetUnconsumedOTPByUserIDForUpdateCall) Return(arg0 *entity.AuthCredential, arg1 error) *MockStoreGetUnconsumedOTPByUserIDForUpdateCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockStoreGetUnconsumedOTPByUserIDForUpdateCall) Do(f func(context.Context, uuid.UUID) (*entity.AuthCredential, error)) *MockStoreGetUnconsumedOTPByUserIDForUpdateCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockStoreGetUnconsumedOTPByUserIDForUpdateCall) DoAndReturn(f func(context.Context, uuid.UUID) (*entity.AuthCredential, error)) *MockStoreGetUnconsumedOTPByUserIDForUpdateCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// PruneOTPExpiredBefore mocks base method. +func (m *MockStore) PruneOTPExpiredBefore(ctx context.Context, cutoff time.Time, limit int64) (int64, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "PruneOTPExpiredBefore", ctx, cutoff, limit) + ret0, _ := ret[0].(int64) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// PruneOTPExpiredBefore indicates an expected call of PruneOTPExpiredBefore. +func (mr *MockStoreMockRecorder) PruneOTPExpiredBefore(ctx, cutoff, limit any) *MockStorePruneOTPExpiredBeforeCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PruneOTPExpiredBefore", reflect.TypeOf((*MockStore)(nil).PruneOTPExpiredBefore), ctx, cutoff, limit) + return &MockStorePruneOTPExpiredBeforeCall{Call: call} +} + +// MockStorePruneOTPExpiredBeforeCall wrap *gomock.Call +type MockStorePruneOTPExpiredBeforeCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockStorePruneOTPExpiredBeforeCall) Return(arg0 int64, arg1 error) *MockStorePruneOTPExpiredBeforeCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockStorePruneOTPExpiredBeforeCall) Do(f func(context.Context, time.Time, int64) (int64, error)) *MockStorePruneOTPExpiredBeforeCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockStorePruneOTPExpiredBeforeCall) DoAndReturn(f func(context.Context, time.Time, int64) (int64, error)) *MockStorePruneOTPExpiredBeforeCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// MockUserService is a mock of UserService interface. +type MockUserService struct { + ctrl *gomock.Controller + recorder *MockUserServiceMockRecorder + isgomock struct{} +} + +// MockUserServiceMockRecorder is the mock recorder for MockUserService. +type MockUserServiceMockRecorder struct { + mock *MockUserService +} + +// NewMockUserService creates a new mock instance. +func NewMockUserService(ctrl *gomock.Controller) *MockUserService { + mock := &MockUserService{ctrl: ctrl} + mock.recorder = &MockUserServiceMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockUserService) EXPECT() *MockUserServiceMockRecorder { + return m.recorder +} + +// GetByEmail mocks base method. +func (m *MockUserService) GetByEmail(ctx context.Context, email string) (*entity.User, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetByEmail", ctx, email) + ret0, _ := ret[0].(*entity.User) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetByEmail indicates an expected call of GetByEmail. +func (mr *MockUserServiceMockRecorder) GetByEmail(ctx, email any) *MockUserServiceGetByEmailCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetByEmail", reflect.TypeOf((*MockUserService)(nil).GetByEmail), ctx, email) + return &MockUserServiceGetByEmailCall{Call: call} +} + +// MockUserServiceGetByEmailCall wrap *gomock.Call +type MockUserServiceGetByEmailCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockUserServiceGetByEmailCall) Return(arg0 *entity.User, arg1 error) *MockUserServiceGetByEmailCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockUserServiceGetByEmailCall) Do(f func(context.Context, string) (*entity.User, error)) *MockUserServiceGetByEmailCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockUserServiceGetByEmailCall) DoAndReturn(f func(context.Context, string) (*entity.User, error)) *MockUserServiceGetByEmailCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// MockKeyring is a mock of Keyring interface. +type MockKeyring struct { + ctrl *gomock.Controller + recorder *MockKeyringMockRecorder + isgomock struct{} +} + +// MockKeyringMockRecorder is the mock recorder for MockKeyring. +type MockKeyringMockRecorder struct { + mock *MockKeyring +} + +// NewMockKeyring creates a new mock instance. +func NewMockKeyring(ctrl *gomock.Controller) *MockKeyring { + mock := &MockKeyring{ctrl: ctrl} + mock.recorder = &MockKeyringMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockKeyring) EXPECT() *MockKeyringMockRecorder { + return m.recorder +} + +// WrapDEK mocks base method. +func (m *MockKeyring) WrapDEK(dek []byte) ([]byte, string, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "WrapDEK", dek) + ret0, _ := ret[0].([]byte) + ret1, _ := ret[1].(string) + ret2, _ := ret[2].(error) + return ret0, ret1, ret2 +} + +// WrapDEK indicates an expected call of WrapDEK. +func (mr *MockKeyringMockRecorder) WrapDEK(dek any) *MockKeyringWrapDEKCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WrapDEK", reflect.TypeOf((*MockKeyring)(nil).WrapDEK), dek) + return &MockKeyringWrapDEKCall{Call: call} +} + +// MockKeyringWrapDEKCall wrap *gomock.Call +type MockKeyringWrapDEKCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockKeyringWrapDEKCall) Return(wrapped []byte, kekID string, err error) *MockKeyringWrapDEKCall { + c.Call = c.Call.Return(wrapped, kekID, err) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockKeyringWrapDEKCall) Do(f func([]byte) ([]byte, string, error)) *MockKeyringWrapDEKCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockKeyringWrapDEKCall) DoAndReturn(f func([]byte) ([]byte, string, error)) *MockKeyringWrapDEKCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// MockCipher is a mock of Cipher interface. +type MockCipher struct { + ctrl *gomock.Controller + recorder *MockCipherMockRecorder + isgomock struct{} +} + +// MockCipherMockRecorder is the mock recorder for MockCipher. +type MockCipherMockRecorder struct { + mock *MockCipher +} + +// NewMockCipher creates a new mock instance. +func NewMockCipher(ctrl *gomock.Controller) *MockCipher { + mock := &MockCipher{ctrl: ctrl} + mock.recorder = &MockCipherMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockCipher) EXPECT() *MockCipherMockRecorder { + return m.recorder +} + +// Encrypt mocks base method. +func (m *MockCipher) Encrypt(dek, plaintext, aad []byte) ([]byte, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Encrypt", dek, plaintext, aad) + ret0, _ := ret[0].([]byte) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Encrypt indicates an expected call of Encrypt. +func (mr *MockCipherMockRecorder) Encrypt(dek, plaintext, aad any) *MockCipherEncryptCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Encrypt", reflect.TypeOf((*MockCipher)(nil).Encrypt), dek, plaintext, aad) + return &MockCipherEncryptCall{Call: call} +} + +// MockCipherEncryptCall wrap *gomock.Call +type MockCipherEncryptCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockCipherEncryptCall) Return(arg0 []byte, arg1 error) *MockCipherEncryptCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockCipherEncryptCall) Do(f func([]byte, []byte, []byte) ([]byte, error)) *MockCipherEncryptCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockCipherEncryptCall) DoAndReturn(f func([]byte, []byte, []byte) ([]byte, error)) *MockCipherEncryptCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// MockTaskScheduler is a mock of TaskScheduler interface. +type MockTaskScheduler struct { + ctrl *gomock.Controller + recorder *MockTaskSchedulerMockRecorder + isgomock struct{} +} + +// MockTaskSchedulerMockRecorder is the mock recorder for MockTaskScheduler. +type MockTaskSchedulerMockRecorder struct { + mock *MockTaskScheduler +} + +// NewMockTaskScheduler creates a new mock instance. +func NewMockTaskScheduler(ctrl *gomock.Controller) *MockTaskScheduler { + mock := &MockTaskScheduler{ctrl: ctrl} + mock.recorder = &MockTaskSchedulerMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockTaskScheduler) EXPECT() *MockTaskSchedulerMockRecorder { + return m.recorder +} + +// Schedule mocks base method. +func (m *MockTaskScheduler) Schedule(ctx context.Context, taskType string, payload any, idempotencyKey string) (uuid.UUID, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Schedule", ctx, taskType, payload, idempotencyKey) + ret0, _ := ret[0].(uuid.UUID) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Schedule indicates an expected call of Schedule. +func (mr *MockTaskSchedulerMockRecorder) Schedule(ctx, taskType, payload, idempotencyKey any) *MockTaskSchedulerScheduleCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Schedule", reflect.TypeOf((*MockTaskScheduler)(nil).Schedule), ctx, taskType, payload, idempotencyKey) + return &MockTaskSchedulerScheduleCall{Call: call} +} + +// MockTaskSchedulerScheduleCall wrap *gomock.Call +type MockTaskSchedulerScheduleCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockTaskSchedulerScheduleCall) Return(arg0 uuid.UUID, arg1 error) *MockTaskSchedulerScheduleCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockTaskSchedulerScheduleCall) Do(f func(context.Context, string, any, string) (uuid.UUID, error)) *MockTaskSchedulerScheduleCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockTaskSchedulerScheduleCall) DoAndReturn(f func(context.Context, string, any, string) (uuid.UUID, error)) *MockTaskSchedulerScheduleCall { + c.Call = c.Call.DoAndReturn(f) + return c +} diff --git a/internal/services/otp/prune.go b/internal/services/otp/prune.go new file mode 100644 index 00000000..328874d7 --- /dev/null +++ b/internal/services/otp/prune.go @@ -0,0 +1,98 @@ +package otp + +import ( + "context" + "time" + + "github.com/ruko1202/xlog" + "github.com/ruko1202/xlog/xfield" + + "github.com/ruko1202/maintmode/internal/utils/xtime" +) + +// maxPruneBatches caps how many batches one Prune call runs, so a single sweep +// can never loop unbounded. The periodic job runs again on the next tick, so any +// leftover eligible rows are drained then; this only bounds per-invocation work. +const maxPruneBatches = 100 + +// defaultPruneBatchLimit is the per-statement DELETE bound used when the caller +// passes a non-positive batchLimit, so the drain loop's "deleted < batchLimit" +// stop condition stays meaningful. +const defaultPruneBatchLimit = 1000 + +// defaultPruneRetention is the age threshold used when the caller passes a +// non-positive retention. +// +// It is 24h, NOT the 365 days its invitation counterpart uses, and the direction +// of safety is why. A one-time code lives minutes, so a day is already a wide +// margin for looking into a failed sign-in; a year-long fallback would silently +// disable the sweep and let code digests and session nonces pile up -- the exact +// thing this job exists to prevent. It must stay equal to the value shipped in +// deployment config, so that an unset retention behaves like the configured one. +const defaultPruneRetention = 24 * time.Hour + +// Prune deletes one-time codes whose expires_at is older than the retention +// window, in bounded batches. Consumed and merely-expired codes leave through +// this same sweep -- consumed_at is not part of the predicate -- and password +// credentials are never eligible. +// +// batchLimit bounds one DELETE so the per-statement lock footprint stays small; +// Prune loops batches until one comes back short (the table is drained for this +// cutoff) or the per-call batch cap is hit. Both tunables come from the cron task +// payload (derived from config). +// +// The cutoff is computed once at call time so every batch in this sweep targets +// the same instant -- rows aging past the threshold mid-sweep wait for the next +// tick rather than shifting the boundary under the loop. +// +// SAFETY. Coercing a non-positive retention is not cosmetic. The per-code attempt +// ceiling is enforced by the row's continued existence: claimSlot refuses to +// issue a new code while it finds a live row with the attempts exhausted, so +// deleting such a row would hand back a fresh code with a fresh counter and turn +// "five attempts per code" into "five attempts per code, unlimited codes". A +// non-positive retention is the only way to reach a cutoff at or after now, and +// hence the only way to make a live row eligible -- so the coercion below is what +// closes that hole. Every positive retention is safe however small, because a row +// is eligible only once its expires_at is already past. +func (s *Service) Prune(ctx context.Context, retention time.Duration, batchLimit int64) error { + ctx, span := xlog.WithOperationSpan(ctx, "service.OTP.Prune") + defer span.End() + + if batchLimit <= 0 { + batchLimit = defaultPruneBatchLimit + } + if retention <= 0 { + retention = defaultPruneRetention + } + + cutoff := xtime.UTCNow().Add(-retention) + + var total int64 + for range maxPruneBatches { + deleted, err := s.store.PruneOTPExpiredBefore(ctx, cutoff, batchLimit) + if err != nil { + xlog.Error(ctx, "failed to prune expired one-time codes batch", + xfield.Time("cutoff", cutoff), + xfield.Int64("prunedSoFar", total), + xfield.Error(err), + ) + return err + } + + total += deleted + if deleted < batchLimit { + break + } + } + + // Logged unconditionally, including a zero-row sweep -- a knowing divergence + // from the invitation sweep, which logs only when it deleted something. This + // job has no metric, so the line is the only evidence it ran: without it a + // cron that stopped firing looks exactly like a cron with nothing to do. + xlog.Info(ctx, "pruned expired one-time codes", + xfield.Int64("count", total), + xfield.Time("cutoff", cutoff), + ) + + return nil +} diff --git a/internal/services/otp/prune_internal_test.go b/internal/services/otp/prune_internal_test.go new file mode 100644 index 00000000..6219b84f --- /dev/null +++ b/internal/services/otp/prune_internal_test.go @@ -0,0 +1,197 @@ +package otp + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/ruko1202/xlog" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" + + mock_otp "github.com/ruko1202/maintmode/internal/pkg/generated/mocks/services/otp" + "github.com/ruko1202/maintmode/internal/utils/xtime" +) + +// The drain loop is exercised over a mocked store: the batch sizes a real +// database would return are exactly what these tests need to script, and the +// predicate itself is covered by the store's own integration tests. +func newPruneService(t *testing.T) (*Service, *mock_otp.MockStore) { + t.Helper() + + store := mock_otp.NewMockStore(gomock.NewController(t)) + + return &Service{store: store}, store +} + +func TestPrune_DrainsUntilShortBatch(t *testing.T) { + t.Parallel() + + svc, store := newPruneService(t) + + gomock.InOrder( + store.EXPECT().PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), int64(2)).Return(int64(2), nil), + store.EXPECT().PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), int64(2)).Return(int64(2), nil), + // A batch shorter than the limit means the table is drained for this + // cutoff, so the loop must stop here rather than call again. + store.EXPECT().PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), int64(2)).Return(int64(1), nil), + ) + + require.NoError(t, svc.Prune(context.Background(), 24*time.Hour, 2)) +} + +// TestPrune_StopsAtBatchCap pins that one invocation cannot loop unbounded: with +// every batch coming back full, the loop still ends after maxPruneBatches. +func TestPrune_StopsAtBatchCap(t *testing.T) { + t.Parallel() + + svc, store := newPruneService(t) + + store.EXPECT(). + PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), int64(10)). + Return(int64(10), nil). + Times(maxPruneBatches) + + require.NoError(t, svc.Prune(context.Background(), 24*time.Hour, 10)) +} + +// TestPrune_UsesOneCutoffForEveryBatch pins that the boundary is computed once, +// so rows aging mid-sweep wait for the next tick instead of shifting the cutoff +// under the loop. +func TestPrune_UsesOneCutoffForEveryBatch(t *testing.T) { + t.Parallel() + + svc, store := newPruneService(t) + + var seen []time.Time + store.EXPECT(). + PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, cutoff time.Time, limit int64) (int64, error) { + seen = append(seen, cutoff) + if len(seen) < 3 { + return limit, nil + } + return 0, nil + }). + Times(3) + + require.NoError(t, svc.Prune(context.Background(), 24*time.Hour, 5)) + require.Len(t, seen, 3) + require.Equal(t, seen[0], seen[1]) + require.Equal(t, seen[0], seen[2]) +} + +// TestPrune_CutoffIsAlwaysInThePast is the safety property: whatever retention +// reaches Prune -- including the non-positive values the fallback absorbs -- the +// store must never be handed a cutoff at or after now. A future cutoff would make +// a live, attempt-exhausted code eligible and reset the guess ceiling that +// claimSlot enforces by the row's continued existence. +func TestPrune_CutoffIsAlwaysInThePast(t *testing.T) { + t.Parallel() + + for _, retention := range []time.Duration{ + -365 * 24 * time.Hour, + -time.Second, + 0, + time.Minute, + 24 * time.Hour, + } { + t.Run(retention.String(), func(t *testing.T) { + t.Parallel() + + svc, store := newPruneService(t) + + var got time.Time + store.EXPECT(). + PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, cutoff time.Time, _ int64) (int64, error) { + got = cutoff + return 0, nil + }) + + require.NoError(t, svc.Prune(context.Background(), retention, 10)) + require.True(t, got.Before(xtime.UTCNow()), + "cutoff %s must be strictly in the past", got) + }) + } +} + +// TestPrune_NonPositiveTunablesFallBack pins both fallbacks. The retention one +// matters most: a non-positive value must become the 24h default rather than a +// cutoff at or past now. +func TestPrune_NonPositiveTunablesFallBack(t *testing.T) { + t.Parallel() + + svc, store := newPruneService(t) + + var gotCutoff time.Time + var gotLimit int64 + store.EXPECT(). + PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, cutoff time.Time, limit int64) (int64, error) { + gotCutoff, gotLimit = cutoff, limit + return 0, nil + }) + + before := xtime.UTCNow() + require.NoError(t, svc.Prune(context.Background(), 0, 0)) + + require.Equal(t, int64(defaultPruneBatchLimit), gotLimit) + // The cutoff must sit a default-retention behind now, give or take the + // microseconds the call itself takes. + require.WithinDuration(t, before.Add(-defaultPruneRetention), gotCutoff, time.Minute) +} + +func TestPrune_PropagatesStoreError(t *testing.T) { + t.Parallel() + + svc, store := newPruneService(t) + + wantErr := errors.New("delete failed") + store.EXPECT(). + PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), gomock.Any()). + Return(int64(0), wantErr) + + require.ErrorIs(t, svc.Prune(context.Background(), 24*time.Hour, 10), wantErr) +} + +// observedPruneCtx returns a context whose logger writes into the returned sink, +// mirroring the seam bootstrapauth's tests use. +func observedPruneCtx(t *testing.T) (context.Context, *observer.ObservedLogs) { + t.Helper() + + core, logs := observer.New(zapcore.DebugLevel) + + return xlog.ContextWithLogger(context.Background(), xlog.NewZapAdapter(zap.New(core))), logs +} + +// TestPrune_LogsCompletionEvenWhenNothingDeleted defends the one place this +// sweep knowingly diverges from its invitation sibling, which wraps the same +// line in `if total > 0`. +// +// The job carries no metric, so this line is the only evidence it ran. Restoring +// the sibling's condition in the name of consistency would make a cron that +// stopped firing indistinguishable from a cron with nothing to collect -- and +// with nothing asserting on it, that regression would be silent. +func TestPrune_LogsCompletionEvenWhenNothingDeleted(t *testing.T) { + t.Parallel() + + svc, store := newPruneService(t) + store.EXPECT(). + PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), gomock.Any()). + Return(int64(0), nil) + + ctx, logs := observedPruneCtx(t) + require.NoError(t, svc.Prune(ctx, 24*time.Hour, 10)) + + done := logs.FilterMessage("pruned expired one-time codes").All() + require.Len(t, done, 1, "a zero-row sweep must still report that it ran") + require.Equal(t, zapcore.InfoLevel, done[0].Level) + + require.Equal(t, int64(0), done[0].ContextMap()["count"], + "the count field is what makes the line useful") +} diff --git a/internal/services/otp/service.go b/internal/services/otp/service.go index d41f440b..7521d226 100644 --- a/internal/services/otp/service.go +++ b/internal/services/otp/service.go @@ -20,7 +20,7 @@ import ( ) // Store is the credential persistence this service needs. Defined consumer-side -// so the service depends on the three operations it performs, not on the whole +// so the service depends only on the operations it performs, not on the whole // store, and so tests can substitute a fake. type Store interface { Create(ctx context.Context, cred *entity.AuthCredential) (*entity.AuthCredential, error) @@ -32,6 +32,10 @@ type Store interface { // is no lock to hold and nothing for it to protect against. GetUnconsumedOTPByUserID(ctx context.Context, userID uuid.UUID) (*entity.AuthCredential, error) ClaimOTPAttempt(ctx context.Context, id uuid.UUID, maxAttempts int16) (bool, error) + // PruneOTPExpiredBefore backs the retention sweep -- the one operation here + // that is not part of a sign-in, since it deletes spent codes rather than + // reading or retiring one. + PruneOTPExpiredBefore(ctx context.Context, cutoff time.Time, limit int64) (int64, error) } // UserService resolves the address to a user. Only the lookup is needed. diff --git a/internal/storages/authcredentials/main_test.go b/internal/storages/authcredentials/main_test.go index 1e0a7657..ffd30239 100644 --- a/internal/storages/authcredentials/main_test.go +++ b/internal/storages/authcredentials/main_test.go @@ -73,6 +73,60 @@ func makeOTP(ctx context.Context, t *testing.T, userID uuid.UUID) *entity.AuthCr return cred } +// makeOTPExpiringAt inserts a one-time code with a caller-chosen expiry, so a +// test can place a row on either side of a prune cutoff. Create inserts +// expires_at straight from the entity, so no back-dating UPDATE is needed here — +// unlike user_invitations, whose created_at is stamped server-side. +func makeOTPExpiringAt( + ctx context.Context, + t *testing.T, + userID uuid.UUID, + expiresAt time.Time, +) *entity.AuthCredential { + t.Helper() + + nonce := uuid.NewString() + + cred, err := store.Create(ctx, &entity.AuthCredential{ + UserID: userID, + Kind: entity.AuthCredentialKindOTP, + SecretHash: uuid.NewString(), + ExpiresAt: &expiresAt, + SessionNonce: &nonce, + }) + require.NoError(t, err) + require.NotNil(t, cred) + + return cred +} + +// makePasswordExpiringAt inserts a password credential that carries an expiry. +// +// Nothing in the schema forbids one — no CHECK ties kind='password' to a NULL +// expires_at — and that is exactly why this fixture exists: it is the only row +// shape that proves the prune's kind guard does real work. A password row with a +// NULL expiry cannot prove it, because NULL < cutoff is never true and such a row +// survives whether or not the guard is there. +func makePasswordExpiringAt( + ctx context.Context, + t *testing.T, + userID uuid.UUID, + expiresAt time.Time, +) *entity.AuthCredential { + t.Helper() + + cred, err := store.Create(ctx, &entity.AuthCredential{ + UserID: userID, + Kind: entity.AuthCredentialKindPassword, + SecretHash: "$argon2id$v=19$m=65536,t=3,p=4$" + uuid.NewString(), + ExpiresAt: &expiresAt, + }) + require.NoError(t, err) + require.NotNil(t, cred) + + return cred +} + // makePassword inserts a password credential for the user. func makePassword(ctx context.Context, t *testing.T, userID uuid.UUID) *entity.AuthCredential { t.Helper() diff --git a/internal/storages/authcredentials/prune.go b/internal/storages/authcredentials/prune.go new file mode 100644 index 00000000..82ba4f49 --- /dev/null +++ b/internal/storages/authcredentials/prune.go @@ -0,0 +1,70 @@ +package authcredentials + +import ( + "context" + "time" + + "github.com/go-jet/jet/v2/postgres" + "github.com/ruko1202/xlog" + + "github.com/ruko1202/maintmode/internal/entity" + "github.com/ruko1202/maintmode/internal/pkg/generated/maintmode/public/table" +) + +// PruneOTPExpiredBefore deletes up to limit one-time codes whose expires_at is +// strictly before cutoff, returning how many it removed. It is the single batch +// behind the retention sweep: the caller loops until a batch deletes fewer than +// limit. +// +// Age is measured on expires_at, not created_at. auth_credentials_otp_expiry_idx +// is the only index on this table and it is on expires_at; for a one-time code +// the two columns differ by exactly the TTL, so nothing is lost by preferring +// the indexed one. +// +// consumed_at is deliberately absent from the predicate. A consumed code still +// carries an expiry and ages out through this same sweep, so no second branch +// (and no second tunable) is needed to collect it. +// +// The `kind = 'otp'` conjunct is load-bearing twice, and must not be "simplified" +// away on the grounds that password rows have a NULL expiry: +// +// - Correctness. Nothing in the schema ties kind='password' to a NULL +// expires_at — there is no such CHECK — so the NULL barrier is an observation +// about today's write paths, not an invariant. A password row that ever +// gained an expiry would be deleted by a predicate resting on NULL alone. +// - The query plan. The index is PARTIAL (WHERE kind = 'otp'), and Postgres can +// only use it if the query's predicate implies the index's. Drop this literal +// and the sweep silently degrades to a sequential scan plus sort. No test can +// observe that, which is why it is written down here. +// +// ORDER BY expires_at ASC matches the index's own direction, so it is served +// directly rather than by a backward scan. +// +// Postgres has no DELETE ... LIMIT, so the batch is bounded by an id-subquery: +// pick the oldest `limit` eligible ids, then delete exactly those. Bounding each +// batch keeps the per-statement lock footprint small on a table that takes a row +// per sign-in attempt. +func (s *Store) PruneOTPExpiredBefore(ctx context.Context, cutoff time.Time, limit int64) (int64, error) { + ctx, span := xlog.WithOperationSpan(ctx, "store.AuthCredentials.PruneOTPExpiredBefore") + defer span.End() + + expired := table.AuthCredentials. + SELECT(table.AuthCredentials.ID). + WHERE( + table.AuthCredentials.Kind.EQ(postgres.String(string(entity.AuthCredentialKindOTP))). + AND(table.AuthCredentials.ExpiresAt.LT(postgres.TimestampzT(cutoff))), + ). + ORDER_BY(table.AuthCredentials.ExpiresAt.ASC()). + LIMIT(limit) + + stmt := table.AuthCredentials. + DELETE(). + WHERE(table.AuthCredentials.ID.IN(expired)) + + res, err := stmt.ExecContext(ctx, s.db.Executor(ctx)) + if err != nil { + return 0, err + } + + return res.RowsAffected() +} diff --git a/internal/storages/authcredentials/prune_test.go b/internal/storages/authcredentials/prune_test.go new file mode 100644 index 00000000..8f28ac3c --- /dev/null +++ b/internal/storages/authcredentials/prune_test.go @@ -0,0 +1,211 @@ +package authcredentials + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + + "github.com/ruko1202/maintmode/internal/utils/xtime" +) + +// Two separate rules govern this file, and only the first is about assertions. +// +// 1. Every assertion counts rows for one freshly created user. The suite runs +// with -count 2 against a shared database, so a table-wide count would race +// whatever else is running. +// +// 2. Every cutoff passed to the sweep stays in the PAST, and close to it. This +// one is easy to miss because it is not about what a test observes but about +// what its DELETE reaches: the statement is bounded by expires_at and limit +// alone -- never by user_id -- so a distant cutoff deletes rows belonging to +// whatever else shares the database, including other packages running +// concurrently under `-p 2`. Scoping the assertions does not make a sweeping +// cutoff safe. +func credentialExists(ctx context.Context, t *testing.T, id uuid.UUID) bool { + t.Helper() + + var n int + require.NoError(t, db.GetContext(ctx, &n, + "SELECT count(*) FROM auth_credentials WHERE id = $1", id)) + + return n > 0 +} + +func countForUser(ctx context.Context, t *testing.T, userID uuid.UUID) int { + t.Helper() + + var n int + require.NoError(t, db.GetContext(ctx, &n, + "SELECT count(*) FROM auth_credentials WHERE user_id = $1", userID)) + + return n +} + +// TestPruneOTPExpiredBefore_DeletesOnlyRowsPastCutoff covers the core age +// boundary: a code expired before the cutoff goes, one expiring after it stays. +func TestPruneOTPExpiredBefore_DeletesOnlyRowsPastCutoff(t *testing.T) { + ctx := context.Background() + now := xtime.UTCNow() + + // Two users, because one live OTP per user is a partial-unique invariant. + oldUser := makeUser(ctx, t) + freshUser := makeUser(ctx, t) + + old := makeOTPExpiringAt(ctx, t, oldUser.ID, now.Add(-48*time.Hour)) + fresh := makeOTPExpiringAt(ctx, t, freshUser.ID, now.Add(10*time.Minute)) + + deleted, err := store.PruneOTPExpiredBefore(ctx, now.Add(-24*time.Hour), 100) + require.NoError(t, err) + require.GreaterOrEqual(t, deleted, int64(1)) + + require.False(t, credentialExists(ctx, t, old.ID), "expired code must be deleted") + require.True(t, credentialExists(ctx, t, fresh.ID), "unexpired code must survive") +} + +// TestPruneOTPExpiredBefore_SparesLiveCode is the security criterion, and the +// reason it stands apart from the boundary test above. +// +// The per-code attempt ceiling is enforced by the row's continued existence: +// claimSlot refuses to issue a new code while it finds a live row with the +// attempts exhausted. Delete such a row and the next request takes the +// first-request branch and hands back a fresh code with a fresh counter, turning +// "five attempts per code" into "five attempts per code, unlimited codes". +// +// The property is checked at the near edge of the permitted range, not only at +// the 24h default: a 1-minute retention is the smallest cutoff config allows, +// and if a future-dated row survives that, it survives every larger one. The +// attempts counter is set to the ceiling so the fixture is exactly the row the +// bypass would need. +func TestPruneOTPExpiredBefore_SparesLiveCode(t *testing.T) { + ctx := context.Background() + now := xtime.UTCNow() + + user := makeUser(ctx, t) + live := makeOTPExpiringAt(ctx, t, user.ID, now.Add(5*time.Minute)) + + // Exhaust the guess ceiling: this is the state the barrier exists to hold. + claimed, err := store.ClaimOTPAttempt(ctx, live.ID, 1) + require.NoError(t, err) + require.True(t, claimed) + + for _, retention := range []time.Duration{time.Minute, 24 * time.Hour} { + _, err := store.PruneOTPExpiredBefore(ctx, now.Add(-retention), 100) + require.NoError(t, err) + require.True(t, credentialExists(ctx, t, live.ID), + "a live code must survive a sweep at retention %s", retention) + } +} + +// TestPruneOTPExpiredBefore_SparesPasswordWithNullExpiry is the NULL barrier: +// a password row has no expiry, and NULL < cutoff is never true. +// +// The cutoff is in the PAST, deliberately. A far-future cutoff would demonstrate +// the same property, but it would also make every OTP row in the database +// eligible -- and the DELETE's reach is not scoped by user_id even though this +// file's assertions are. Under `-p 2` another package's DB-backed suite can be +// running against the same table, so a sweeping cutoff would delete fixtures out +// from under it: a flaky red in services/otp's claim tests, and worse, a silent +// green in TestVerify_ExpiredCodeIsRetired, whose "row is gone" assertion cannot +// tell a retired row from one this sweep ate. A past cutoff proves the barrier +// just as well, because the password row would be caught by it if it had any +// expiry at all. +func TestPruneOTPExpiredBefore_SparesPasswordWithNullExpiry(t *testing.T) { + ctx := context.Background() + + user := makeUser(ctx, t) + pwd := makePassword(ctx, t, user.ID) + + _, err := store.PruneOTPExpiredBefore(ctx, xtime.UTCNow().Add(-24*time.Hour), 100) + require.NoError(t, err) + + require.True(t, credentialExists(ctx, t, pwd.ID)) + require.Equal(t, 1, countForUser(ctx, t, user.ID)) +} + +// TestPruneOTPExpiredBefore_SparesPasswordWithExpiry is the kind guard, and the +// only test that can prove it exists. Unlike the NULL-expiry case above, this row +// WOULD be deleted by a predicate that dropped `kind = 'otp'`. +func TestPruneOTPExpiredBefore_SparesPasswordWithExpiry(t *testing.T) { + ctx := context.Background() + now := xtime.UTCNow() + + user := makeUser(ctx, t) + pwd := makePasswordExpiringAt(ctx, t, user.ID, now.Add(-72*time.Hour)) + + deleted, err := store.PruneOTPExpiredBefore(ctx, now.Add(-24*time.Hour), 100) + require.NoError(t, err) + require.GreaterOrEqual(t, deleted, int64(0)) + + require.True(t, credentialExists(ctx, t, pwd.ID), + "a password row must survive even with an expiry older than the cutoff") + require.Equal(t, 1, countForUser(ctx, t, user.ID)) +} + +// TestPruneOTPExpiredBefore_ConsumedRowsFollowTheSameThreshold pins that +// consumed_at plays no part in the predicate: a consumed code leaves on age +// alone, and a consumed-but-recent code stays. +func TestPruneOTPExpiredBefore_ConsumedRowsFollowTheSameThreshold(t *testing.T) { + ctx := context.Background() + now := xtime.UTCNow() + + agedUser := makeUser(ctx, t) + recentUser := makeUser(ctx, t) + + aged := makeOTPExpiringAt(ctx, t, agedUser.ID, now.Add(-48*time.Hour)) + recent := makeOTPExpiringAt(ctx, t, recentUser.ID, now.Add(-time.Minute)) + + // consumed_at is excluded from Create's column list by design, so the state + // is reached through ConsumeOTP. It does not check expiry, so an already-aged + // row can be consumed. + for _, id := range []uuid.UUID{aged.ID, recent.ID} { + ok, err := store.ConsumeOTP(ctx, id) + require.NoError(t, err) + require.True(t, ok) + } + + _, err := store.PruneOTPExpiredBefore(ctx, now.Add(-24*time.Hour), 100) + require.NoError(t, err) + + require.False(t, credentialExists(ctx, t, aged.ID), + "a consumed code past the cutoff leaves through the same sweep") + require.True(t, credentialExists(ctx, t, recent.ID), + "being consumed grants no early deletion") +} + +// TestPruneOTPExpiredBefore_RespectsLimit pins that one call never removes more +// than the batch bound, which is what makes the service's drain loop meaningful. +func TestPruneOTPExpiredBefore_RespectsLimit(t *testing.T) { + ctx := context.Background() + now := xtime.UTCNow() + + // One OTP per user, because one live OTP per user is a partial-unique + // invariant, so a surviving id is exactly a surviving row. + ids := make([]uuid.UUID, 0, 3) + for range 3 { + u := makeUser(ctx, t) + ids = append(ids, makeOTPExpiringAt(ctx, t, u.ID, now.Add(-48*time.Hour)).ID) + } + + deleted, err := store.PruneOTPExpiredBefore(ctx, now.Add(-24*time.Hour), 2) + require.NoError(t, err) + require.LessOrEqual(t, deleted, int64(2), "one call must not exceed the limit") + + remaining := 0 + for _, id := range ids { + if credentialExists(ctx, t, id) { + remaining++ + } + } + require.GreaterOrEqual(t, remaining, 1, + "with 3 eligible rows and a limit of 2, at least one must be left for the next batch") + + // Drain the rest so the fixtures do not linger. + _, err = store.PruneOTPExpiredBefore(ctx, now.Add(-24*time.Hour), 100) + require.NoError(t, err) + for _, id := range ids { + require.False(t, credentialExists(ctx, t, id)) + } +}