# Phase K — Reprocessing Job Runner

**Branch**: `feat/fraud-reprocessing`
**Parent**: `feat/fraud-rules-jube-parity`
**Jube reference**: `Reprocess` admin page + `EntityAnalysisInline*Reprocess` workers.

---

## 1. Goal

Replay archived payloads through the current `MwRisk.Pipeline.run/2` so
operators can validate the effect of a rule edit on historical traffic
before publishing it. Useful for:

- "If I lower the velocity threshold on Rule X, how many new alerts last 7 days?"
- Bulk re-scoring after a sanctions list update.
- Backfilling activation rows for newly-added rules.

Only Activation rules flagged `enable_reprocessing = true` participate in
the replay path (matches Jube semantics).

---

## 2. Architectural pieces

| Concern | Module |
|---|---|
| Job table          | `risk_reprocessing_jobs` (NEW)         |
| Worker             | `MwRisk.ReprocessingWorker` (Oban)     |
| Source of payloads | `risk_activation_watcher.payload_snapshot` + `risk_cases` payload |
| Admin UI           | `ReprocessingLive` at `/admin/fraud/reprocessing` |
| Result sink        | `risk_reprocessing_results` (NEW)      |

---

## 3. Work breakdown

### 3.1 New schemas

**`risk_reprocessing_jobs`** (new migration):
```elixir
add :tenant_id,         :integer, null: false
add :entity_model_id,   :integer, null: false
add :name,              :string,  null: false
add :status,            :string,  null: false, default: "queued"
   # queued | running | completed | failed | cancelled
add :source,            :string,  null: false
   # watcher | cases | upload
add :source_filter,     :map      # JSON: date range, rule ids, entity keys
add :total_count,       :integer, default: 0
add :processed_count,   :integer, default: 0
add :matched_count,     :integer, default: 0
add :error_count,       :integer, default: 0
add :started_at,        :utc_datetime
add :completed_at,      :utc_datetime
add :created_by,        :string
add :version,           :integer, default: 1
timestamps()
```

**`risk_reprocessing_results`** (rolled-up per fired rule):
```elixir
add :job_id,           :integer, null: false
add :activation_rule_id, :integer
add :match_count,      :integer, default: 0
add :new_match_count,  :integer, default: 0    # weren't matched in original run
add :sample_payloads,  :map                    # up to 10 examples
timestamps(updated_at: false)
```

Both indexed on `(tenant_id, ...)`.

### 3.2 Oban worker

`apps/mw_risk/lib/mw_risk/reprocessing_worker.ex`:

```elixir
use Oban.Worker, queue: :reprocessing, max_attempts: 2

def perform(%Oban.Job{args: %{"job_id" => jid}}) do
  job = Reprocessing.get_job!(jid)
  Reprocessing.mark_running(job)

  source_stream(job)
  |> Stream.chunk_every(50)
  |> Stream.each(&process_chunk(&1, job))
  |> Stream.run()

  Reprocessing.mark_completed(job)
end

defp process_chunk(payloads, job) do
  payloads
  |> Task.async_stream(fn pl ->
       Pipeline.run(%Context{tenant_id: job.tenant_id,
                             assigns: %{entity_model_id: job.entity_model_id},
                             request: pl}, reprocessing: true)
     end, max_concurrency: 8, timeout: 5_000)
  |> Enum.each(&record_result(&1, job))
end
```

Key invariants:
- `Pipeline.run(_, reprocessing: true)` **suppresses all side-effect dispatch**
  (no case opens, no notifications, no TTL increments). Outcomes are recorded
  only in `risk_reprocessing_results`.
- Only Activation rules with `enable_reprocessing = true` are evaluated;
  others are short-circuited inside `ActivationEngine` when the
  `reprocessing` flag is set.

### 3.3 Context

`InfraRepo.Risk.Reprocessing`:
- `list_jobs/1`, `get_job!/1`, `create_job/1`, `cancel_job/1`
- `mark_running/1`, `mark_completed/1`, `mark_failed/2`
- `record_result/2`
- `summary(job_id)` → `%{by_rule: %{rule_id => %{matched, new_matched, sample}}}`

### 3.4 LiveView

`/admin/fraud/reprocessing`:
- **New Job** drawer:
  - Name
  - Source (watcher / cases / upload-jsonl)
  - Date range
  - Rule scope (multi-select Activation rules)
  - "Estimate" button — runs a quick `count` query and shows projected total.
- **Jobs** table: name, status (pill), progress bar (`processed_count / total_count`), started/completed, results link.
- **Results** drawer: per-rule match counts, "new matches" highlight, expandable sample payloads (JSON).
- Auto-refresh every 2 s while a job is `running` via PubSub on `"reprocessing:#{tenant_id}"`.

### 3.5 Pipeline plumbing

Add `reprocessing: boolean()` option to `MwRisk.Pipeline.run/2`. When true:
- Pass it through `Context.assigns[:reprocessing]`.
- `ActivationEngine` skips rules with `enable_reprocessing == false`.
- `SideEffectDispatcher.dispatch/2` becomes a no-op (return the effect list
  as `{:ok, effect}` tuples without executing).

### 3.6 Tests

- Worker: 100 payloads through a known rule set, assert counts match.
- Pipeline `reprocessing: true` suppresses dispatch (mock dispatcher must not be called).
- LiveView: create job, simulate worker progress via PubSub, assert UI updates.

### 3.7 Seeds

No data seeded; ship a "Try sample run" button on the empty-state that
synthesises 100 random transactions and runs them through the demo
Activation rules.

---

## 4. Acceptance criteria

- [ ] Job created via UI enqueues an Oban job in the `:reprocessing` queue.
- [ ] Worker processes payloads, never triggers a real side-effect dispatch.
- [ ] Results page shows per-rule match counts within 5 s of completion for a 1k-payload job.
- [ ] Cancelling a running job stops further chunk processing within 2 s.

---

## 5. Out of scope

- Diff view: "rule X used to match 50, now matches 70 — show me the 20 new ones."
- Bulk re-scoring writing back to `risk_cases` (this is *evaluation*, not a destructive replay).
- Streaming reprocessing for > 10 M payloads (use Flow / Broadway when needed).
- Federated reprocessing across multiple tenants in one job.