---
phase: 06-background-worker
plan: 02
type: execute
wave: 2
depends_on:
  - 06-01
files_modified:
  - backend/cmd/worker/main.go
  - backend/justfile
  - backend/README.md
autonomous: true
requirements:
  - WORK-01
  - WORK-02
  - WORK-03

must_haves:
  truths:
    - "D-02: cmd/worker/main.go runs rivermigrate at startup before constructing river.Client"
    - "D-05: cmd/worker/main.go registers HeartbeatWorker and OrphanCleanupWorker via river.AddWorker"
    - "D-07: cmd/worker/main.go passes slog.Default() as river.Config.Logger (single river.Client)"
    - "D-03: cmd/worker/main.go starts HeartbeatWorker as a PeriodicJob every 1 minute with RunOnStart: true (periodic jobs only)"
    - "D-03: cmd/worker/main.go starts OrphanCleanupWorker as a PeriodicJob every 1 hour with RunOnStart: false"
    - "D-07: Graceful shutdown calls riverClient.StopAndCancel with a 10-second timeout context"
    - "go build ./cmd/worker/... exits 0"
    - "D-08: just worker target exists in justfile; README.md documents single-worker constraint"
    - "README.md documents how to run the worker locally alongside the web binary"
  artifacts:
    - path: "backend/cmd/worker/main.go"
      provides: "full river wiring — rivermigrate + river.Client + periodic jobs + graceful shutdown"
      contains: "rivermigrate.New"
    - path: "backend/justfile"
      provides: "worker target for local dev"
      contains: "just worker"
    - path: "backend/README.md"
      provides: "worker local dev instructions"
      contains: "just worker"
  key_links:
    - from: "backend/cmd/worker/main.go"
      to: "backend/internal/jobs/"
      via: "jobs.HeartbeatWorker, jobs.NewOrphanCleanupWorker, jobs.SlogErrorHandler"
      pattern: "jobs\\."
    - from: "backend/cmd/worker/main.go"
      to: "rivermigrate"
      via: "rivermigrate.New(riverpgxv5.New(pool), nil).Migrate(ctx, rivermigrate.DirectionUp, nil)"
      pattern: "rivermigrate\\.New"
    - from: "backend/cmd/worker/main.go"
      to: "river.Client"
      via: "river.NewClient(riverpgxv5.New(pool), &river.Config{...})"
      pattern: "river\\.NewClient"
---

<objective>
Replace the noop body in `cmd/worker/main.go` with the full river runtime: rivermigrate at startup, S3 store init, signal context construction, worker registration, river.Client construction and start, block on signal, StopAndCancel graceful shutdown. Add the `just worker` justfile target with MinIO dev defaults. Document the worker in README.md.

Purpose: Closes WORK-01 (worker binary runs and processes jobs), WORK-02 (heartbeat job observable in logs within 1 minute of startup), WORK-03 (graceful shutdown matching web binary pattern).
Output: Fully functional `cmd/worker` binary; `just worker` runs it locally; README documents it.
</objective>

<execution_context>
@/Users/arthur.belleville/Documents/perso/projects/xtablo-source/.claude/get-shit-done/workflows/execute-plan.md
@/Users/arthur.belleville/Documents/perso/projects/xtablo-source/.claude/get-shit-done/templates/summary.md
</execution_context>

<context>
@.planning/ROADMAP.md
@.planning/phases/06-background-worker/06-CONTEXT.md
@.planning/phases/06-background-worker/06-RESEARCH.md
@.planning/phases/06-background-worker/06-PATTERNS.md
@.planning/phases/06-background-worker/06-01-SUMMARY.md

<interfaces>
From backend/internal/jobs/ (created in Plan 01):
  jobs.HeartbeatArgs{}          — Kind() = "heartbeat"
  jobs.HeartbeatWorker{}        — embeds river.WorkerDefaults[HeartbeatArgs]
  jobs.OrphanCleanupArgs{}      — Kind() = "orphan_file_cleanup"
  jobs.NewOrphanCleanupWorker(pool *pgxpool.Pool, store files.FileStorer) *OrphanCleanupWorker
  jobs.SlogErrorHandler{}       — implements river.ErrorHandler

From backend/internal/files/store.go:
  files.NewStore(ctx, endpoint, bucket, region, accessKey, secretKey, usePathStyle bool) (*Store, error)
  — S3 env var names: S3_ENDPOINT, S3_BUCKET, S3_REGION, S3_ACCESS_KEY, S3_SECRET_KEY, S3_USE_PATH_STYLE

From backend/internal/db/pool.go:
  db.NewPool(ctx context.Context, dsn string) (*pgxpool.Pool, error)

From backend/cmd/worker/main.go (current skeleton — to be fully replaced):
  package main
  — already has: env/dsn loading, slog.SetDefault, db.NewPool, signal.NotifyContext, pool.Close pattern

From rivermigrate (per RESEARCH.md Pattern 2):
  rivermigrate.New(riverpgxv5.New(pool), nil) → migrator
  migrator.Migrate(ctx, rivermigrate.DirectionUp, nil) → (res, err)
  res.Versions — range to log each applied version

From river (per RESEARCH.md Pattern 3):
  river.NewWorkers() → workers
  river.AddWorker(workers, &HeartbeatWorker{})
  river.NewClient(riverpgxv5.New(pool), &river.Config{
    Logger:       slog.Default(),
    Workers:      workers,
    Queues:       map[string]river.QueueConfig{river.QueueDefault: {MaxWorkers: 10}},
    ErrorHandler: &SlogErrorHandler{},
    PeriodicJobs: [...],
  }) → (riverClient, err)
  riverClient.Start(ctx) → err   // MUST use live signal context, not background
  riverClient.StopAndCancel(timeoutCtx) → err

CRITICAL ORDERING CONSTRAINT (RESEARCH Pitfall 2 + PATTERNS.md):
  1. context.Background() for startup I/O (pool, rivermigrate, S3)
  2. signal.NotifyContext AFTER all startup I/O succeeds
  3. riverClient.Start(sigCtx) — passes the signal context; jobs inherit it
  4. <-sigCtx.Done() — wait for SIGINT/SIGTERM
  5. StopAndCancel with a fresh context.WithTimeout(context.Background(), 10*time.Second)
  6. pool.Close()
  Do NOT create signal.NotifyContext before rivermigrate or S3 init.
</interfaces>
</context>

<tasks>

<task type="auto">
  <name>Task 1: Replace cmd/worker/main.go with full river wiring</name>
  <files>backend/cmd/worker/main.go</files>
  <read_first>
    - backend/cmd/worker/main.go — read the current skeleton to understand what to preserve and what to replace
    - backend/cmd/web/main.go — S3 env var loading pattern (lines 85-100) and graceful shutdown pattern (lines 144-148) to replicate
    - .planning/phases/06-background-worker/06-RESEARCH.md — full cmd/worker/main.go code example (Pattern 2, 3, 5) and Pitfall 2
    - .planning/phases/06-background-worker/06-PATTERNS.md — critical ordering constraint and pool close ordering
  </read_first>
  <action>
    Replace the entire content of backend/cmd/worker/main.go. The new file must implement the following sequence exactly (per RESEARCH Pattern ordering and PATTERNS.md critical ordering constraint):

    Imports to include: "context", "log/slog", "os", "os/signal", "syscall", "time", "backend/internal/db", "backend/internal/files", "backend/internal/jobs", "backend/internal/web", "github.com/riverqueue/river", "github.com/riverqueue/river/riverdriver/riverpgxv5", "github.com/riverqueue/river/rivermigrate".

    main() function sequence:
    1. Read ENV (default "development") and DATABASE_URL env vars. Call slog.SetDefault(slog.New(web.NewSlogHandler(env, os.Stdout))). Guard on empty DATABASE_URL (slog.Error + os.Exit(1)).

    2. Create startup context: ctx := context.Background() (NOT signal context yet — critical per PATTERNS.md ordering constraint).

    3. Connect DB pool: db.NewPool(ctx, dsn). On error: slog.Error + os.Exit(1). Note: defer pool.Close() is NOT used here — explicit pool.Close() at the end of shutdown sequence (after StopAndCancel) per PATTERNS.md pool close ordering.

    4. Run rivermigrate: rivermigrate.New(riverpgxv5.New(pool), nil). On init error: slog.Error + os.Exit(1). Call .Migrate(ctx, rivermigrate.DirectionUp, nil). On migrate error: slog.Error + os.Exit(1). Range over res.Versions and log each with slog.Info("river migration applied", "version", v.Version).

    5. Init S3 store: read env vars S3_ENDPOINT, S3_BUCKET, S3_ACCESS_KEY, S3_SECRET_KEY, S3_REGION (default "us-east-1" if empty), S3_USE_PATH_STYLE == "true". Call files.NewStore(ctx, endpoint, bucket, region, accessKey, secretKey, usePathStyle). On error: slog.Error + os.Exit(1).

    6. Create signal context AFTER all startup I/O: sigCtx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM). defer stop().

    7. Register workers: workers := river.NewWorkers(). river.AddWorker(workers, &jobs.HeartbeatWorker{}). river.AddWorker(workers, jobs.NewOrphanCleanupWorker(pool, fileStore)).

    8. Construct river client: river.NewClient(riverpgxv5.New(pool), &river.Config{ Logger: slog.Default(), Workers: workers, Queues: map[string]river.QueueConfig{river.QueueDefault: {MaxWorkers: 10}}, ErrorHandler: &jobs.SlogErrorHandler{}, PeriodicJobs: []*river.PeriodicJob{ river.NewPeriodicJob(river.PeriodicInterval(1*time.Minute), func() (river.JobArgs, *river.InsertOpts) { return jobs.HeartbeatArgs{}, nil }, &river.PeriodicJobOpts{RunOnStart: true}), river.NewPeriodicJob(river.PeriodicInterval(1*time.Hour), func() (river.JobArgs, *river.InsertOpts) { return jobs.OrphanCleanupArgs{}, nil }, nil), }, }). On error: slog.Error + os.Exit(1).

    9. Start client: riverClient.Start(sigCtx). On error: slog.Error + os.Exit(1).

    10. Log slog.Info("worker ready").

    11. <-sigCtx.Done(). Log slog.Info("shutting down").

    12. Create stopCtx: context.WithTimeout(context.Background(), 10*time.Second). defer stopCancel(). Call riverClient.StopAndCancel(stopCtx). On error: slog.Warn("river stop did not finish cleanly", "err", err).

    13. pool.Close(). Log slog.Info("shutdown complete").

    The file package comment should read: "Command worker is the background job processor for Xtablo. It runs river periodic jobs against the same Postgres as cmd/web."
  </action>
  <verify>
    <automated>cd /Users/arthur.belleville/Documents/perso/projects/xtablo-source/backend && go build ./cmd/worker/...</automated>
  </verify>
  <done>
    go build ./cmd/worker/... exits 0. The file contains "rivermigrate.New", "river.NewClient", "StopAndCancel", "HeartbeatArgs", "OrphanCleanupArgs".
  </done>
  <acceptance_criteria>
    - `cd backend && go build ./cmd/worker/...` exits 0
    - `grep "rivermigrate.New" backend/cmd/worker/main.go` returns a line
    - `grep "river.NewClient" backend/cmd/worker/main.go` returns a line
    - `grep "StopAndCancel" backend/cmd/worker/main.go` returns a line
    - `grep "HeartbeatArgs" backend/cmd/worker/main.go` returns a line
    - `grep "OrphanCleanupArgs" backend/cmd/worker/main.go` returns a line
    - `grep "RunOnStart.*true" backend/cmd/worker/main.go` returns a line (confirms heartbeat RunOnStart)
    - `grep "slog.Default()" backend/cmd/worker/main.go` returns a line (river Logger wired to slog)
    - `cd backend && go build ./...` exits 0 (whole module compiles)
  </acceptance_criteria>
</task>

<task type="auto">
  <name>Task 2: Add just worker target + document worker in README.md</name>
  <files>backend/justfile, backend/README.md</files>
  <read_first>
    - backend/justfile — read the dev target (lines 111-113) and variable declarations (lines 38-39) to match the style exactly
    - backend/README.md — read the full file to find the right insertion point for the worker section
    - .planning/phases/06-background-worker/06-PATTERNS.md — new worker target style (last Pattern section)
    - .planning/phases/06-background-worker/06-CONTEXT.md — D-08 single-worker constraint to document
  </read_first>
  <action>
    justfile — append the `worker` target after the `dev` target (keep file order: dev, then worker). The target:
    - depends on db-up (same as dev)
    - passes these env vars on separate continuation lines: DATABASE_URL='{{ database_url }}', S3_ENDPOINT='http://localhost:9000', S3_BUCKET='xtablo', S3_REGION='us-east-1', S3_ACCESS_KEY='minioadmin', S3_SECRET_KEY='minioadmin', S3_USE_PATH_STYLE='true'
    - runs: go run ./cmd/worker
    - precede the target with a comment: "# Start the worker binary (development — requires db-up and MinIO running)."
    - Match the env-var-per-continuation-line style of the existing dev target exactly.

    README.md — add a "Running the Worker" section. Insert it after the existing "Development" or "Running" section (whatever comes last before any deployment/production section). The section must include:
    - Command: just worker (requires db-up and just dev or MinIO running separately)
    - What to expect: structured logs appear immediately; "worker ready" log within a few seconds; "worker heartbeat" log appears within ~1 minute of startup (RunOnStart fires immediately on first tick, so first heartbeat appears within seconds)
    - Note on D-08: only one worker process should run at a time for v1; do not run multiple worker instances
    - Graceful shutdown: send SIGINT (Ctrl+C) and observe "shutting down" then "shutdown complete" in logs
    - Observing failed job retries: river logs "job error" with job_id, job_kind, attempt, max_attempts, err fields on each failure; max 25 attempts before discard
  </action>
  <verify>
    <automated>cd /Users/arthur.belleville/Documents/perso/projects/xtablo-source/backend && grep -c "^worker:" justfile && grep -c "just worker" README.md</automated>
  </verify>
  <done>
    justfile contains a `worker:` target. README.md mentions `just worker`. Both grep counts return >= 1.
  </done>
  <acceptance_criteria>
    - `grep "^worker:" backend/justfile` returns exactly 1 line
    - `grep -A 20 "^worker:" backend/justfile | grep -c "S3_ENDPOINT"` returns 1 (S3_ENDPOINT confirmed inside the worker target block)
    - `grep "go run ./cmd/worker" backend/justfile` returns a line inside the worker target
    - `grep "just worker" backend/README.md` returns at least 1 line
    - `grep -i "single.*worker\|one worker\|multiple worker" backend/README.md` returns a line (D-08 documented)
    - `grep "worker heartbeat" backend/README.md` returns a line (observable log message documented)
    - `cd backend && just --list 2>&1 | grep worker` returns a line confirming the target is registered
  </acceptance_criteria>
</task>

</tasks>

<threat_model>
## Trust Boundaries

| Boundary | Description |
|----------|-------------|
| Worker process → Postgres (startup) | rivermigrate acquires advisory lock during migration; cmd/worker must not run concurrently with another instance during migration phase |
| Worker process → S3/MinIO | Orphan cleanup deletes objects using credentials loaded from env; no user-supplied key values reach S3 in Phase 6 |

## STRIDE Threat Register

| Threat ID | Category | Component | Disposition | Mitigation Plan |
|-----------|----------|-----------|-------------|-----------------|
| T-06-05 | Tampering | rivermigrate concurrent execution | mitigate | D-08 documents single-worker constraint; rivermigrate uses advisory lock (pg_advisory_lock) to prevent concurrent migration runs; README documents do-not-run-multiple-instances |
| T-06-06 | Denial of Service | river.Client.Start with already-cancelled context (Pitfall 2) | mitigate | signal.NotifyContext created AFTER all startup I/O per PATTERNS.md ordering constraint; Start receives a live context |
| T-06-07 | Information Disclosure | S3 MinIO credentials in justfile dev target | accept | Dev-only credentials (minioadmin/minioadmin) are the MinIO default; justfile is committed and these are not production secrets; .env.example documents production vars separately |
| T-06-08 | Tampering | Orphan cleanup deletes wrong S3 key | mitigate | Key comes from tablo_files.s3_key column (set at upload time in Phase 5); not reconstructed; only orphan rows (no matching tablo) are targeted by the query |
</threat_model>

<verification>
After Plan 02 is complete:
- `cd backend && go build ./...` exits 0
- `cd backend && go build ./cmd/worker/...` exits 0
- `grep "rivermigrate.New" backend/cmd/worker/main.go` returns a line
- `grep "^worker:" backend/justfile` returns a line
- `cd backend && just --list 2>&1 | grep worker` shows the worker target
- `grep "just worker" backend/README.md` returns a line
</verification>

<success_criteria>
- cmd/worker/main.go compiles with full river wiring
- rivermigrate runs before river.Client starts (verified by code structure)
- Both periodic jobs registered (heartbeat 1min RunOnStart:true, orphan cleanup 1hr)
- Graceful shutdown via StopAndCancel(10s timeout context)
- just worker target present and documented in README
- go build ./... exits 0 — full module clean
</success_criteria>

<output>
After completion, create `.planning/phases/06-background-worker/06-02-SUMMARY.md`
</output>
