# Phase 4 — Campaigns Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use
> `superpowers:subagent-driven-development` (recommended) or
> `superpowers:executing-plans` to implement this plan task-by-task. Steps use
> checkbox (`- [ ]`) syntax for tracking.

**Goal:** Build the Campaigns module — create a campaign (template + segment +
per-variable mapping + schedule), resolve a consent-aware audience, send
template messages to many contacts through a rate-limited BullMQ queue with
per-recipient delivery tracking, and pause/resume/cancel/retry — fully replacing
the AiSensy outbound dependency.

**Architecture:** One new NestJS module in `apps/api` —
`CampaignsModule` (`CampaignsService` for CRUD/control, `CampaignSendService`
for audience resolution + recipient-row creation + enqueue, a
`CampaignsController` exposing `/api/campaigns`). One new BullMQ worker
processor in `apps/worker` — `campaign-send.processor.ts`, a pure function with
injected deps that resolves the variable mapping for one recipient and calls the
Phase 1 send path. **Five pure, exhaustively-tested helpers** go in
`packages/shared/src/campaigns`: `resolveVariableMapping` (FR-05.2),
`buildAudienceWhere` (FR-05.4), `aggregateCampaignStats` (FR-05.15),
`isCampaignComplete` / `terminalRecipientStatuses` (FR-05.13), and the
`CampaignVariableMapping` Zod schema. Delivery webhooks for campaign messages
already flow through Phase 1's `processEnvelope`; this phase extends its
`handleStatus` to reflect a campaign message's status onto its
`campaign_recipients` row and recompute the campaign `stats` (FR-05.10). The
React app gains real Campaigns screens (list, builder, detail) behind the
existing `/campaigns` nav placeholder.

**Tech Stack:** NestJS 11 + Fastify, Prisma 7 / PostgreSQL, Zod 3, BullMQ 5,
Vitest 3 on the backend; React 19 + Vite 6 + react-router 7 +
`@tanstack/react-query` 5 on the frontend. No new runtime dependencies.

**References:** `../specs/05-campaigns.md` (authoritative for FR-05.*/AC-05.*),
`../docs/integrations.md` §1.5 (Meta rate limits / messaging tiers), §1.6
(error codes — `130429` rate limit → exponential backoff), `../docs/design.md`
§3.3 ("Outbound campaign" flow), `../packages/db/prisma/schema.prisma`
(authoritative `Campaign`, `CampaignRecipient`, `CampaignStatus`,
`RecipientStatus` models / enums and field names).

**Conventions:**
- Every task ends in a commit (`feat:` / `fix:` / `test:` / `chore:`).
- Tests use Vitest. Run from the repo root with `pnpm test` or per-package with
  `pnpm --filter <pkg> test <pattern>`.
- **All external HTTP/Meta calls are mocked in tests** — never a live send.
  Prisma is mocked with plain `vi.fn()` delegates (see the pattern in
  `apps/api/src/contacts/segments.service.test.ts`). `MessageService` is mocked
  with a `vi.fn()` `send` method; the worker processor takes its collaborators
  (queue, prisma, send) as **injected deps** and is tested as a pure function
  (see `apps/worker/src/processors/flow-dispatch.processor.ts`).
- No secret value is ever committed or logged. There are no new secrets this
  phase — Meta credentials come from the encrypted `connections` store via the
  existing `MetaConfigService` used inside `MessageService`.
- **Docker is broken on this machine.** Verification relies on `pnpm lint`,
  `pnpm typecheck`, `pnpm test`, `pnpm build` and unit tests against mocks. Any
  check needing a live Postgres, live Redis, or live Meta API is marked **"live
  verification deferred"** — it is not a blocker for the phase.
- **Migrations are deferred.** Task 1 inspects the schema; the `Campaign` and
  `CampaignRecipient` models already carry every field this spec needs (see
  "Schema check" below). If a gap is found it is an **additive, nullable**
  column, made in `schema.prisma`, with the SQL generated via
  `prisma migrate diff` and committed as a numbered migration folder, applied
  later when a DB is available. Tests use mocked Prisma and never need the
  column to physically exist.
- Web verification is `pnpm --filter @whatapp/web typecheck` + `build` + the
  component tests in this plan. Live click-through in a browser is **"live
  verification deferred"** (no live API).
- The Phase 1 unified send is `MessageService.send(msg, options)` in
  `apps/api/src/messages/message.service.ts`. It enforces window + consent,
  logs the `messages` row, links `campaignId`/`templateId` via `options`, and
  returns a structured `MetaSendResult` (`{ ok, wamid?, error? }`). **REUSE it
  — never reimplement sending.**

---

## Schema check

The existing `Campaign` and `CampaignRecipient` models
(`packages/db/prisma/schema.prisma`) already carry every field this phase
writes:

| Spec need | Model field |
|---|---|
| FR-05.1 name / template / segment / mapping | `Campaign.name`, `templateId`, `segmentId`, `variableMapping` (`Json`, default `{}`) |
| FR-05.6 status / schedule / lifecycle times | `Campaign.status` (`CampaignStatus`, default `draft`), `scheduledAt`, `startedAt`, `completedAt` |
| FR-05.15 stats | `Campaign.stats` (`Json`, default `{}`) |
| FR-05.1 author | `Campaign.createdBy` |
| FR-05.6 recipient row | `CampaignRecipient.campaignId`, `contactId`, `waId`, `status` (`RecipientStatus`, default `pending`) |
| FR-05.8 send outcome | `CampaignRecipient.messageId`, `error` (`Json`), `sentAt` |
| FR-05.2 skip reason | `CampaignRecipient.skipReason` |
| FR-05.9 idempotency guard | `CampaignRecipient.@@unique([campaignId, contactId])` |
| FR-05.14 recipient query by status | `CampaignRecipient.@@index([campaignId, status])` |
| FR-05.10 message↔campaign link | `Message.campaignId` + `Message.@@index([campaignId])` |

`CampaignStatus` = `draft | scheduled | sending | paused | completed |
cancelled | failed`.
`RecipientStatus` = `pending | sent | delivered | read | failed | skipped`.
**No migration is required this phase.** Task 1 verifies this against the live
schema file before any code is written; if (and only if) a gap is found it adds
an additive nullable column + deferred migration per the Convention above.

---

## Decisions locked in this plan

These resolve the ambiguous points the spec leaves open. They are
**authoritative for the build** — do not improvise different behaviour.

1. **Stats shape.** `Campaign.stats` JSON =
   `{ total, pending, sent, delivered, read, failed, skipped }` — one integer
   per `RecipientStatus` plus `pending` and `total`. `aggregateCampaignStats`
   (Task 3) is the single function that produces this object from a list of
   recipient statuses; everywhere stats are written, they are recomputed by it,
   never incremented ad-hoc. (Spec FR-05.15 names
   `total/sent/delivered/read/failed/skipped`; `pending` is added so the
   progress bar can show outstanding work — a superset, spec-compatible.)
2. **Variable mapping shape.** `Campaign.variableMapping` JSON is an object
   keyed by template variable position (`"1"`, `"2"`, …). Each value is one of:
   `{ "kind": "fixed", "value": "<string>" }` or
   `{ "kind": "field", "path": "<contact field or attributes path>" }`. Allowed
   `field` paths: top-level `Contact` columns `displayName`, `profileName`,
   `phoneE164`, `country`, `waId`; or `attributes.<key>` for any key in the
   `Contact.attributes` JSON. The `CampaignVariableMapping` Zod schema (Task 2)
   is the single source of truth for this shape, imported by the API DTO and
   the resolver.
3. **Audience resolution timing.** The audience is resolved **at send time**
   (FR-05.5) inside `CampaignSendService.start()`, by re-evaluating the
   segment. `GET /api/campaigns/:id/audience` (FR-05.4) re-evaluates it live for
   the builder preview. Both go through `buildAudienceWhere` (Task 3) so the
   counts the builder shows match exactly what `start()` will enqueue.
4. **Scheduling mechanism.** `POST /:id/send` with no `scheduledAt` (or a
   past one) starts immediately. With a future `scheduledAt` the campaign moves
   to `scheduled` and a **BullMQ delayed job** on a new `campaign-start` queue
   fires at `scheduledAt`; its processor calls the same `start()` path. This
   keeps scheduling in Redis (durable, survives restarts) with no extra cron.
   `campaign-start` is added to `apps/worker/src/queues.ts`.
5. **Rate limiter.** The `campaign-send` BullMQ `Queue`/`Worker` is configured
   with `limiter: { max: 20, duration: 1000 }` — 20 jobs/second, well under
   Meta's ~80 msg/s per-number throughput (`integrations.md` §1.5) and far
   under even the lowest 250/24h messaging tier in burst terms. The constant
   lives in `apps/worker/src/queues.ts` as `CAMPAIGN_SEND_LIMITER` so it is
   declared once. Per-recipient jobs get `attempts: 5` with
   `backoff: { type: "exponential", delay: 5000 }`; the processor only signals
   a retry (re-throws) for a Meta `130429` rate error — every other failure is
   recorded on the recipient row and the job succeeds (FR-05.8).
6. **Webhook → recipient reflection (FR-05.10).** Phase 1's shared
   `handleStatus` in `packages/shared/src/meta/envelope-processor.ts` already
   updates the `messages` row on a delivery/read/failed status. This phase
   extends `handleStatus`: after it updates a message, **if that message has a
   `campaignId`**, it (a) updates the matching `campaign_recipients` row
   (matched by `messageId`) to the same terminal status and (b) recomputes and
   writes the campaign's `stats`. This is done through two new optional deps on
   `EnvelopeProcessorDeps` so the shared package stays framework-agnostic and
   the logic is unit-tested with mocked Prisma. No separate hook, no polling.
7. **Completion (FR-05.13).** After any recipient row reaches a terminal state
   — in the worker processor, in the webhook reflection, and in retry-failed —
   a `CampaignsService.checkCompletion(campaignId)` is called. It loads all
   recipient statuses, and if every row is terminal
   (`sent|delivered|read|failed|skipped`) and the campaign is not already
   `completed`/`cancelled`, sets `status=completed` + `completedAt`.
   `isCampaignComplete` (Task 3) is the pure predicate it uses.

---

## File structure

New files this phase creates:

**Shared (`packages/shared/src/campaigns`)**
- `types.ts` — `CampaignVariableMapping` / `VariableSource` Zod schemas +
  inferred types; the canonical shape of `Campaign.variableMapping`.
- `resolve-mapping.ts` — `resolveVariableMapping(mapping, contact, requiredVars)`
  pure function + `ResolvedMapping` type (FR-05.2).
- `resolve-mapping.test.ts` — exhaustive resolver tests.
- `audience.ts` — `buildAudienceWhere(segmentWhere, templateCategory)` pure
  function (FR-05.4) + `terminalRecipientStatuses` / `isCampaignComplete`
  (FR-05.13) + `aggregateCampaignStats` (FR-05.15) and the `CampaignStats` type.
- `audience.test.ts` — tests for the four pure helpers above.
- modify `packages/shared/src/index.ts` — barrel exports for the new files.

**Shared — webhook reflection (`packages/shared/src/meta`)**
- modify `envelope-processor.ts` — extend `EnvelopeProcessorDeps` and
  `handleStatus` for campaign-recipient reflection (FR-05.10).
- modify `envelope-processor.test.ts` — tests for the campaign-reflection path.

**API — Campaigns (`apps/api/src/campaigns`)**
- `dto.ts` — Zod schemas for every campaigns body/query.
- `campaigns.service.ts` + `.test.ts` — CRUD, edit-guard, pause/resume/cancel,
  retry-failed, list/detail/recipients, `checkCompletion`.
- `campaign-send.service.ts` + `.test.ts` — audience resolution, recipient-row
  creation, enqueue, schedule.
- `campaigns.controller.ts` + `.test.ts` — `/api/campaigns` routes.
- `campaign-queue.provider.ts` — NestJS providers for the `campaign-send` and
  `campaign-start` BullMQ `Queue`s.
- `campaigns.module.ts`.

**API — wiring**
- modify `apps/api/src/app.module.ts` — register `CampaignsModule`.
- modify `apps/api/src/webhooks/webhook-processor.service.ts` — pass the two new
  campaign-reflection deps into `processEnvelope`.

**Worker (`apps/worker/src`)**
- modify `queues.ts` — add `QUEUE_CAMPAIGN_START`, `CAMPAIGN_SEND_LIMITER`,
  `CAMPAIGN_SEND_JOB_OPTS`.
- `processors/campaign-send.processor.ts` + `.test.ts` — the per-recipient
  send job, a pure function with injected deps.
- `processors/campaign-start.processor.ts` + `.test.ts` — the scheduled-start
  job, a pure function with injected deps.
- modify `main.ts` — register `Worker`s for `campaign-send` (with the limiter)
  and `campaign-start`.
- `lib/internal-api.ts` + `.test.ts` — a thin authenticated client the worker
  uses to call the API's internal campaign endpoints (start / send-one).

**API — internal endpoints (`apps/api/src/campaigns`)**
- `internal-campaigns.controller.ts` + `.test.ts` — `POST
  /internal/campaigns/:id/start` and `POST
  /internal/campaigns/recipients/:recipientId/send`, guarded by the existing
  `CallbackAuthGuard` (`apps/api/src/messages/callback-auth.guard.ts`).

> **Worker↔API boundary.** The worker has no Nest DI and no `MessageService`.
> The send path and Prisma writes live in the API. So the worker processors are
> thin: they call back into the API's **internal** endpoints (Bearer
> `CALLBACK_SECRET`, same pattern as Phase 5's n8n callbacks) which do the real
> work via `CampaignSendService`. This keeps `MessageService` as the one send
> path and keeps all business logic unit-testable in the API package. The
> worker processors are tested as pure functions with the internal-API client
> mocked.

**Web (`apps/web/src`)**
- `lib/campaigns-api.ts` — typed fetch functions for the campaigns API.
- `components/campaigns/CampaignStatusBadge.tsx` + `.test.tsx` — status badge.
- `components/campaigns/CampaignProgress.tsx` + `.test.tsx` — progress bar +
  stat tiles from `CampaignStats`.
- `components/campaigns/VariableMappingEditor.tsx` + `.test.tsx` — per-variable
  source picker with a live sample preview.
- `components/campaigns/RecipientTable.tsx` + `.test.tsx` — recipient list with
  status filter.
- `routes/Campaigns.tsx` + `.test.tsx` — campaigns list screen.
- `routes/CampaignBuilder.tsx` + `.test.tsx` — create/edit screen with live
  sendable count.
- `routes/CampaignDetail.tsx` + `.test.tsx` — detail screen with progress,
  stats, recipient table, and controls.
- modify `apps/web/src/App.tsx` — replace the `/campaigns` placeholder with real
  routes (`/campaigns`, `/campaigns/new`, `/campaigns/:id`,
  `/campaigns/:id/edit`).

---

## Task 1: Verify the Campaign schema covers the spec

**Files:**
- Read only: `packages/db/prisma/schema.prisma`

No code is written in this task — it locks in the "no migration needed"
assumption before any dependent task runs.

- [ ] **Step 1: Read `model Campaign` and `model CampaignRecipient`**

Confirm every field in the "Schema check" table above is present with the
stated name and type, that `CampaignRecipient` has `@@unique([campaignId,
contactId])` and `@@index([campaignId, status])`, and that `Message` has
`campaignId` + `@@index([campaignId])`.

- [ ] **Step 2: Decide**

- If every field is present (expected) — write nothing. Record in the commit
  message that the schema already covers Phase 4.
- If a field is genuinely missing — add it as an **additive, nullable** column,
  run `pnpm --filter @whatapp/db db:generate`, generate the SQL with
  `pnpm exec prisma migrate diff --from-migrations ./prisma/migrations
  --to-schema-datamodel ./prisma/schema.prisma --script` (from `packages/db`),
  save it under
  `packages/db/prisma/migrations/<YYYYMMDDHHMMSS>_campaign_phase4/migration.sql`,
  and run `pnpm exec prisma validate` (expect "valid"). Mark applying it **"live
  verification deferred"**.

- [ ] **Step 3: Commit**

```bash
git add -A
git commit -m "chore: verify Campaign schema covers Phase 4 (no migration needed)"
```

(If Step 2 produced a migration, the message is instead
`feat: add additive Campaign column for Phase 4 (migration deferred)`.)

---

## Task 2: Shared variable-mapping schema & types

**Files:**
- Create: `packages/shared/src/campaigns/types.ts`
- Modify: `packages/shared/src/index.ts`

The canonical Zod schema for `Campaign.variableMapping`. Both the API DTO
(Task 9) and the resolver (Task 3) import from here so the shapes never drift.
No test of its own — exercised by Task 3 and Task 9.

- [ ] **Step 1: Implement `packages/shared/src/campaigns/types.ts`:**

```typescript
import { z } from "zod";

/**
 * The canonical shape of a campaign's `variableMapping` JSON column (FR-05.2).
 * The key is the template variable position as a string ("1", "2", ...). Each
 * value is a source: a fixed string, or a contact field/attribute path.
 */

/** Top-level Contact columns a mapping may read directly. */
export const CONTACT_FIELD_PATHS = [
  "displayName",
  "profileName",
  "phoneE164",
  "country",
  "waId",
] as const;
export type ContactFieldPath = (typeof CONTACT_FIELD_PATHS)[number];

/** A fixed literal string used verbatim for every recipient. */
export const fixedSourceSchema = z.object({
  kind: z.literal("fixed"),
  value: z.string(),
});

/**
 * A per-recipient value pulled from the contact: either a top-level column
 * (one of CONTACT_FIELD_PATHS) or an `attributes.<key>` path into the
 * Contact.attributes JSON.
 */
export const fieldSourceSchema = z.object({
  kind: z.literal("field"),
  path: z
    .string()
    .min(1)
    .refine(
      (p) =>
        (CONTACT_FIELD_PATHS as readonly string[]).includes(p) ||
        /^attributes\.[A-Za-z0-9_]+$/.test(p),
      { message: "path must be a Contact column or an attributes.<key> path" },
    ),
});

/** One template variable's source. */
export const variableSourceSchema = z.discriminatedUnion("kind", [
  fixedSourceSchema,
  fieldSourceSchema,
]);
export type VariableSource = z.infer<typeof variableSourceSchema>;

/** position ("1", "2", ...) -> source. */
export const campaignVariableMappingSchema = z.record(
  z.string().regex(/^[1-9][0-9]*$/, "key must be a 1-based variable index"),
  variableSourceSchema,
);
export type CampaignVariableMapping = z.infer<
  typeof campaignVariableMappingSchema
>;
```

- [ ] **Step 2: Add a barrel export to `packages/shared/src/index.ts`:**

```typescript
export * from "./campaigns/types";
```

- [ ] **Step 3: Verify it builds**

Run: `pnpm --filter @whatapp/shared typecheck`
Expected: PASS (no test yet — the schema is exercised by Task 3).

- [ ] **Step 4: Commit**

```bash
git add packages/shared/src/campaigns/types.ts packages/shared/src/index.ts
git commit -m "feat: add campaign variable-mapping schema to shared"
```

---

## Task 3: Shared pure helpers — resolver, audience, stats, completion

**Files:**
- Create: `packages/shared/src/campaigns/resolve-mapping.ts`
- Create: `packages/shared/src/campaigns/resolve-mapping.test.ts`
- Create: `packages/shared/src/campaigns/audience.ts`
- Create: `packages/shared/src/campaigns/audience.test.ts`
- Modify: `packages/shared/src/index.ts`

The four pure, exhaustively-tested units the rest of the phase builds on. No
Prisma, no NestJS, no I/O.

- [ ] **Step 1: Write the failing test `resolve-mapping.test.ts`:**

```typescript
import { describe, it, expect } from "vitest";
import { resolveVariableMapping } from "./resolve-mapping";
import type { CampaignVariableMapping } from "./types";

const contact = {
  waId: "9715000",
  displayName: "Ahmed",
  profileName: null,
  phoneE164: "+9715000",
  country: "AE",
  attributes: { agentName: "Sara", city: "" },
};

describe("resolveVariableMapping", () => {
  it("resolves a fixed source verbatim", () => {
    const m: CampaignVariableMapping = { "1": { kind: "fixed", value: "Hi" } };
    expect(resolveVariableMapping(m, contact, ["1"])).toEqual({
      ok: true,
      values: { "1": "Hi" },
    });
  });

  it("resolves a top-level contact field", () => {
    const m: CampaignVariableMapping = {
      "1": { kind: "field", path: "displayName" },
    };
    expect(resolveVariableMapping(m, contact, ["1"])).toEqual({
      ok: true,
      values: { "1": "Ahmed" },
    });
  });

  it("resolves an attributes.<key> path", () => {
    const m: CampaignVariableMapping = {
      "1": { kind: "field", path: "attributes.agentName" },
    };
    expect(resolveVariableMapping(m, contact, ["1"])).toEqual({
      ok: true,
      values: { "1": "Sara" },
    });
  });

  it("skips when a required variable's field value is null", () => {
    const m: CampaignVariableMapping = {
      "1": { kind: "field", path: "profileName" },
    };
    expect(resolveVariableMapping(m, contact, ["1"])).toEqual({
      ok: false,
      skipReason: "missing_variable",
    });
  });

  it("skips when a required variable's field value is an empty string", () => {
    const m: CampaignVariableMapping = {
      "1": { kind: "field", path: "attributes.city" },
    };
    expect(resolveVariableMapping(m, contact, ["1"])).toEqual({
      ok: false,
      skipReason: "missing_variable",
    });
  });

  it("skips when a required variable has no mapping entry at all", () => {
    expect(resolveVariableMapping({}, contact, ["1"])).toEqual({
      ok: false,
      skipReason: "missing_variable",
    });
  });

  it("skips when an attributes key is absent from the contact", () => {
    const m: CampaignVariableMapping = {
      "1": { kind: "field", path: "attributes.missing" },
    };
    expect(resolveVariableMapping(m, contact, ["1"])).toEqual({
      ok: false,
      skipReason: "missing_variable",
    });
  });

  it("resolves every required variable and ignores extra mapping keys", () => {
    const m: CampaignVariableMapping = {
      "1": { kind: "fixed", value: "A" },
      "2": { kind: "field", path: "displayName" },
      "3": { kind: "fixed", value: "unused" },
    };
    expect(resolveVariableMapping(m, contact, ["1", "2"])).toEqual({
      ok: true,
      values: { "1": "A", "2": "Ahmed" },
    });
  });

  it("treats a fixed empty string as a valid (present) value", () => {
    const m: CampaignVariableMapping = { "1": { kind: "fixed", value: "" } };
    expect(resolveVariableMapping(m, contact, ["1"])).toEqual({
      ok: true,
      values: { "1": "" },
    });
  });
});
```

- [ ] **Step 2: Run it — expect FAIL**

Run: `pnpm --filter @whatapp/shared test resolve-mapping`
Expected: FAIL — `resolveVariableMapping` is not defined.

- [ ] **Step 3: Implement `packages/shared/src/campaigns/resolve-mapping.ts`:**

```typescript
import {
  CONTACT_FIELD_PATHS,
  type CampaignVariableMapping,
  type ContactFieldPath,
} from "./types";

/** The minimal contact shape the resolver reads. */
export interface ResolverContact {
  waId: string;
  displayName: string | null;
  profileName: string | null;
  phoneE164: string | null;
  country: string | null;
  attributes: Record<string, unknown>;
}

/** Outcome of resolving a campaign's mapping for one recipient (FR-05.2). */
export type ResolvedMapping =
  | { ok: true; values: Record<string, string> }
  | { ok: false; skipReason: "missing_variable" };

/** Reads one field/attribute path from a contact; returns null if absent. */
function readPath(contact: ResolverContact, path: string): string | null {
  if ((CONTACT_FIELD_PATHS as readonly string[]).includes(path)) {
    const v = contact[path as ContactFieldPath];
    return typeof v === "string" ? v : null;
  }
  const key = path.slice("attributes.".length);
  const v = contact.attributes[key];
  if (v == null) return null;
  return typeof v === "string" ? v : String(v);
}

/**
 * Resolves a campaign's `variableMapping` for one recipient (FR-05.2). Every
 * variable in `requiredVars` must resolve to a non-empty value. A `fixed`
 * source is always present (even an empty string). A `field` source missing a
 * value, empty, or absent from the mapping -> the recipient is skipped with
 * `skipReason="missing_variable"` rather than sent a broken message.
 */
export function resolveVariableMapping(
  mapping: CampaignVariableMapping,
  contact: ResolverContact,
  requiredVars: string[],
): ResolvedMapping {
  const values: Record<string, string> = {};
  for (const position of requiredVars) {
    const source = mapping[position];
    if (!source) return { ok: false, skipReason: "missing_variable" };
    if (source.kind === "fixed") {
      values[position] = source.value;
      continue;
    }
    const raw = readPath(contact, source.path);
    if (raw === null || raw === "") {
      return { ok: false, skipReason: "missing_variable" };
    }
    values[position] = raw;
  }
  return { ok: true, values };
}
```

- [ ] **Step 4: Run it — expect PASS**

Run: `pnpm --filter @whatapp/shared test resolve-mapping`
Expected: PASS — all 9 tests green.

- [ ] **Step 5: Write the failing test `audience.test.ts`:**

```typescript
import { describe, it, expect } from "vitest";
import {
  buildAudienceWhere,
  aggregateCampaignStats,
  isCampaignComplete,
  TERMINAL_RECIPIENT_STATUSES,
} from "./audience";

describe("buildAudienceWhere", () => {
  const seg = { tags: { has: "vip" } };

  it("always excludes blocked contacts and keeps the segment clause", () => {
    const w = buildAudienceWhere(seg, "UTILITY");
    expect(w.isBlocked).toBe(false);
    expect(w.AND).toEqual([seg]);
  });

  it("excludes opted_out for a MARKETING template", () => {
    expect(buildAudienceWhere(seg, "MARKETING").consentState).toEqual({
      not: "opted_out",
    });
  });

  it("excludes only opted_out (unknown allowed) for UTILITY", () => {
    expect(buildAudienceWhere(seg, "UTILITY").consentState).toEqual({
      not: "opted_out",
    });
  });

  it("excludes only opted_out for AUTHENTICATION", () => {
    expect(buildAudienceWhere(seg, "AUTHENTICATION").consentState).toEqual({
      not: "opted_out",
    });
  });
});

describe("aggregateCampaignStats", () => {
  it("counts each status and totals them", () => {
    expect(
      aggregateCampaignStats([
        "pending",
        "sent",
        "sent",
        "delivered",
        "read",
        "failed",
        "skipped",
      ]),
    ).toEqual({
      total: 7,
      pending: 1,
      sent: 2,
      delivered: 1,
      read: 1,
      failed: 1,
      skipped: 1,
    });
  });

  it("returns an all-zero object for no recipients", () => {
    expect(aggregateCampaignStats([])).toEqual({
      total: 0,
      pending: 0,
      sent: 0,
      delivered: 0,
      read: 0,
      failed: 0,
      skipped: 0,
    });
  });
});

describe("isCampaignComplete", () => {
  it("is true when every recipient is terminal", () => {
    expect(
      isCampaignComplete(["sent", "delivered", "read", "failed", "skipped"]),
    ).toBe(true);
  });

  it("is false when any recipient is still pending", () => {
    expect(isCampaignComplete(["sent", "pending"])).toBe(false);
  });

  it("is false for an empty recipient list", () => {
    expect(isCampaignComplete([])).toBe(false);
  });

  it("TERMINAL_RECIPIENT_STATUSES omits pending and includes sent", () => {
    expect(TERMINAL_RECIPIENT_STATUSES).not.toContain("pending");
    expect(TERMINAL_RECIPIENT_STATUSES).toContain("sent");
  });
});
```

- [ ] **Step 6: Run it — expect FAIL**

Run: `pnpm --filter @whatapp/shared test audience`
Expected: FAIL — `buildAudienceWhere` is not defined.

- [ ] **Step 7: Implement `packages/shared/src/campaigns/audience.ts`:**

```typescript
/** Recipient statuses that count as terminal — no further work (FR-05.13). */
export const TERMINAL_RECIPIENT_STATUSES = [
  "sent",
  "delivered",
  "read",
  "failed",
  "skipped",
] as const;

/** Template category as used for the consent rule (FR-05.4). */
export type AudienceTemplateCategory =
  | "MARKETING"
  | "UTILITY"
  | "AUTHENTICATION";

/** Keyed integer counts produced by aggregateCampaignStats (FR-05.15). */
export interface CampaignStats {
  total: number;
  pending: number;
  sent: number;
  delivered: number;
  read: number;
  failed: number;
  skipped: number;
}

/**
 * Builds the Prisma `Contact` where-clause for a campaign's audience (FR-05.4).
 * `segmentWhere` is the compiled segment definition. Blocked contacts are
 * always excluded; `opted_out` is always excluded. `unknown` consent is
 * allowed for every category — the spec excludes only `opted_out` for
 * marketing, never `unknown`. The category is accepted for future policy
 * tightening but does not change the clause today.
 */
export function buildAudienceWhere(
  segmentWhere: Record<string, unknown>,
  _category: AudienceTemplateCategory,
): Record<string, unknown> {
  return {
    AND: [segmentWhere],
    isBlocked: false,
    consentState: { not: "opted_out" },
  };
}

/** Tallies a list of recipient statuses into a CampaignStats object. */
export function aggregateCampaignStats(statuses: string[]): CampaignStats {
  const s: CampaignStats = {
    total: statuses.length,
    pending: 0,
    sent: 0,
    delivered: 0,
    read: 0,
    failed: 0,
    skipped: 0,
  };
  const counts = s as unknown as Record<string, number>;
  for (const status of statuses) {
    if (status !== "total" && status in s) counts[status] += 1;
  }
  return s;
}

/**
 * True when a campaign is complete (FR-05.13): there is at least one recipient
 * and every recipient row is in a terminal state.
 */
export function isCampaignComplete(statuses: string[]): boolean {
  if (statuses.length === 0) return false;
  const terminal = new Set<string>(TERMINAL_RECIPIENT_STATUSES);
  return statuses.every((s) => terminal.has(s));
}
```

- [ ] **Step 8: Run it — expect PASS**

Run: `pnpm --filter @whatapp/shared test audience`
Expected: PASS — all tests green.

- [ ] **Step 9: Add barrel exports to `packages/shared/src/index.ts`:**

```typescript
export * from "./campaigns/resolve-mapping";
export * from "./campaigns/audience";
```

- [ ] **Step 10: Verify the package builds**

Run: `pnpm --filter @whatapp/shared typecheck && pnpm --filter @whatapp/shared build`
Expected: PASS.

- [ ] **Step 11: Commit**

```bash
git add packages/shared/src/campaigns packages/shared/src/index.ts
git commit -m "feat: add campaign resolver/audience/stats/completion helpers"
```

---

## Task 4: Reflect delivery webhooks onto campaign recipients (FR-05.10)

**Files:**
- Modify: `packages/shared/src/meta/envelope-processor.ts`
- Modify: `packages/shared/src/meta/envelope-processor.test.ts`

Phase 1's `handleStatus` updates the `messages` row on every delivery/read/
failed status. This task extends it: when the updated message carries a
`campaignId`, also update the matching `campaign_recipients` row and recompute
the campaign `stats`. The new behaviour is exposed as one optional callback dep
so the shared package stays framework-agnostic; the API wires it in Task 12.

> **Read `handleStatus` first.** In `envelope-processor.ts`, `handleStatus`
> currently: `findFirst({ where: { wamid } })` on `message`, appends a
> `messageStatusEvent`, checks `isForwardStatus`, then `prisma.message.update`.
> The `message.findFirst`/`findUnique` select returns `{ id, status }` today.
> This task widens that select to also return `campaignId` and adds the
> reflection step after the successful `message.update`.

- [ ] **Step 1: Write the failing test — add to `envelope-processor.test.ts`:**

```typescript
import { describe, it, expect, vi } from "vitest";
import { processEnvelope } from "./envelope-processor";

// A minimal delivered-status envelope helper. Reuse the existing test's
// envelope builder if one is present; otherwise inline this shape.
function deliveredEnvelope(wamid: string) {
  return {
    object: "whatsapp_business_account",
    entry: [
      {
        id: "waba1",
        changes: [
          {
            field: "messages",
            value: {
              messaging_product: "whatsapp",
              metadata: { phone_number_id: "p1" },
              statuses: [
                {
                  id: wamid,
                  status: "delivered",
                  timestamp: "1700000000",
                  recipient_id: "9715000",
                },
              ],
            },
          },
        ],
      },
    ],
  };
}

describe("processEnvelope — campaign reflection (FR-05.10)", () => {
  it("reflects a delivered status onto the campaign recipient", async () => {
    const reflectCampaignStatus = vi.fn().mockResolvedValue(undefined);
    const prisma = {
      message: {
        findFirst: vi.fn().mockResolvedValue({
          id: "m1",
          status: "sent",
          campaignId: "camp1",
        }),
        update: vi.fn().mockResolvedValue({}),
      },
      messageStatusEvent: { create: vi.fn().mockResolvedValue({}) },
    };

    await processEnvelope(deliveredEnvelope("wamid-1"), {
      prisma: prisma as never,
      mediaQueue: { add: vi.fn() } as never,
      flowQueue: { add: vi.fn() } as never,
      reflectCampaignStatus,
    });

    expect(reflectCampaignStatus).toHaveBeenCalledWith({
      campaignId: "camp1",
      messageId: "m1",
      status: "delivered",
    });
  });

  it("does not call reflection for a non-campaign message", async () => {
    const reflectCampaignStatus = vi.fn();
    const prisma = {
      message: {
        findFirst: vi.fn().mockResolvedValue({
          id: "m2",
          status: "sent",
          campaignId: null,
        }),
        update: vi.fn().mockResolvedValue({}),
      },
      messageStatusEvent: { create: vi.fn().mockResolvedValue({}) },
    };

    await processEnvelope(deliveredEnvelope("wamid-2"), {
      prisma: prisma as never,
      mediaQueue: { add: vi.fn() } as never,
      flowQueue: { add: vi.fn() } as never,
      reflectCampaignStatus,
    });

    expect(reflectCampaignStatus).not.toHaveBeenCalled();
  });

  it("is a safe no-op when reflectCampaignStatus dep is absent", async () => {
    const prisma = {
      message: {
        findFirst: vi.fn().mockResolvedValue({
          id: "m3",
          status: "sent",
          campaignId: "camp1",
        }),
        update: vi.fn().mockResolvedValue({}),
      },
      messageStatusEvent: { create: vi.fn().mockResolvedValue({}) },
    };

    await expect(
      processEnvelope(deliveredEnvelope("wamid-3"), {
        prisma: prisma as never,
        mediaQueue: { add: vi.fn() } as never,
        flowQueue: { add: vi.fn() } as never,
      }),
    ).resolves.not.toThrow();
  });
});
```

> If the existing test file already has a private envelope builder, delete the
> local `deliveredEnvelope` here and reuse it. Keep the three new `it` cases.

- [ ] **Step 2: Run it — expect FAIL**

Run: `pnpm --filter @whatapp/shared test envelope-processor`
Expected: FAIL — `reflectCampaignStatus` is not a known dep / not called.

- [ ] **Step 3: Extend `EnvelopeProcessorDeps` in `envelope-processor.ts`:**

Add this optional dep to the `EnvelopeProcessorDeps` interface (alongside the
existing `recordQualityUpdate` / `onOptOut` optional callbacks):

```typescript
  /**
   * Reflects a campaign message's delivery status onto its
   * `campaign_recipients` row and the campaign `stats` (FR-05.10). Optional —
   * only the API wires it; absent in unit contexts that do not exercise
   * campaigns.
   */
  reflectCampaignStatus?: (args: {
    campaignId: string;
    messageId: string;
    status: "sent" | "delivered" | "read" | "failed";
  }) => Promise<void>;
```

- [ ] **Step 4: Widen the message lookup select in `handleStatus`:**

Change the `message` delegate type and the `findFirst` call so the returned row
also carries `campaignId`. Update the `prisma.message.findFirst` /
`findUnique` type annotations near the top of the file to return
`{ id: string; status: string; campaignId: string | null }`. The call site
needs no `select` change if the existing query returns the whole row; if it has
an explicit `select`, add `campaignId: true`.

- [ ] **Step 5: Add the reflection call after the successful `message.update`:**

At the end of `handleStatus`, immediately after
`await deps.prisma.message.update(...)`:

```typescript
  // FR-05.10 — reflect a campaign message's status onto its recipient row.
  if (message.campaignId && deps.reflectCampaignStatus) {
    await deps.reflectCampaignStatus({
      campaignId: message.campaignId,
      messageId: message.id,
      status: status.status as "sent" | "delivered" | "read" | "failed",
    });
  }
```

- [ ] **Step 6: Run it — expect PASS**

Run: `pnpm --filter @whatapp/shared test envelope-processor`
Expected: PASS — the three new cases green and every pre-existing
envelope-processor test still green.

- [ ] **Step 7: Build the shared package**

Run: `pnpm --filter @whatapp/shared typecheck && pnpm --filter @whatapp/shared build`
Expected: PASS.

- [ ] **Step 8: Commit**

```bash
git add packages/shared/src/meta/envelope-processor.ts \
  packages/shared/src/meta/envelope-processor.test.ts
git commit -m "feat: reflect campaign message delivery status via webhook"
```

---

## Task 5: Worker queue constants — limiter and start queue

**Files:**
- Modify: `apps/worker/src/queues.ts`

Declare the rate limiter, the per-recipient job options, and the new
`campaign-start` queue name once, so the API (which enqueues) and the worker
(which consumes) share the exact same config.

- [ ] **Step 1: Edit `apps/worker/src/queues.ts`** — add the start-queue name
  to the constants and the `ALL_QUEUE_NAMES` array:

```typescript
export const QUEUE_CAMPAIGN_START = "campaign-start";
```

Add `QUEUE_CAMPAIGN_START` to `ALL_QUEUE_NAMES` (so worker bootstrapping and
health checks see it).

- [ ] **Step 2: Add the limiter and job-option constants** to the same file:

```typescript
import type { JobsOptions, RateLimiterOptions } from "bullmq";

/**
 * Rate limiter for the `campaign-send` queue (FR-05.7). 20 jobs/second is well
 * under Meta's ~80 msg/s per-number throughput (integrations.md §1.5) and
 * comfortably paced for every messaging tier. Used by both the BullMQ Worker
 * (apps/worker) and is the documented ceiling for enqueue rate.
 */
export const CAMPAIGN_SEND_LIMITER: RateLimiterOptions = {
  max: 20,
  duration: 1000,
};

/**
 * Per-recipient job options (FR-05.8): up to 5 attempts with exponential
 * backoff. The processor only re-throws (triggering a retry) for a Meta
 * `130429` rate error; all other failures are recorded on the recipient row
 * and the job completes. `removeOnComplete` keeps Redis small.
 */
export const CAMPAIGN_SEND_JOB_OPTS: JobsOptions = {
  attempts: 5,
  backoff: { type: "exponential", delay: 5000 },
  removeOnComplete: 1000,
  removeOnFail: 5000,
};
```

- [ ] **Step 3: Typecheck**

Run: `pnpm --filter @whatapp/worker typecheck`
Expected: PASS.

- [ ] **Step 4: Commit**

```bash
git add apps/worker/src/queues.ts
git commit -m "feat: add campaign-send limiter and campaign-start queue"
```

---

## Task 6: Worker internal-API client

**Files:**
- Create: `apps/worker/src/lib/internal-api.ts`
- Create: `apps/worker/src/lib/internal-api.test.ts`

A thin authenticated client the worker processors use to call the API's
internal campaign endpoints. It POSTs with `Authorization: Bearer
{CALLBACK_SECRET}` (the same secret the API's `CallbackAuthGuard` checks). It is
the only thing the worker processors touch the API with — keeping the send path
and Prisma writes entirely in the API.

- [ ] **Step 1: Write the failing test `internal-api.test.ts`:**

```typescript
import { describe, it, expect, vi, beforeEach } from "vitest";
import { createInternalApiClient } from "./internal-api";

describe("createInternalApiClient", () => {
  beforeEach(() => {
    vi.restoreAllMocks();
  });

  it("POSTs to the start endpoint with the bearer secret", async () => {
    const fetchMock = vi
      .fn()
      .mockResolvedValue({ ok: true, status: 200, json: async () => ({}) });
    const client = createInternalApiClient({
      baseUrl: "http://api:3000",
      callbackSecret: "s3cr3t",
      fetchImpl: fetchMock as never,
    });

    await client.startCampaign("camp1");

    expect(fetchMock).toHaveBeenCalledWith(
      "http://api:3000/internal/campaigns/camp1/start",
      expect.objectContaining({
        method: "POST",
        headers: expect.objectContaining({
          Authorization: "Bearer s3cr3t",
        }),
      }),
    );
  });

  it("returns the parsed body from sendRecipient", async () => {
    const fetchMock = vi.fn().mockResolvedValue({
      ok: true,
      status: 200,
      json: async () => ({ result: "sent", retriable: false }),
    });
    const client = createInternalApiClient({
      baseUrl: "http://api:3000",
      callbackSecret: "s3cr3t",
      fetchImpl: fetchMock as never,
    });

    const res = await client.sendRecipient("rec1");

    expect(res).toEqual({ result: "sent", retriable: false });
    expect(fetchMock).toHaveBeenCalledWith(
      "http://api:3000/internal/campaigns/recipients/rec1/send",
      expect.objectContaining({ method: "POST" }),
    );
  });

  it("throws on a non-2xx response so BullMQ can retry", async () => {
    const fetchMock = vi.fn().mockResolvedValue({
      ok: false,
      status: 503,
      json: async () => ({}),
    });
    const client = createInternalApiClient({
      baseUrl: "http://api:3000",
      callbackSecret: "s3cr3t",
      fetchImpl: fetchMock as never,
    });

    await expect(client.sendRecipient("rec1")).rejects.toThrow(/503/);
  });
});
```

- [ ] **Step 2: Run it — expect FAIL**

Run: `pnpm --filter @whatapp/worker test internal-api`
Expected: FAIL — `createInternalApiClient` is not defined.

- [ ] **Step 3: Implement `apps/worker/src/lib/internal-api.ts`:**

```typescript
/** The result the API returns from the send-recipient internal endpoint. */
export interface SendRecipientResult {
  /** Outcome for this recipient. */
  result: "sent" | "failed" | "skipped" | "already_done";
  /** True only when the failure was a Meta 130429 rate limit (FR-05.8). */
  retriable: boolean;
}

/** Config for the internal-API client. */
export interface InternalApiConfig {
  baseUrl: string;
  callbackSecret: string;
  /** Injectable for tests; defaults to global fetch. */
  fetchImpl?: typeof fetch;
}

/** The internal-API client surface the worker processors use. */
export interface InternalApiClient {
  /** Triggers a campaign's audience resolution + recipient enqueue. */
  startCampaign: (campaignId: string) => Promise<void>;
  /** Sends one recipient and returns the structured outcome. */
  sendRecipient: (recipientId: string) => Promise<SendRecipientResult>;
}

/**
 * Builds the worker's authenticated client for the API's internal campaign
 * endpoints. Every call carries `Authorization: Bearer {callbackSecret}`,
 * matching the API's `CallbackAuthGuard`. A non-2xx response throws so the
 * BullMQ job retries.
 */
export function createInternalApiClient(
  config: InternalApiConfig,
): InternalApiClient {
  const doFetch = config.fetchImpl ?? fetch;
  const headers = {
    "Content-Type": "application/json",
    Authorization: `Bearer ${config.callbackSecret}`,
  };

  async function post(path: string): Promise<unknown> {
    const res = await doFetch(`${config.baseUrl}${path}`, {
      method: "POST",
      headers,
    });
    if (!res.ok) {
      throw new Error(`internal API ${path} returned ${res.status}`);
    }
    return res.json();
  }

  return {
    async startCampaign(campaignId) {
      await post(`/internal/campaigns/${campaignId}/start`);
    },
    async sendRecipient(recipientId) {
      const body = (await post(
        `/internal/campaigns/recipients/${recipientId}/send`,
      )) as SendRecipientResult;
      return body;
    },
  };
}
```

- [ ] **Step 4: Run it — expect PASS**

Run: `pnpm --filter @whatapp/worker test internal-api`
Expected: PASS — all 3 tests green.

- [ ] **Step 5: Commit**

```bash
git add apps/worker/src/lib/internal-api.ts apps/worker/src/lib/internal-api.test.ts
git commit -m "feat: add worker internal-API client for campaigns"
```

---

## Task 7: Worker `campaign-send` recipient processor

**Files:**
- Create: `apps/worker/src/processors/campaign-send.processor.ts`
- Create: `apps/worker/src/processors/campaign-send.processor.test.ts`

The per-recipient job (FR-05.8, FR-05.9). It is a **pure function with injected
deps** — exactly the `flow-dispatch.processor.ts` pattern. It calls the API's
`sendRecipient` internal endpoint (which does the real mapping resolution +
`MessageService.send` + recipient-row update) and decides whether the BullMQ
job should retry: it **re-throws only when the result is `retriable`** (a Meta
`130429` rate error) so BullMQ applies the exponential backoff from
`CAMPAIGN_SEND_JOB_OPTS`; every other outcome completes the job.

- [ ] **Step 1: Write the failing test `campaign-send.processor.test.ts`:**

```typescript
import { describe, it, expect, vi } from "vitest";
import { processCampaignSendJob } from "./campaign-send.processor";

function deps(over: Partial<Parameters<typeof processCampaignSendJob>[1]> = {}) {
  return {
    sendRecipient: vi
      .fn()
      .mockResolvedValue({ result: "sent", retriable: false }),
    log: vi.fn(),
    ...over,
  };
}

describe("processCampaignSendJob", () => {
  it("calls sendRecipient with the recipient id", async () => {
    const d = deps();
    await processCampaignSendJob({ recipientId: "rec1" }, d);
    expect(d.sendRecipient).toHaveBeenCalledWith("rec1");
  });

  it("completes the job when the recipient is sent", async () => {
    const d = deps();
    await expect(
      processCampaignSendJob({ recipientId: "rec1" }, d),
    ).resolves.not.toThrow();
  });

  it("completes the job when the recipient was skipped", async () => {
    const d = deps({
      sendRecipient: vi
        .fn()
        .mockResolvedValue({ result: "skipped", retriable: false }),
    });
    await expect(
      processCampaignSendJob({ recipientId: "rec1" }, d),
    ).resolves.not.toThrow();
  });

  it("completes the job (no retry) on a permanent failure", async () => {
    const d = deps({
      sendRecipient: vi
        .fn()
        .mockResolvedValue({ result: "failed", retriable: false }),
    });
    await expect(
      processCampaignSendJob({ recipientId: "rec1" }, d),
    ).resolves.not.toThrow();
  });

  it("re-throws on a retriable (130429) failure so BullMQ backs off", async () => {
    const d = deps({
      sendRecipient: vi
        .fn()
        .mockResolvedValue({ result: "failed", retriable: true }),
    });
    await expect(
      processCampaignSendJob({ recipientId: "rec1" }, d),
    ).rejects.toThrow(/rate limit/i);
  });

  it("does nothing harmful when the recipient is already done (idempotent)", async () => {
    const d = deps({
      sendRecipient: vi
        .fn()
        .mockResolvedValue({ result: "already_done", retriable: false }),
    });
    await expect(
      processCampaignSendJob({ recipientId: "rec1" }, d),
    ).resolves.not.toThrow();
  });
});
```

- [ ] **Step 2: Run it — expect FAIL**

Run: `pnpm --filter @whatapp/worker test campaign-send.processor`
Expected: FAIL — `processCampaignSendJob` is not defined.

- [ ] **Step 3: Implement `apps/worker/src/processors/campaign-send.processor.ts`:**

```typescript
import type { SendRecipientResult } from "../lib/internal-api";

/** The data carried by one `campaign-send` job. */
export interface CampaignSendJobData {
  recipientId: string;
}

/** Injected collaborators for the recipient processor. */
export interface CampaignSendDeps {
  /** Calls the API's internal send-recipient endpoint. */
  sendRecipient: (recipientId: string) => Promise<SendRecipientResult>;
  /** Logging sink. */
  log: (message: string) => void;
}

/**
 * Processes one `campaign-send` job (FR-05.8, FR-05.9). It delegates the real
 * work — variable-mapping resolution, the unified send, the recipient-row
 * update, idempotency guard — to the API's internal endpoint. Its only local
 * decision is retry control: a `retriable` outcome (Meta `130429` rate limit)
 * is re-thrown so BullMQ applies the configured exponential backoff; every
 * other outcome completes the job (a permanent failure is already recorded on
 * the recipient row by the API and must not retry).
 */
export async function processCampaignSendJob(
  data: CampaignSendJobData,
  deps: CampaignSendDeps,
): Promise<void> {
  const outcome = await deps.sendRecipient(data.recipientId);
  deps.log(
    `campaign-send: recipient ${data.recipientId} -> ${outcome.result}` +
      (outcome.retriable ? " (retriable)" : ""),
  );
  if (outcome.retriable) {
    // Re-throw so BullMQ retries with exponential backoff (FR-05.8).
    throw new Error(
      `campaign-send: recipient ${data.recipientId} hit Meta rate limit ` +
        `(130429) — retrying`,
    );
  }
}
```

- [ ] **Step 4: Run it — expect PASS**

Run: `pnpm --filter @whatapp/worker test campaign-send.processor`
Expected: PASS — all 6 tests green.

- [ ] **Step 5: Commit**

```bash
git add apps/worker/src/processors/campaign-send.processor.ts \
  apps/worker/src/processors/campaign-send.processor.test.ts
git commit -m "feat: add campaign-send recipient worker processor"
```

---

## Task 8: Worker `campaign-start` scheduled-start processor

**Files:**
- Create: `apps/worker/src/processors/campaign-start.processor.ts`
- Create: `apps/worker/src/processors/campaign-start.processor.test.ts`

The job that fires at a campaign's `scheduledAt` (FR-05.6). It is a pure
function with injected deps; it simply calls the API's `startCampaign` internal
endpoint, which performs audience resolution and recipient enqueue.

- [ ] **Step 1: Write the failing test `campaign-start.processor.test.ts`:**

```typescript
import { describe, it, expect, vi } from "vitest";
import { processCampaignStartJob } from "./campaign-start.processor";

describe("processCampaignStartJob", () => {
  it("calls startCampaign with the campaign id", async () => {
    const startCampaign = vi.fn().mockResolvedValue(undefined);
    await processCampaignStartJob(
      { campaignId: "camp1" },
      { startCampaign, log: vi.fn() },
    );
    expect(startCampaign).toHaveBeenCalledWith("camp1");
  });

  it("propagates an error so BullMQ retries the scheduled start", async () => {
    const startCampaign = vi.fn().mockRejectedValue(new Error("api down"));
    await expect(
      processCampaignStartJob(
        { campaignId: "camp1" },
        { startCampaign, log: vi.fn() },
      ),
    ).rejects.toThrow(/api down/);
  });
});
```

- [ ] **Step 2: Run it — expect FAIL**

Run: `pnpm --filter @whatapp/worker test campaign-start.processor`
Expected: FAIL — `processCampaignStartJob` is not defined.

- [ ] **Step 3: Implement `apps/worker/src/processors/campaign-start.processor.ts`:**

```typescript
/** The data carried by one `campaign-start` job. */
export interface CampaignStartJobData {
  campaignId: string;
}

/** Injected collaborators for the scheduled-start processor. */
export interface CampaignStartDeps {
  /** Calls the API's internal start-campaign endpoint. */
  startCampaign: (campaignId: string) => Promise<void>;
  /** Logging sink. */
  log: (message: string) => void;
}

/**
 * Processes one `campaign-start` job (FR-05.6) — fired by a BullMQ delayed job
 * at the campaign's `scheduledAt`. It delegates to the API, which resolves the
 * audience at this moment (FR-05.5) and enqueues one `campaign-send` job per
 * sendable recipient. An error propagates so BullMQ retries the start.
 */
export async function processCampaignStartJob(
  data: CampaignStartJobData,
  deps: CampaignStartDeps,
): Promise<void> {
  deps.log(`campaign-start: starting campaign ${data.campaignId}`);
  await deps.startCampaign(data.campaignId);
}
```

- [ ] **Step 4: Run it — expect PASS**

Run: `pnpm --filter @whatapp/worker test campaign-start.processor`
Expected: PASS — both tests green.

- [ ] **Step 5: Commit**

```bash
git add apps/worker/src/processors/campaign-start.processor.ts \
  apps/worker/src/processors/campaign-start.processor.test.ts
git commit -m "feat: add campaign-start scheduled-start worker processor"
```

---

## Task 9: Campaigns DTOs

**Files:**
- Create: `apps/api/src/campaigns/dto.ts`

Zod schemas for every campaigns request body and query. The variable-mapping
shape is **imported from `@whatapp/shared`** (`campaignVariableMappingSchema`)
so the API and the resolver can never drift. Follow the DTO pattern in
`apps/api/src/templates/dto.ts`.

- [ ] **Step 1: Implement `apps/api/src/campaigns/dto.ts`:**

```typescript
import { z } from "zod";
import { campaignVariableMappingSchema } from "@whatapp/shared";

/** `POST /api/campaigns` body (FR-05.1). */
export const createCampaignSchema = z.object({
  name: z.string().min(1).max(120),
  templateId: z.string().uuid(),
  segmentId: z.string().uuid(),
  variableMapping: campaignVariableMappingSchema.default({}),
  /** Optional ISO datetime — when set at create time, schedules the campaign. */
  scheduledAt: z.string().datetime().optional(),
});
export type CreateCampaignDto = z.infer<typeof createCampaignSchema>;

/** `PATCH /api/campaigns/:id` body (FR-05.3) — every field optional. */
export const patchCampaignSchema = z
  .object({
    name: z.string().min(1).max(120),
    templateId: z.string().uuid(),
    segmentId: z.string().uuid(),
    variableMapping: campaignVariableMappingSchema,
    scheduledAt: z.string().datetime().nullable(),
  })
  .partial();
export type PatchCampaignDto = z.infer<typeof patchCampaignSchema>;

/** `POST /api/campaigns/:id/send` body (FR-05.6). */
export const sendCampaignSchema = z.object({
  /** Future ISO datetime to schedule; omitted/past => send immediately. */
  scheduledAt: z.string().datetime().optional(),
});
export type SendCampaignDto = z.infer<typeof sendCampaignSchema>;

/** `GET /api/campaigns/:id/recipients` query (FR-05.14). */
export const recipientsQuerySchema = z.object({
  status: z
    .enum(["pending", "sent", "delivered", "read", "failed", "skipped"])
    .optional(),
  take: z.coerce.number().int().min(1).max(200).default(50),
  skip: z.coerce.number().int().min(0).default(0),
});
export type RecipientsQueryDto = z.infer<typeof recipientsQuerySchema>;

/** `GET /api/campaigns` query (FR-05.14). */
export const campaignsQuerySchema = z.object({
  status: z
    .enum([
      "draft",
      "scheduled",
      "sending",
      "paused",
      "completed",
      "cancelled",
      "failed",
    ])
    .optional(),
});
export type CampaignsQueryDto = z.infer<typeof campaignsQuerySchema>;
```

- [ ] **Step 2: Typecheck**

Run: `pnpm --filter @whatapp/api typecheck`
Expected: PASS (the file is exercised by Tasks 10–12).

- [ ] **Step 3: Commit**

```bash
git add apps/api/src/campaigns/dto.ts
git commit -m "feat: add campaigns DTO schemas"
```

---

## Task 10: `CampaignsService` — CRUD, edit-guard, control, completion

**Files:**
- Create: `apps/api/src/campaigns/campaigns.service.ts`
- Create: `apps/api/src/campaigns/campaigns.service.test.ts`

Owns campaign lifecycle except the actual sending (Task 11). Covers FR-05.1
(create, `APPROVED`-template check — AC-05.1), FR-05.3 (edit only while
`draft`/`scheduled`), FR-05.11 (pause/resume/cancel), FR-05.12 (retry-failed),
FR-05.13 (`checkCompletion`), FR-05.14 (list/detail/recipients), FR-05.15
(stats via `aggregateCampaignStats`). Prisma is mocked with `vi.fn()` delegates.

> **Dependency note.** `retryFailed` and `start` need to enqueue jobs; that is
> `CampaignSendService` (Task 11). To avoid a circular import, `CampaignsService`
> does **not** import `CampaignSendService`. Instead: `retryFailed` resets the
> `failed` rows to `pending` and returns the list of recipient ids to re-enqueue
> — the **controller** (Task 12) hands them to `CampaignSendService.enqueueRecipients`.

- [ ] **Step 1: Write the failing test `campaigns.service.test.ts`** — covers:

```typescript
import { describe, it, expect, vi, beforeEach } from "vitest";
import { CampaignsService } from "./campaigns.service";

function makePrisma() {
  return {
    campaign: {
      create: vi.fn(),
      findUnique: vi.fn(),
      findMany: vi.fn(),
      update: vi.fn(),
    },
    template: { findUnique: vi.fn() },
    campaignRecipient: {
      findMany: vi.fn(),
      updateMany: vi.fn(),
      count: vi.fn(),
    },
  };
}

describe("CampaignsService", () => {
  let prisma: ReturnType<typeof makePrisma>;
  let svc: CampaignsService;

  beforeEach(() => {
    prisma = makePrisma();
    svc = new CampaignsService(prisma as never);
  });

  describe("create (FR-05.1, AC-05.1)", () => {
    it("creates a draft campaign for an APPROVED template", async () => {
      prisma.template.findUnique.mockResolvedValue({
        id: "t1",
        status: "APPROVED",
      });
      prisma.campaign.create.mockResolvedValue({ id: "c1", status: "draft" });

      const res = await svc.create(
        {
          name: "Promo",
          templateId: "t1",
          segmentId: "s1",
          variableMapping: {},
        },
        "user1",
      );

      expect(res).toMatchObject({ id: "c1", status: "draft" });
      expect(prisma.campaign.create).toHaveBeenCalledWith(
        expect.objectContaining({
          data: expect.objectContaining({
            name: "Promo",
            templateId: "t1",
            segmentId: "s1",
            status: "draft",
            createdBy: "user1",
          }),
        }),
      );
    });

    it("rejects a non-APPROVED template (AC-05.1)", async () => {
      prisma.template.findUnique.mockResolvedValue({
        id: "t1",
        status: "PENDING",
      });
      await expect(
        svc.create(
          {
            name: "Promo",
            templateId: "t1",
            segmentId: "s1",
            variableMapping: {},
          },
          "user1",
        ),
      ).rejects.toThrow(/APPROVED/);
    });

    it("rejects a missing template", async () => {
      prisma.template.findUnique.mockResolvedValue(null);
      await expect(
        svc.create(
          {
            name: "Promo",
            templateId: "missing",
            segmentId: "s1",
            variableMapping: {},
          },
          "user1",
        ),
      ).rejects.toThrow(/not found/i);
    });
  });

  describe("update (FR-05.3)", () => {
    it("edits a draft campaign", async () => {
      prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "draft",
      });
      prisma.campaign.update.mockResolvedValue({ id: "c1", name: "New" });
      const res = await svc.update("c1", { name: "New" });
      expect(res).toMatchObject({ name: "New" });
    });

    it("edits a scheduled campaign", async () => {
      prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "scheduled",
      });
      prisma.campaign.update.mockResolvedValue({ id: "c1" });
      await expect(svc.update("c1", { name: "x" })).resolves.toBeDefined();
    });

    it("rejects editing a sending campaign", async () => {
      prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "sending",
      });
      await expect(svc.update("c1", { name: "x" })).rejects.toThrow(
        /draft or scheduled/i,
      );
    });
  });

  describe("control (FR-05.11)", () => {
    it("pause moves a sending campaign to paused", async () => {
      prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "sending",
      });
      prisma.campaign.update.mockResolvedValue({ id: "c1", status: "paused" });
      const r = await svc.pause("c1");
      expect(r.status).toBe("paused");
    });

    it("pause rejects a non-sending campaign", async () => {
      prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "draft",
      });
      await expect(svc.pause("c1")).rejects.toThrow(/sending/i);
    });

    it("resume moves a paused campaign back to sending", async () => {
      prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "paused",
      });
      prisma.campaign.update.mockResolvedValue({ id: "c1", status: "sending" });
      const r = await svc.resume("c1");
      expect(r.status).toBe("sending");
    });

    it("resume rejects a non-paused campaign", async () => {
      prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "sending",
      });
      await expect(svc.resume("c1")).rejects.toThrow(/paused/i);
    });

    it("cancel marks a sending campaign cancelled", async () => {
      prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "sending",
      });
      prisma.campaign.update.mockResolvedValue({
        id: "c1",
        status: "cancelled",
      });
      const r = await svc.cancel("c1");
      expect(r.status).toBe("cancelled");
    });

    it("cancel rejects an already-completed campaign", async () => {
      prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "completed",
      });
      await expect(svc.cancel("c1")).rejects.toThrow(/cannot be cancelled/i);
    });
  });

  describe("retryFailed (FR-05.12)", () => {
    it("resets failed rows to pending and returns their ids", async () => {
      prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "completed",
      });
      prisma.campaignRecipient.findMany.mockResolvedValue([
        { id: "r1" },
        { id: "r2" },
      ]);
      prisma.campaignRecipient.updateMany.mockResolvedValue({ count: 2 });
      prisma.campaign.update.mockResolvedValue({ id: "c1" });

      const ids = await svc.retryFailed("c1");

      expect(ids).toEqual(["r1", "r2"]);
      expect(prisma.campaignRecipient.updateMany).toHaveBeenCalledWith({
        where: { campaignId: "c1", status: "failed" },
        data: { status: "pending", error: undefined, messageId: null },
      });
      // campaign moves back to sending so new sends are processed.
      expect(prisma.campaign.update).toHaveBeenCalledWith(
        expect.objectContaining({
          data: expect.objectContaining({ status: "sending" }),
        }),
      );
    });

    it("returns an empty list when there are no failed recipients", async () => {
      prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "completed",
      });
      prisma.campaignRecipient.findMany.mockResolvedValue([]);
      const ids = await svc.retryFailed("c1");
      expect(ids).toEqual([]);
    });
  });

  describe("checkCompletion (FR-05.13, AC-05.8)", () => {
    it("completes when every recipient is terminal", async () => {
      prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "sending",
      });
      prisma.campaignRecipient.findMany.mockResolvedValue([
        { status: "sent" },
        { status: "failed" },
        { status: "skipped" },
      ]);
      prisma.campaign.update.mockResolvedValue({});
      await svc.checkCompletion("c1");
      expect(prisma.campaign.update).toHaveBeenCalledWith(
        expect.objectContaining({
          data: expect.objectContaining({ status: "completed" }),
        }),
      );
    });

    it("does not complete while a recipient is still pending", async () => {
      prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "sending",
      });
      prisma.campaignRecipient.findMany.mockResolvedValue([
        { status: "sent" },
        { status: "pending" },
      ]);
      await svc.checkCompletion("c1");
      expect(prisma.campaign.update).not.toHaveBeenCalled();
    });

    it("does not re-complete an already-cancelled campaign", async () => {
      prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "cancelled",
      });
      prisma.campaignRecipient.findMany.mockResolvedValue([
        { status: "skipped" },
      ]);
      await svc.checkCompletion("c1");
      expect(prisma.campaign.update).not.toHaveBeenCalled();
    });
  });

  describe("recomputeStats (FR-05.15)", () => {
    it("writes aggregated stats from the recipient rows", async () => {
      prisma.campaignRecipient.findMany.mockResolvedValue([
        { status: "sent" },
        { status: "delivered" },
        { status: "pending" },
      ]);
      prisma.campaign.update.mockResolvedValue({});
      await svc.recomputeStats("c1");
      expect(prisma.campaign.update).toHaveBeenCalledWith({
        where: { id: "c1" },
        data: {
          stats: {
            total: 3,
            pending: 1,
            sent: 1,
            delivered: 1,
            read: 0,
            failed: 0,
            skipped: 0,
          },
        },
      });
    });
  });

  describe("list / get / recipients (FR-05.14)", () => {
    it("list filters by status when given", async () => {
      prisma.campaign.findMany.mockResolvedValue([]);
      await svc.list({ status: "sending" });
      expect(prisma.campaign.findMany).toHaveBeenCalledWith(
        expect.objectContaining({ where: { status: "sending" } }),
      );
    });

    it("get throws NotFound for an unknown id", async () => {
      prisma.campaign.findUnique.mockResolvedValue(null);
      await expect(svc.get("missing")).rejects.toThrow(/not found/i);
    });

    it("recipients filters by status and paginates", async () => {
      prisma.campaign.findUnique.mockResolvedValue({ id: "c1" });
      prisma.campaignRecipient.findMany.mockResolvedValue([]);
      prisma.campaignRecipient.count.mockResolvedValue(0);
      await svc.recipients("c1", { status: "failed", take: 50, skip: 0 });
      expect(prisma.campaignRecipient.findMany).toHaveBeenCalledWith(
        expect.objectContaining({
          where: { campaignId: "c1", status: "failed" },
          take: 50,
          skip: 0,
        }),
      );
    });
  });
});
```

- [ ] **Step 2: Run it — expect FAIL**

Run: `pnpm --filter @whatapp/api test campaigns.service`
Expected: FAIL — `CampaignsService` is not defined.

- [ ] **Step 3: Implement `apps/api/src/campaigns/campaigns.service.ts`:**

```typescript
import {
  Injectable,
  NotFoundException,
  BadRequestException,
} from "@nestjs/common";
import {
  aggregateCampaignStats,
  isCampaignComplete,
} from "@whatapp/shared";
import { PrismaService } from "../prisma/prisma.service";
import type {
  CreateCampaignDto,
  PatchCampaignDto,
  CampaignsQueryDto,
  RecipientsQueryDto,
} from "./dto";

/**
 * Campaign lifecycle: create, edit, the pause/resume/cancel/retry control
 * surface, completion detection, stats, and the reporting reads. The actual
 * audience resolution and queued sending live in `CampaignSendService`
 * (Task 11) to keep this class free of BullMQ.
 */
@Injectable()
export class CampaignsService {
  constructor(private readonly prisma: PrismaService) {}

  /** `POST /api/campaigns` — create a draft (FR-05.1, AC-05.1). */
  async create(dto: CreateCampaignDto, createdBy: string): Promise<unknown> {
    const template = await this.prisma.template.findUnique({
      where: { id: dto.templateId },
    });
    if (!template) {
      throw new NotFoundException(`Template ${dto.templateId} not found`);
    }
    if (template.status !== "APPROVED") {
      throw new BadRequestException(
        "A campaign can only use an APPROVED template",
      );
    }
    return this.prisma.campaign.create({
      data: {
        name: dto.name,
        templateId: dto.templateId,
        segmentId: dto.segmentId,
        variableMapping: dto.variableMapping as object,
        scheduledAt: dto.scheduledAt ? new Date(dto.scheduledAt) : null,
        status: dto.scheduledAt ? "scheduled" : "draft",
        createdBy,
      },
    });
  }

  /** `GET /api/campaigns` — list with optional status filter (FR-05.14). */
  async list(query: CampaignsQueryDto): Promise<unknown[]> {
    return this.prisma.campaign.findMany({
      where: query.status ? { status: query.status } : {},
      orderBy: { createdAt: "desc" },
      include: { template: true, segment: true },
    });
  }

  /** `GET /api/campaigns/:id` — full detail (FR-05.14). */
  async get(id: string): Promise<unknown> {
    const campaign = await this.prisma.campaign.findUnique({
      where: { id },
      include: { template: true, segment: true },
    });
    if (!campaign) throw new NotFoundException(`Campaign ${id} not found`);
    return campaign;
  }

  /** Loads a campaign or throws — internal helper. */
  private async require(id: string): Promise<{ id: string; status: string }> {
    const campaign = await this.prisma.campaign.findUnique({ where: { id } });
    if (!campaign) throw new NotFoundException(`Campaign ${id} not found`);
    return campaign as { id: string; status: string };
  }

  /** `PATCH /api/campaigns/:id` — edit only while draft/scheduled (FR-05.3). */
  async update(id: string, dto: PatchCampaignDto): Promise<unknown> {
    const campaign = await this.require(id);
    if (campaign.status !== "draft" && campaign.status !== "scheduled") {
      throw new BadRequestException(
        "A campaign can only be edited while draft or scheduled",
      );
    }
    const data: Record<string, unknown> = {};
    if (dto.name !== undefined) data["name"] = dto.name;
    if (dto.templateId !== undefined) data["templateId"] = dto.templateId;
    if (dto.segmentId !== undefined) data["segmentId"] = dto.segmentId;
    if (dto.variableMapping !== undefined) {
      data["variableMapping"] = dto.variableMapping as object;
    }
    if (dto.scheduledAt !== undefined) {
      data["scheduledAt"] = dto.scheduledAt
        ? new Date(dto.scheduledAt)
        : null;
    }
    return this.prisma.campaign.update({ where: { id }, data });
  }

  /** `POST /api/campaigns/:id/pause` (FR-05.11). */
  async pause(id: string): Promise<{ id: string; status: string }> {
    const campaign = await this.require(id);
    if (campaign.status !== "sending") {
      throw new BadRequestException("Only a sending campaign can be paused");
    }
    return this.prisma.campaign.update({
      where: { id },
      data: { status: "paused" },
    }) as Promise<{ id: string; status: string }>;
  }

  /** `POST /api/campaigns/:id/resume` (FR-05.11). */
  async resume(id: string): Promise<{ id: string; status: string }> {
    const campaign = await this.require(id);
    if (campaign.status !== "paused") {
      throw new BadRequestException("Only a paused campaign can be resumed");
    }
    return this.prisma.campaign.update({
      where: { id },
      data: { status: "sending" },
    }) as Promise<{ id: string; status: string }>;
  }

  /** `POST /api/campaigns/:id/cancel` (FR-05.11). */
  async cancel(id: string): Promise<{ id: string; status: string }> {
    const campaign = await this.require(id);
    if (
      campaign.status === "completed" ||
      campaign.status === "cancelled" ||
      campaign.status === "draft"
    ) {
      throw new BadRequestException(
        `A ${campaign.status} campaign cannot be cancelled`,
      );
    }
    return this.prisma.campaign.update({
      where: { id },
      data: { status: "cancelled" },
    }) as Promise<{ id: string; status: string }>;
  }

  /**
   * `POST /api/campaigns/:id/retry-failed` (FR-05.12). Resets every `failed`
   * recipient row to `pending`, moves the campaign back to `sending`, and
   * returns the recipient ids the controller must re-enqueue.
   */
  async retryFailed(id: string): Promise<string[]> {
    await this.require(id);
    const failed = await this.prisma.campaignRecipient.findMany({
      where: { campaignId: id, status: "failed" },
      select: { id: true },
    });
    if (failed.length === 0) return [];
    await this.prisma.campaignRecipient.updateMany({
      where: { campaignId: id, status: "failed" },
      data: { status: "pending", error: undefined, messageId: null },
    });
    await this.prisma.campaign.update({
      where: { id },
      data: { status: "sending", completedAt: null },
    });
    return failed.map((r: { id: string }) => r.id);
  }

  /**
   * Recomputes `stats` from the current recipient rows (FR-05.15). Called
   * after every recipient-status change.
   */
  async recomputeStats(id: string): Promise<void> {
    const rows = await this.prisma.campaignRecipient.findMany({
      where: { campaignId: id },
      select: { status: true },
    });
    const stats = aggregateCampaignStats(
      rows.map((r: { status: string }) => r.status),
    );
    await this.prisma.campaign.update({
      where: { id },
      data: { stats: stats as object },
    });
  }

  /**
   * Completes the campaign when every recipient is terminal (FR-05.13,
   * AC-05.8). A `completed`/`cancelled` campaign is left untouched.
   */
  async checkCompletion(id: string): Promise<void> {
    const campaign = await this.require(id);
    if (campaign.status === "completed" || campaign.status === "cancelled") {
      return;
    }
    const rows = await this.prisma.campaignRecipient.findMany({
      where: { campaignId: id },
      select: { status: true },
    });
    if (isCampaignComplete(rows.map((r: { status: string }) => r.status))) {
      await this.prisma.campaign.update({
        where: { id },
        data: { status: "completed", completedAt: new Date() },
      });
    }
  }

  /** `GET /api/campaigns/:id/recipients` — per-recipient list (FR-05.14). */
  async recipients(
    id: string,
    query: RecipientsQueryDto,
  ): Promise<{ items: unknown[]; total: number }> {
    await this.require(id);
    const where = {
      campaignId: id,
      ...(query.status ? { status: query.status } : {}),
    };
    const [items, total] = await Promise.all([
      this.prisma.campaignRecipient.findMany({
        where,
        orderBy: { createdAt: "asc" },
        take: query.take,
        skip: query.skip,
        include: { contact: true },
      }),
      this.prisma.campaignRecipient.count({ where }),
    ]);
    return { items, total };
  }
}
```

- [ ] **Step 4: Run it — expect PASS**

Run: `pnpm --filter @whatapp/api test campaigns.service`
Expected: PASS — every case green.

- [ ] **Step 5: Commit**

```bash
git add apps/api/src/campaigns/campaigns.service.ts \
  apps/api/src/campaigns/campaigns.service.test.ts
git commit -m "feat: add CampaignsService CRUD, control, completion, stats"
```

---

## Task 11: `CampaignSendService` — audience, enqueue, per-recipient send

**Files:**
- Create: `apps/api/src/campaigns/campaign-queue.provider.ts`
- Create: `apps/api/src/campaigns/campaign-send.service.ts`
- Create: `apps/api/src/campaign-send.service.test.ts` (see note — actually
  `apps/api/src/campaigns/campaign-send.service.test.ts`)

The heart of the module. `CampaignSendService`:
- `resolveAudience(campaignId)` — re-evaluates the segment via
  `compileSegmentDefinition` + `buildAudienceWhere` and returns the matched and
  sendable contacts (FR-05.4, FR-05.5).
- `schedule(campaignId, scheduledAt)` — for a future time, sets the campaign
  `scheduled` and adds a delayed `campaign-start` job (FR-05.6).
- `start(campaignId)` — moves the campaign to `sending`, sets `startedAt`,
  creates one `pending` `campaign_recipients` row per sendable contact, and adds
  one `campaign-send` job per recipient (FR-05.6, FR-05.7).
- `enqueueRecipients(campaignId, recipientIds)` — adds `campaign-send` jobs for
  a given set of recipients (used by retry-failed).
- `sendOneRecipient(recipientId)` — resolves the variable mapping, calls
  `MessageService.send` with the template, updates the recipient row, returns a
  structured `SendRecipientOutcome` (FR-05.8, FR-05.9).

`MetaSendResult.error.code === 130429` is the only retriable error.

> **Worker↔API note.** The recipient processor (Task 7) calls
> `sendOneRecipient` through the internal endpoint (Task 12). `start` and
> `schedule` are also reachable via the internal endpoint so the worker's
> `campaign-start` job can call them.

- [ ] **Step 1: Implement `apps/api/src/campaigns/campaign-queue.provider.ts`:**

```typescript
import { Queue } from "bullmq";
import {
  QUEUE_CAMPAIGN_SEND,
  QUEUE_CAMPAIGN_START,
  createRedisConnection,
} from "@whatapp/worker/queues";

/** DI tokens for the two campaign BullMQ queues. */
export const CAMPAIGN_SEND_QUEUE = Symbol("CAMPAIGN_SEND_QUEUE");
export const CAMPAIGN_START_QUEUE = Symbol("CAMPAIGN_START_QUEUE");

/**
 * NestJS providers for the campaign queues. The API only *adds* jobs; the
 * worker consumes them. Queue construction does not need the rate limiter
 * (that is a Worker-side option) — the API enqueues freely and the worker's
 * limiter paces consumption (FR-05.7).
 */
export const campaignQueueProviders = [
  {
    provide: CAMPAIGN_SEND_QUEUE,
    useFactory: () =>
      new Queue(QUEUE_CAMPAIGN_SEND, { connection: createRedisConnection() }),
  },
  {
    provide: CAMPAIGN_START_QUEUE,
    useFactory: () =>
      new Queue(QUEUE_CAMPAIGN_START, { connection: createRedisConnection() }),
  },
];
```

> If `@whatapp/worker/queues` is not an exposed subpath, mirror the existing
> Phase 1 approach in `apps/api/src/webhooks/webhook-queue.provider.ts` — it
> already imports queue-name constants. Match whatever import path that file
> uses (it is the established pattern); the constant *names*
> (`QUEUE_CAMPAIGN_SEND`, `QUEUE_CAMPAIGN_START`, `CAMPAIGN_SEND_JOB_OPTS`,
> `createRedisConnection`) are unchanged regardless of path.

- [ ] **Step 2: Write the failing test
  `apps/api/src/campaigns/campaign-send.service.test.ts`:**

```typescript
import { describe, it, expect, vi, beforeEach } from "vitest";
import { CampaignSendService } from "./campaign-send.service";

function makeDeps() {
  const prisma = {
    campaign: { findUnique: vi.fn(), update: vi.fn() },
    contact: { findMany: vi.fn(), count: vi.fn() },
    campaignRecipient: {
      createMany: vi.fn(),
      findMany: vi.fn(),
      findUnique: vi.fn(),
      update: vi.fn(),
    },
  };
  const sendQueue = { add: vi.fn().mockResolvedValue({}) };
  const startQueue = { add: vi.fn().mockResolvedValue({}) };
  const messageService = { send: vi.fn() };
  const campaignsService = {
    recomputeStats: vi.fn().mockResolvedValue(undefined),
    checkCompletion: vi.fn().mockResolvedValue(undefined),
  };
  return { prisma, sendQueue, startQueue, messageService, campaignsService };
}

const APPROVED_TEMPLATE = {
  id: "t1",
  name: "promo",
  language: "en",
  status: "APPROVED",
  category: "MARKETING",
  variables: ["1"],
};

describe("CampaignSendService", () => {
  let d: ReturnType<typeof makeDeps>;
  let svc: CampaignSendService;

  beforeEach(() => {
    d = makeDeps();
    svc = new CampaignSendService(
      d.prisma as never,
      d.sendQueue as never,
      d.startQueue as never,
      d.messageService as never,
      d.campaignsService as never,
    );
  });

  describe("resolveAudience (FR-05.4, FR-05.5)", () => {
    it("returns matched and sendable counts excluding blocked/opted-out", async () => {
      d.prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        segment: { definition: { tags: ["vip"] } },
        template: APPROVED_TEMPLATE,
      });
      // matched (segment-only) and sendable (segment + consent/block).
      d.prisma.contact.count
        .mockResolvedValueOnce(100) // matched
        .mockResolvedValueOnce(82); // sendable
      const res = await svc.resolveAudience("c1");
      expect(res).toMatchObject({ matched: 100, sendable: 82 });
    });
  });

  describe("schedule (FR-05.6)", () => {
    it("sets the campaign scheduled and adds a delayed start job", async () => {
      d.prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "draft",
      });
      d.prisma.campaign.update.mockResolvedValue({});
      const future = new Date(Date.now() + 3_600_000);
      await svc.schedule("c1", future);
      expect(d.prisma.campaign.update).toHaveBeenCalledWith(
        expect.objectContaining({
          data: expect.objectContaining({
            status: "scheduled",
            scheduledAt: future,
          }),
        }),
      );
      expect(d.startQueue.add).toHaveBeenCalledWith(
        "campaign-start",
        { campaignId: "c1" },
        expect.objectContaining({ delay: expect.any(Number) }),
      );
    });
  });

  describe("start (FR-05.6, FR-05.7, AC-05.3)", () => {
    it("creates one pending recipient row per sendable contact and enqueues", async () => {
      d.prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "draft",
        segment: { definition: { tags: ["vip"] } },
        template: APPROVED_TEMPLATE,
      });
      d.prisma.contact.findMany.mockResolvedValue([
        { id: "ct1", waId: "971500001" },
        { id: "ct2", waId: "971500002" },
      ]);
      d.prisma.campaign.update.mockResolvedValue({});
      d.prisma.campaignRecipient.createMany.mockResolvedValue({ count: 2 });
      d.prisma.campaignRecipient.findMany.mockResolvedValue([
        { id: "r1" },
        { id: "r2" },
      ]);

      await svc.start("c1");

      expect(d.prisma.campaign.update).toHaveBeenCalledWith(
        expect.objectContaining({
          data: expect.objectContaining({
            status: "sending",
            startedAt: expect.any(Date),
          }),
        }),
      );
      expect(d.prisma.campaignRecipient.createMany).toHaveBeenCalledWith(
        expect.objectContaining({
          data: [
            { campaignId: "c1", contactId: "ct1", waId: "971500001" },
            { campaignId: "c1", contactId: "ct2", waId: "971500002" },
          ],
          skipDuplicates: true,
        }),
      );
      expect(d.sendQueue.add).toHaveBeenCalledTimes(2);
      expect(d.campaignsService.recomputeStats).toHaveBeenCalledWith("c1");
    });

    it("refuses to start a campaign that is not draft/scheduled", async () => {
      d.prisma.campaign.findUnique.mockResolvedValue({
        id: "c1",
        status: "sending",
        segment: { definition: {} },
        template: APPROVED_TEMPLATE,
      });
      await expect(svc.start("c1")).rejects.toThrow(/cannot be started/i);
    });
  });

  describe("sendOneRecipient (FR-05.8, FR-05.9, AC-05.4, AC-05.5)", () => {
    function recipientRow(over = {}) {
      return {
        id: "r1",
        campaignId: "c1",
        contactId: "ct1",
        waId: "971500001",
        status: "pending",
        contact: {
          waId: "971500001",
          displayName: "Ahmed",
          profileName: null,
          phoneE164: "+971500001",
          country: "AE",
          attributes: {},
        },
        campaign: {
          id: "c1",
          status: "sending",
          variableMapping: { "1": { kind: "fixed", value: "Hi" } },
          template: APPROVED_TEMPLATE,
        },
        ...over,
      };
    }

    it("skips a recipient missing a required variable (AC-05.4)", async () => {
      d.prisma.campaignRecipient.findUnique.mockResolvedValue(
        recipientRow({
          campaign: {
            id: "c1",
            status: "sending",
            variableMapping: {
              "1": { kind: "field", path: "attributes.missing" },
            },
            template: APPROVED_TEMPLATE,
          },
        }),
      );
      d.prisma.campaignRecipient.update.mockResolvedValue({});

      const res = await svc.sendOneRecipient("r1");

      expect(res).toEqual({ result: "skipped", retriable: false });
      expect(d.messageService.send).not.toHaveBeenCalled();
      expect(d.prisma.campaignRecipient.update).toHaveBeenCalledWith(
        expect.objectContaining({
          data: expect.objectContaining({
            status: "skipped",
            skipReason: "missing_variable",
          }),
        }),
      );
    });

    it("sends a template message and marks the recipient sent", async () => {
      d.prisma.campaignRecipient.findUnique.mockResolvedValue(recipientRow());
      d.prisma.campaignRecipient.update.mockResolvedValue({});
      d.messageService.send.mockResolvedValue({ ok: true, wamid: "wamid-1" });

      const res = await svc.sendOneRecipient("r1");

      expect(res).toEqual({ result: "sent", retriable: false });
      expect(d.messageService.send).toHaveBeenCalledWith(
        expect.objectContaining({ type: "template", to: "971500001" }),
        expect.objectContaining({ campaignId: "c1", templateId: "t1" }),
      );
      expect(d.prisma.campaignRecipient.update).toHaveBeenCalledWith(
        expect.objectContaining({
          data: expect.objectContaining({
            status: "sent",
            messageId: "wamid-1",
          }),
        }),
      );
    });

    it("marks a recipient failed on a permanent Meta error", async () => {
      d.prisma.campaignRecipient.findUnique.mockResolvedValue(recipientRow());
      d.prisma.campaignRecipient.update.mockResolvedValue({});
      d.messageService.send.mockResolvedValue({
        ok: false,
        error: { code: 131026, title: "undeliverable", message: "bad" },
      });

      const res = await svc.sendOneRecipient("r1");

      expect(res).toEqual({ result: "failed", retriable: false });
      expect(d.prisma.campaignRecipient.update).toHaveBeenCalledWith(
        expect.objectContaining({
          data: expect.objectContaining({ status: "failed" }),
        }),
      );
    });

    it("returns retriable=true on a Meta 130429 rate error (FR-05.8)", async () => {
      d.prisma.campaignRecipient.findUnique.mockResolvedValue(recipientRow());
      d.messageService.send.mockResolvedValue({
        ok: false,
        error: { code: 130429, title: "rate", message: "slow down" },
      });

      const res = await svc.sendOneRecipient("r1");

      expect(res).toEqual({ result: "failed", retriable: true });
      // The row is NOT marked failed on a retriable error — it stays pending.
      expect(d.prisma.campaignRecipient.update).not.toHaveBeenCalled();
    });

    it("is idempotent — an already-sent recipient is not re-sent (AC-05.5)", async () => {
      d.prisma.campaignRecipient.findUnique.mockResolvedValue(
        recipientRow({ status: "sent" }),
      );
      const res = await svc.sendOneRecipient("r1");
      expect(res).toEqual({ result: "already_done", retriable: false });
      expect(d.messageService.send).not.toHaveBeenCalled();
    });

    it("does not send when the campaign is paused", async () => {
      d.prisma.campaignRecipient.findUnique.mockResolvedValue(
        recipientRow({
          campaign: {
            id: "c1",
            status: "paused",
            variableMapping: { "1": { kind: "fixed", value: "Hi" } },
            template: APPROVED_TEMPLATE,
          },
        }),
      );
      const res = await svc.sendOneRecipient("r1");
      expect(res).toEqual({ result: "already_done", retriable: false });
      expect(d.messageService.send).not.toHaveBeenCalled();
    });

    it("recomputes stats and checks completion after a terminal outcome", async () => {
      d.prisma.campaignRecipient.findUnique.mockResolvedValue(recipientRow());
      d.prisma.campaignRecipient.update.mockResolvedValue({});
      d.messageService.send.mockResolvedValue({ ok: true, wamid: "wamid-1" });
      await svc.sendOneRecipient("r1");
      expect(d.campaignsService.recomputeStats).toHaveBeenCalledWith("c1");
      expect(d.campaignsService.checkCompletion).toHaveBeenCalledWith("c1");
    });
  });
});
```

- [ ] **Step 3: Run it — expect FAIL**

Run: `pnpm --filter @whatapp/api test campaign-send.service`
Expected: FAIL — `CampaignSendService` is not defined.

- [ ] **Step 4: Implement `apps/api/src/campaigns/campaign-send.service.ts`:**

```typescript
import {
  Injectable,
  Inject,
  NotFoundException,
  BadRequestException,
} from "@nestjs/common";
import type { Queue } from "bullmq";
import {
  compileSegmentDefinition,
  buildAudienceWhere,
  resolveVariableMapping,
  type AudienceTemplateCategory,
  type SegmentDefinition,
} from "@whatapp/shared";
import { CAMPAIGN_SEND_JOB_OPTS } from "@whatapp/worker/queues";
import { PrismaService } from "../prisma/prisma.service";
import { MessageService } from "../messages/message.service";
import { CampaignsService } from "./campaigns.service";
import {
  CAMPAIGN_SEND_QUEUE,
  CAMPAIGN_START_QUEUE,
} from "./campaign-queue.provider";

/** The structured outcome of sending one recipient (mirrors the worker type). */
export interface SendRecipientOutcome {
  result: "sent" | "failed" | "skipped" | "already_done";
  retriable: boolean;
}

/** Audience-resolution result for the builder preview (FR-05.4). */
export interface AudienceResult {
  matched: number;
  sendable: number;
}

/** Meta error code for a rate-limit hit — the one retriable error (§1.6). */
const META_RATE_LIMIT_CODE = 130429;

/**
 * Audience resolution, recipient-row creation, rate-limited enqueue, and the
 * per-recipient send. The send itself goes through the Phase 1
 * `MessageService.send` — never reimplemented here.
 */
@Injectable()
export class CampaignSendService {
  constructor(
    private readonly prisma: PrismaService,
    @Inject(CAMPAIGN_SEND_QUEUE) private readonly sendQueue: Queue,
    @Inject(CAMPAIGN_START_QUEUE) private readonly startQueue: Queue,
    private readonly messages: MessageService,
    private readonly campaigns: CampaignsService,
  ) {}

  /** Loads a campaign with its template + segment, or throws. */
  private async loadCampaign(id: string): Promise<{
    id: string;
    status: string;
    variableMapping: unknown;
    segment: { definition: unknown } | null;
    template: {
      id: string;
      name: string;
      language: string;
      category: string;
      variables: unknown;
    };
  }> {
    const campaign = await this.prisma.campaign.findUnique({
      where: { id },
      include: { segment: true, template: true },
    });
    if (!campaign) throw new NotFoundException(`Campaign ${id} not found`);
    return campaign as never;
  }

  /**
   * Re-evaluates the segment now (FR-05.5) and returns the matched count and
   * the sendable count (matched minus blocked minus opted-out — FR-05.4).
   */
  async resolveAudience(campaignId: string): Promise<AudienceResult> {
    const campaign = await this.loadCampaign(campaignId);
    if (!campaign.segment) {
      throw new BadRequestException("Campaign has no segment");
    }
    const segWhere = compileSegmentDefinition(
      campaign.segment.definition as SegmentDefinition,
    );
    const matched = await this.prisma.contact.count({ where: segWhere });
    const sendableWhere = buildAudienceWhere(
      segWhere,
      campaign.template.category as AudienceTemplateCategory,
    );
    const sendable = await this.prisma.contact.count({
      where: sendableWhere as never,
    });
    return { matched, sendable };
  }

  /**
   * Schedules a campaign for a future time (FR-05.6): status -> scheduled and a
   * delayed `campaign-start` job fires at `scheduledAt`.
   */
  async schedule(campaignId: string, scheduledAt: Date): Promise<void> {
    const campaign = await this.loadCampaign(campaignId);
    if (campaign.status !== "draft" && campaign.status !== "scheduled") {
      throw new BadRequestException(
        `A ${campaign.status} campaign cannot be scheduled`,
      );
    }
    await this.prisma.campaign.update({
      where: { id: campaignId },
      data: { status: "scheduled", scheduledAt },
    });
    const delay = Math.max(0, scheduledAt.getTime() - Date.now());
    await this.startQueue.add(
      "campaign-start",
      { campaignId },
      { delay, removeOnComplete: true, removeOnFail: false },
    );
  }

  /**
   * Starts a campaign now (FR-05.6, FR-05.7, AC-05.3): resolves the audience,
   * creates one `pending` recipient row per sendable contact, enqueues one
   * `campaign-send` job per recipient, and refreshes stats.
   */
  async start(campaignId: string): Promise<void> {
    const campaign = await this.loadCampaign(campaignId);
    if (campaign.status !== "draft" && campaign.status !== "scheduled") {
      throw new BadRequestException(
        `A ${campaign.status} campaign cannot be started`,
      );
    }
    if (!campaign.segment) {
      throw new BadRequestException("Campaign has no segment");
    }
    const segWhere = compileSegmentDefinition(
      campaign.segment.definition as SegmentDefinition,
    );
    const sendableWhere = buildAudienceWhere(
      segWhere,
      campaign.template.category as AudienceTemplateCategory,
    );
    const contacts: { id: string; waId: string }[] =
      await this.prisma.contact.findMany({
        where: sendableWhere as never,
        select: { id: true, waId: true },
      });

    await this.prisma.campaign.update({
      where: { id: campaignId },
      data: { status: "sending", startedAt: new Date() },
    });
    await this.prisma.campaignRecipient.createMany({
      data: contacts.map((c) => ({
        campaignId,
        contactId: c.id,
        waId: c.waId,
      })),
      skipDuplicates: true,
    });

    const rows: { id: string }[] =
      await this.prisma.campaignRecipient.findMany({
        where: { campaignId, status: "pending" },
        select: { id: true },
      });
    await this.enqueueRecipients(campaignId, rows.map((r) => r.id));
    await this.campaigns.recomputeStats(campaignId);
    await this.campaigns.checkCompletion(campaignId);
  }

  /** Adds one rate-limited `campaign-send` job per recipient (FR-05.7). */
  async enqueueRecipients(
    _campaignId: string,
    recipientIds: string[],
  ): Promise<void> {
    for (const recipientId of recipientIds) {
      await this.sendQueue.add(
        "campaign-send",
        { recipientId },
        CAMPAIGN_SEND_JOB_OPTS,
      );
    }
  }

  /**
   * Sends one recipient (FR-05.8, FR-05.9). Idempotent: a non-`pending` row, or
   * a campaign that is not `sending`, returns `already_done` without sending. A
   * missing required variable -> `skipped`. A Meta `130429` -> `retriable=true`
   * and the row is left `pending` for the BullMQ retry. Any other Meta error ->
   * the row is `failed`.
   */
  async sendOneRecipient(recipientId: string): Promise<SendRecipientOutcome> {
    const recipient = await this.prisma.campaignRecipient.findUnique({
      where: { id: recipientId },
      include: {
        contact: true,
        campaign: { include: { template: true } },
      },
    });
    if (!recipient) {
      throw new NotFoundException(`Recipient ${recipientId} not found`);
    }
    const r = recipient as never as {
      id: string;
      status: string;
      waId: string;
      contact: {
        waId: string;
        displayName: string | null;
        profileName: string | null;
        phoneE164: string | null;
        country: string | null;
        attributes: Record<string, unknown>;
      };
      campaign: {
        id: string;
        status: string;
        variableMapping: Record<string, never>;
        template: {
          id: string;
          name: string;
          language: string;
          variables: string[];
        };
      };
    };

    // Idempotency (FR-05.9, AC-05.5) — only a pending row in a sending/paused
    // campaign is eligible; anything else is a no-op. A paused campaign holds.
    if (r.status !== "pending" || r.campaign.status !== "sending") {
      return { result: "already_done", retriable: false };
    }

    // Resolve the per-recipient variable mapping (FR-05.2, AC-05.4).
    const requiredVars = (r.campaign.template.variables ?? []).map((v) =>
      String(v),
    );
    const resolved = resolveVariableMapping(
      r.campaign.variableMapping,
      r.contact,
      requiredVars,
    );
    if (!resolved.ok) {
      await this.prisma.campaignRecipient.update({
        where: { id: recipientId },
        data: { status: "skipped", skipReason: resolved.skipReason },
      });
      await this.campaigns.recomputeStats(r.campaign.id);
      await this.campaigns.checkCompletion(r.campaign.id);
      return { result: "skipped", retriable: false };
    }

    // Build the template OutboundMessage and send via the Phase 1 path.
    const components = [
      {
        type: "body" as const,
        parameters: requiredVars.map((position) => ({
          type: "text" as const,
          text: resolved.values[position] ?? "",
        })),
      },
    ];
    const result = await this.messages.send(
      {
        to: r.waId,
        type: "template",
        content: {
          name: r.campaign.template.name,
          language: r.campaign.template.language,
          components,
        },
      },
      { campaignId: r.campaign.id, templateId: r.campaign.template.id },
    );

    if (result.ok) {
      await this.prisma.campaignRecipient.update({
        where: { id: recipientId },
        data: {
          status: "sent",
          messageId: result.wamid ?? null,
          sentAt: new Date(),
          error: undefined,
        },
      });
      await this.campaigns.recomputeStats(r.campaign.id);
      await this.campaigns.checkCompletion(r.campaign.id);
      return { result: "sent", retriable: false };
    }

    // A 130429 rate error is retriable — leave the row pending for BullMQ.
    if (result.error?.code === META_RATE_LIMIT_CODE) {
      return { result: "failed", retriable: true };
    }

    // Any other Meta error fails this recipient only (FR-05.8).
    await this.prisma.campaignRecipient.update({
      where: { id: recipientId },
      data: { status: "failed", error: (result.error as object) ?? undefined },
    });
    await this.campaigns.recomputeStats(r.campaign.id);
    await this.campaigns.checkCompletion(r.campaign.id);
    return { result: "failed", retriable: false };
  }
}
```

> **Type check the `OutboundMessage` shape.** The exact field names for a
> `type: "template"` message (`content.name` / `content.language` /
> `content.components`) must match the `OutboundMessage` union in
> `packages/shared` that `MessageService.send` consumes — `message.service.ts`
> reads `msg.content.name` for a template, confirming `content.name`. If the
> shared type names the language field differently (e.g. a nested
> `language: { code }`), adjust the object built here to match it exactly. Do
> not change the shared type.

- [ ] **Step 5: Run it — expect PASS**

Run: `pnpm --filter @whatapp/api test campaign-send.service`
Expected: PASS — every case green.

- [ ] **Step 6: Commit**

```bash
git add apps/api/src/campaigns/campaign-queue.provider.ts \
  apps/api/src/campaigns/campaign-send.service.ts \
  apps/api/src/campaigns/campaign-send.service.test.ts
git commit -m "feat: add CampaignSendService audience, enqueue, per-recipient send"
```

---

## Task 12: Campaigns controllers + module wiring + webhook reflection dep

**Files:**
- Create: `apps/api/src/campaigns/campaigns.controller.ts`
- Create: `apps/api/src/campaigns/campaigns.controller.test.ts`
- Create: `apps/api/src/campaigns/internal-campaigns.controller.ts`
- Create: `apps/api/src/campaigns/internal-campaigns.controller.test.ts`
- Create: `apps/api/src/campaigns/campaigns.module.ts`
- Modify: `apps/api/src/app.module.ts`
- Modify: `apps/api/src/webhooks/webhook-processor.service.ts`
- Modify: `apps/api/src/webhooks/webhooks.module.ts` (if it needs the
  campaigns providers — see Step 5)

Exposes the API surface, registers the module, and wires the FR-05.10 webhook
reflection dep into `processEnvelope`. Follow the controller pattern in
`apps/api/src/templates/templates.controller.ts` (Zod-parsed bodies,
`@Roles()` guards) and `apps/api/src/messages/internal-messages.controller.ts`
(`CallbackAuthGuard`).

- [ ] **Step 1: Write the failing test `campaigns.controller.test.ts`:**

```typescript
import { describe, it, expect, vi, beforeEach } from "vitest";
import { CampaignsController } from "./campaigns.controller";

function makeServices() {
  return {
    campaigns: {
      create: vi.fn(),
      list: vi.fn(),
      get: vi.fn(),
      update: vi.fn(),
      pause: vi.fn(),
      resume: vi.fn(),
      cancel: vi.fn(),
      retryFailed: vi.fn(),
      recipients: vi.fn(),
    },
    send: {
      resolveAudience: vi.fn(),
      schedule: vi.fn(),
      start: vi.fn(),
      enqueueRecipients: vi.fn(),
    },
  };
}

describe("CampaignsController", () => {
  let s: ReturnType<typeof makeServices>;
  let ctrl: CampaignsController;

  beforeEach(() => {
    s = makeServices();
    ctrl = new CampaignsController(s.campaigns as never, s.send as never);
  });

  it("POST /campaigns validates the body and creates (FR-05.1)", async () => {
    s.campaigns.create.mockResolvedValue({ id: "c1" });
    const res = await ctrl.create(
      {
        name: "Promo",
        templateId: "11111111-1111-1111-1111-111111111111",
        segmentId: "22222222-2222-2222-2222-222222222222",
      },
      { user: { id: "u1" } } as never,
    );
    expect(res).toEqual({ id: "c1" });
    expect(s.campaigns.create).toHaveBeenCalled();
  });

  it("POST /campaigns rejects an invalid body", async () => {
    await expect(
      ctrl.create({ name: "" } as never, { user: { id: "u1" } } as never),
    ).rejects.toThrow();
  });

  it("GET /campaigns/:id/audience resolves the audience (FR-05.4)", async () => {
    s.send.resolveAudience.mockResolvedValue({ matched: 10, sendable: 8 });
    const res = await ctrl.audience("c1");
    expect(res).toEqual({ matched: 10, sendable: 8 });
  });

  it("POST /campaigns/:id/send with no schedule starts now (FR-05.6)", async () => {
    s.send.start.mockResolvedValue(undefined);
    await ctrl.send("c1", {});
    expect(s.send.start).toHaveBeenCalledWith("c1");
  });

  it("POST /campaigns/:id/send with a future time schedules (FR-05.6)", async () => {
    s.send.schedule.mockResolvedValue(undefined);
    const future = new Date(Date.now() + 3_600_000).toISOString();
    await ctrl.send("c1", { scheduledAt: future });
    expect(s.send.schedule).toHaveBeenCalledWith("c1", new Date(future));
  });

  it("POST /campaigns/:id/retry-failed re-enqueues failed recipients (FR-05.12)", async () => {
    s.campaigns.retryFailed.mockResolvedValue(["r1", "r2"]);
    s.send.enqueueRecipients.mockResolvedValue(undefined);
    await ctrl.retryFailed("c1");
    expect(s.send.enqueueRecipients).toHaveBeenCalledWith("c1", ["r1", "r2"]);
  });

  it("POST /campaigns/:id/pause delegates to the service (FR-05.11)", async () => {
    s.campaigns.pause.mockResolvedValue({ id: "c1", status: "paused" });
    const res = await ctrl.pause("c1");
    expect(res).toMatchObject({ status: "paused" });
  });
});
```

- [ ] **Step 2: Implement `apps/api/src/campaigns/campaigns.controller.ts`:**

```typescript
import {
  Controller,
  Get,
  Post,
  Patch,
  Param,
  Body,
  Query,
  Req,
  UseGuards,
} from "@nestjs/common";
import { JwtAuthGuard } from "../auth/jwt-auth.guard";
import { Roles } from "../auth/roles.decorator";
import { RolesGuard } from "../auth/roles.guard";
import { CampaignsService } from "./campaigns.service";
import { CampaignSendService } from "./campaign-send.service";
import {
  createCampaignSchema,
  patchCampaignSchema,
  sendCampaignSchema,
  campaignsQuerySchema,
  recipientsQuerySchema,
} from "./dto";

/**
 * `/api/campaigns` — the campaign control surface. Reads need `viewer`;
 * mutations need `marketing` (admins inherit). Bodies are Zod-parsed at the
 * boundary — invalid input throws before any service call.
 */
@Controller("api/campaigns")
@UseGuards(JwtAuthGuard, RolesGuard)
export class CampaignsController {
  constructor(
    private readonly campaigns: CampaignsService,
    private readonly send: CampaignSendService,
  ) {}

  @Get()
  @Roles("viewer", "marketing", "admin")
  async list(@Query() query: unknown): Promise<unknown[]> {
    return this.campaigns.list(campaignsQuerySchema.parse(query ?? {}));
  }

  @Post()
  @Roles("marketing", "admin")
  async create(
    @Body() body: unknown,
    @Req() req: { user: { id: string } },
  ): Promise<unknown> {
    const dto = createCampaignSchema.parse(body);
    return this.campaigns.create(dto, req.user.id);
  }

  @Get(":id")
  @Roles("viewer", "marketing", "admin")
  async get(@Param("id") id: string): Promise<unknown> {
    return this.campaigns.get(id);
  }

  @Patch(":id")
  @Roles("marketing", "admin")
  async update(
    @Param("id") id: string,
    @Body() body: unknown,
  ): Promise<unknown> {
    return this.campaigns.update(id, patchCampaignSchema.parse(body));
  }

  @Get(":id/audience")
  @Roles("viewer", "marketing", "admin")
  async audience(@Param("id") id: string): Promise<unknown> {
    return this.send.resolveAudience(id);
  }

  @Post(":id/send")
  @Roles("marketing", "admin")
  async send_(
    @Param("id") id: string,
    @Body() body: unknown,
  ): Promise<{ ok: true }> {
    return this.send(id, body);
  }

  /** Shared by send_ and the test — schedule when future, else start now. */
  async send(id: string, body: unknown): Promise<{ ok: true }> {
    const dto = sendCampaignSchema.parse(body);
    if (dto.scheduledAt && new Date(dto.scheduledAt).getTime() > Date.now()) {
      await this.send.schedule(id, new Date(dto.scheduledAt));
    } else {
      await this.send.start(id);
    }
    return { ok: true };
  }

  @Post(":id/pause")
  @Roles("marketing", "admin")
  async pause(@Param("id") id: string): Promise<unknown> {
    return this.campaigns.pause(id);
  }

  @Post(":id/resume")
  @Roles("marketing", "admin")
  async resume(@Param("id") id: string): Promise<unknown> {
    return this.campaigns.resume(id);
  }

  @Post(":id/cancel")
  @Roles("marketing", "admin")
  async cancel(@Param("id") id: string): Promise<unknown> {
    return this.campaigns.cancel(id);
  }

  @Post(":id/retry-failed")
  @Roles("marketing", "admin")
  async retryFailed(@Param("id") id: string): Promise<{ requeued: number }> {
    const ids = await this.campaigns.retryFailed(id);
    await this.send.enqueueRecipients(id, ids);
    return { requeued: ids.length };
  }

  @Get(":id/recipients")
  @Roles("viewer", "marketing", "admin")
  async recipients(
    @Param("id") id: string,
    @Query() query: unknown,
  ): Promise<unknown> {
    return this.campaigns.recipients(
      id,
      recipientsQuerySchema.parse(query ?? {}),
    );
  }
}
```

> **Naming note.** NestJS routes the decorated method; the test calls the
> plain methods. Keep one route method per endpoint. To avoid the `send`
> name colliding with the injected `send` service, the route handler is
> `send_` (decorated `@Post(":id/send")`) delegating to a plain `send(id, body)`
> method that the test drives directly. Confirm `@Roles` / guard decorator
> names against `apps/api/src/templates/templates.controller.ts` and adjust
> imports to the exact paths used there (`auth/roles.decorator`,
> `auth/jwt-auth.guard` — match the real files).

- [ ] **Step 3: Run the controller test — expect PASS**

Run: `pnpm --filter @whatapp/api test campaigns.controller`
Expected: PASS.

- [ ] **Step 4: Write + implement the internal controller**

Test `internal-campaigns.controller.test.ts`:

```typescript
import { describe, it, expect, vi, beforeEach } from "vitest";
import { InternalCampaignsController } from "./internal-campaigns.controller";

describe("InternalCampaignsController", () => {
  let send: { start: ReturnType<typeof vi.fn>; sendOneRecipient: ReturnType<typeof vi.fn> };
  let ctrl: InternalCampaignsController;

  beforeEach(() => {
    send = { start: vi.fn(), sendOneRecipient: vi.fn() };
    ctrl = new InternalCampaignsController(send as never);
  });

  it("POST /internal/campaigns/:id/start delegates to start", async () => {
    send.start.mockResolvedValue(undefined);
    const res = await ctrl.start("c1");
    expect(send.start).toHaveBeenCalledWith("c1");
    expect(res).toEqual({ ok: true });
  });

  it("POST /internal/.../recipients/:id/send returns the outcome", async () => {
    send.sendOneRecipient.mockResolvedValue({
      result: "sent",
      retriable: false,
    });
    const res = await ctrl.sendRecipient("r1");
    expect(res).toEqual({ result: "sent", retriable: false });
  });
});
```

Implement `internal-campaigns.controller.ts`:

```typescript
import { Controller, Post, Param, UseGuards } from "@nestjs/common";
import { CallbackAuthGuard } from "../messages/callback-auth.guard";
import { CampaignSendService } from "./campaign-send.service";
import type { SendRecipientOutcome } from "./campaign-send.service";

/**
 * Internal campaign endpoints the BullMQ worker calls (the worker has no Nest
 * DI). Guarded by `CallbackAuthGuard` — every request must carry
 * `Authorization: Bearer {CALLBACK_SECRET}`. These are the worker's only entry
 * into the campaign send path, keeping `MessageService` the single send path.
 */
@Controller("internal/campaigns")
@UseGuards(CallbackAuthGuard)
export class InternalCampaignsController {
  constructor(private readonly send: CampaignSendService) {}

  /** Triggered by a `campaign-start` job — resolve audience + enqueue. */
  @Post(":id/start")
  async start(@Param("id") id: string): Promise<{ ok: true }> {
    await this.send.start(id);
    return { ok: true };
  }

  /** Triggered by a `campaign-send` job — send one recipient. */
  @Post("recipients/:recipientId/send")
  async sendRecipient(
    @Param("recipientId") recipientId: string,
  ): Promise<SendRecipientOutcome> {
    return this.send.sendOneRecipient(recipientId);
  }
}
```

Run: `pnpm --filter @whatapp/api test internal-campaigns.controller`
Expected: PASS.

- [ ] **Step 5: Implement `apps/api/src/campaigns/campaigns.module.ts`:**

```typescript
import { Module } from "@nestjs/common";
import { PrismaModule } from "../prisma/prisma.module";
import { MessagesModule } from "../messages/messages.module";
import { CampaignsService } from "./campaigns.service";
import { CampaignSendService } from "./campaign-send.service";
import { CampaignsController } from "./campaigns.controller";
import { InternalCampaignsController } from "./internal-campaigns.controller";
import { campaignQueueProviders } from "./campaign-queue.provider";

/**
 * The Campaigns module (Phase 4). Imports `MessagesModule` to reuse the Phase 1
 * `MessageService` send path. Exports `CampaignsService` + `CampaignSendService`
 * so the webhook module can wire the FR-05.10 reflection callback.
 */
@Module({
  imports: [PrismaModule, MessagesModule],
  controllers: [CampaignsController, InternalCampaignsController],
  providers: [
    CampaignsService,
    CampaignSendService,
    ...campaignQueueProviders,
  ],
  exports: [CampaignsService, CampaignSendService],
})
export class CampaignsModule {}
```

> Confirm `MessagesModule` exports `MessageService` — if it does not, add it to
> that module's `exports` array (a one-line, additive change; note it in the
> commit). `PrismaModule` is global per the Phase 0 convention; importing it is
> harmless but match what `TemplatesModule` does.

- [ ] **Step 6: Register the module in `apps/api/src/app.module.ts`:**

Add `CampaignsModule` to the `imports` array of `AppModule`.

- [ ] **Step 7: Wire the webhook reflection dep (FR-05.10)**

In `apps/api/src/webhooks/webhook-processor.service.ts`, inject
`CampaignsService` and `PrismaService` (already injected) and pass a
`reflectCampaignStatus` callback into `processEnvelope`:

```typescript
      reflectCampaignStatus: async ({ campaignId, messageId, status }) => {
        // Map the message status onto the recipient row, then refresh stats.
        await this.prisma.campaignRecipient.updateMany({
          where: { campaignId, messageId },
          data: { status },
        });
        await this.campaigns.recomputeStats(campaignId);
        await this.campaigns.checkCompletion(campaignId);
      },
```

Add `CampaignsService` to the `WebhookProcessorService` constructor and import
`CampaignsModule` into `WebhooksModule` (so `CampaignsService` is injectable
there). Update `webhook-processor.service.test.ts`: add a mocked
`CampaignsService` (`recomputeStats`/`checkCompletion` as `vi.fn()`) to the
constructor and one test asserting that when `processEnvelope` invokes the
`reflectCampaignStatus` dep, `campaignRecipient.updateMany` + `recomputeStats`
+ `checkCompletion` are called.

> Note `RecipientStatus` has no `sent`-vs-`queued` mismatch with `MsgStatus`:
> the webhook only ever yields `sent|delivered|read|failed`, all valid
> `RecipientStatus` values, so the `status` passes straight through.

- [ ] **Step 8: Full API verification**

Run: `pnpm --filter @whatapp/api lint && pnpm --filter @whatapp/api typecheck && pnpm --filter @whatapp/api test && pnpm --filter @whatapp/api build`
Expected: PASS. (No live DB/Redis/Meta is touched — all mocked.)

- [ ] **Step 9: Commit**

```bash
git add apps/api/src/campaigns apps/api/src/app.module.ts \
  apps/api/src/webhooks/webhook-processor.service.ts \
  apps/api/src/webhooks/webhook-processor.service.test.ts \
  apps/api/src/webhooks/webhooks.module.ts \
  apps/api/src/messages/messages.module.ts
git commit -m "feat: add campaigns controllers, module, webhook reflection wiring"
```

---

## Task 13: Register the worker BullMQ workers

**Files:**
- Modify: `apps/worker/src/main.ts`

Wires the two new processors into running BullMQ `Worker`s. Follow the existing
`Worker` registrations in `main.ts` for `webhook-processing` / `media-download`
/ `flow-dispatch`.

- [ ] **Step 1: Add a `campaign-send` Worker** in `main.ts`:

```typescript
import { Worker } from "bullmq";
import {
  QUEUE_CAMPAIGN_SEND,
  QUEUE_CAMPAIGN_START,
  CAMPAIGN_SEND_LIMITER,
  createRedisConnection,
} from "./queues";
import { processCampaignSendJob } from "./processors/campaign-send.processor";
import { processCampaignStartJob } from "./processors/campaign-start.processor";
import { createInternalApiClient } from "./lib/internal-api";

// Built once from env — INTERNAL_API_URL + CALLBACK_SECRET (both required,
// fail fast at startup if missing, per the Phase 0 config convention).
const internalApi = createInternalApiClient({
  baseUrl: requireEnv("INTERNAL_API_URL"),
  callbackSecret: requireEnv("CALLBACK_SECRET"),
});

const campaignSendWorker = new Worker(
  QUEUE_CAMPAIGN_SEND,
  async (job) =>
    processCampaignSendJob(job.data, {
      sendRecipient: (id) => internalApi.sendRecipient(id),
      log: (m) => console.log(m),
    }),
  { connection: createRedisConnection(), limiter: CAMPAIGN_SEND_LIMITER },
);

const campaignStartWorker = new Worker(
  QUEUE_CAMPAIGN_START,
  async (job) =>
    processCampaignStartJob(job.data, {
      startCampaign: (id) => internalApi.startCampaign(id),
      log: (m) => console.log(m),
    }),
  { connection: createRedisConnection() },
);
```

Use the file's existing `requireEnv` helper (or env-access pattern); add the
two new workers to whatever graceful-shutdown / `.close()` list `main.ts` keeps.
Add `INTERNAL_API_URL` to `.env.example` with an empty value (no secret — it is
a URL).

- [ ] **Step 2: Typecheck + build the worker**

Run: `pnpm --filter @whatapp/worker lint && pnpm --filter @whatapp/worker typecheck && pnpm --filter @whatapp/worker test && pnpm --filter @whatapp/worker build`
Expected: PASS. Actually starting the workers against live Redis is **"live
verification deferred"**.

- [ ] **Step 3: Commit**

```bash
git add apps/worker/src/main.ts .env.example
git commit -m "feat: register campaign-send and campaign-start BullMQ workers"
```

---

## Task 14: Web — campaigns API client

**Files:**
- Create: `apps/web/src/lib/campaigns-api.ts`

Typed fetch functions over the existing `apps/web/src/lib/api.ts` wrapper.
Follow `apps/web/src/lib/templates-api.ts`.

- [ ] **Step 1: Implement `apps/web/src/lib/campaigns-api.ts`:**

```typescript
import { api } from "./api";

/** Campaign status as returned by the API. */
export type CampaignStatus =
  | "draft"
  | "scheduled"
  | "sending"
  | "paused"
  | "completed"
  | "cancelled"
  | "failed";

/** Recipient status as returned by the API. */
export type RecipientStatus =
  | "pending"
  | "sent"
  | "delivered"
  | "read"
  | "failed"
  | "skipped";

/** Keyed campaign stats (FR-05.15). */
export interface CampaignStats {
  total: number;
  pending: number;
  sent: number;
  delivered: number;
  read: number;
  failed: number;
  skipped: number;
}

/** One variable source — fixed string or contact field path (FR-05.2). */
export type VariableSource =
  | { kind: "fixed"; value: string }
  | { kind: "field"; path: string };

/** position -> source. */
export type VariableMapping = Record<string, VariableSource>;

/** A campaign summary/detail row. */
export interface Campaign {
  id: string;
  name: string;
  templateId: string;
  segmentId: string | null;
  variableMapping: VariableMapping;
  status: CampaignStatus;
  scheduledAt: string | null;
  startedAt: string | null;
  completedAt: string | null;
  stats: CampaignStats;
  createdAt: string;
  template?: { id: string; name: string; status: string; variables: string[] };
  segment?: { id: string; name: string } | null;
}

/** One recipient row with its contact. */
export interface CampaignRecipient {
  id: string;
  waId: string;
  status: RecipientStatus;
  skipReason: string | null;
  error: unknown;
  messageId: string | null;
  sentAt: string | null;
  contact?: { id: string; displayName: string | null; waId: string };
}

/** Audience resolution result (FR-05.4). */
export interface AudienceResult {
  matched: number;
  sendable: number;
}

export const campaignsApi = {
  list: (status?: CampaignStatus) =>
    api.get<Campaign[]>(
      `/api/campaigns${status ? `?status=${status}` : ""}`,
    ),
  get: (id: string) => api.get<Campaign>(`/api/campaigns/${id}`),
  create: (body: {
    name: string;
    templateId: string;
    segmentId: string;
    variableMapping: VariableMapping;
    scheduledAt?: string;
  }) => api.post<Campaign>("/api/campaigns", body),
  update: (id: string, body: Partial<{ name: string; variableMapping: VariableMapping; scheduledAt: string | null }>) =>
    api.patch<Campaign>(`/api/campaigns/${id}`, body),
  audience: (id: string) =>
    api.get<AudienceResult>(`/api/campaigns/${id}/audience`),
  send: (id: string, scheduledAt?: string) =>
    api.post<{ ok: true }>(
      `/api/campaigns/${id}/send`,
      scheduledAt ? { scheduledAt } : {},
    ),
  pause: (id: string) => api.post<Campaign>(`/api/campaigns/${id}/pause`, {}),
  resume: (id: string) => api.post<Campaign>(`/api/campaigns/${id}/resume`, {}),
  cancel: (id: string) => api.post<Campaign>(`/api/campaigns/${id}/cancel`, {}),
  retryFailed: (id: string) =>
    api.post<{ requeued: number }>(`/api/campaigns/${id}/retry-failed`, {}),
  recipients: (id: string, status?: RecipientStatus) =>
    api.get<{ items: CampaignRecipient[]; total: number }>(
      `/api/campaigns/${id}/recipients${status ? `?status=${status}` : ""}`,
    ),
};
```

> Match the `api` export shape to the real `apps/web/src/lib/api.ts` — if it
> exports named functions (`apiGet`/`apiPost`) rather than an `api` object,
> adapt the calls accordingly. The function *names* and return types above are
> the contract for Tasks 15–17.

- [ ] **Step 2: Typecheck**

Run: `pnpm --filter @whatapp/web typecheck`
Expected: PASS.

- [ ] **Step 3: Commit**

```bash
git add apps/web/src/lib/campaigns-api.ts
git commit -m "feat: add web campaigns API client"
```

---

## Task 15: Web — campaign UI components

**Files:**
- Create: `apps/web/src/components/campaigns/CampaignStatusBadge.tsx` + `.test.tsx`
- Create: `apps/web/src/components/campaigns/CampaignProgress.tsx` + `.test.tsx`
- Create: `apps/web/src/components/campaigns/RecipientTable.tsx` + `.test.tsx`
- Create: `apps/web/src/components/campaigns/VariableMappingEditor.tsx` + `.test.tsx`

Four presentational components, each unit-tested with RTL. Follow the existing
`apps/web/src/components/templates/StatusBadge.tsx` pattern.

- [ ] **Step 1: `CampaignStatusBadge`** — test then implement. A `<span>` whose
  text is the status and whose `data-status` attribute is the status, with a
  colour class per status.

```tsx
// CampaignStatusBadge.test.tsx
import { render, screen } from "@testing-library/react";
import { describe, it, expect } from "vitest";
import { CampaignStatusBadge } from "./CampaignStatusBadge";

describe("CampaignStatusBadge", () => {
  it("renders the status text", () => {
    render(<CampaignStatusBadge status="sending" />);
    expect(screen.getByText("sending")).toBeInTheDocument();
  });
  it("exposes the status via data-status", () => {
    render(<CampaignStatusBadge status="paused" />);
    expect(screen.getByText("paused")).toHaveAttribute(
      "data-status",
      "paused",
    );
  });
});
```

```tsx
// CampaignStatusBadge.tsx
import type { CampaignStatus } from "../../lib/campaigns-api";

const COLOR: Record<CampaignStatus, string> = {
  draft: "badge-gray",
  scheduled: "badge-blue",
  sending: "badge-amber",
  paused: "badge-orange",
  completed: "badge-green",
  cancelled: "badge-gray",
  failed: "badge-red",
};

/** Coloured status pill for a campaign. */
export function CampaignStatusBadge({ status }: { status: CampaignStatus }) {
  return (
    <span className={`badge ${COLOR[status]}`} data-status={status}>
      {status}
    </span>
  );
}
```

- [ ] **Step 2: `CampaignProgress`** — test then implement. Takes a
  `CampaignStats`; renders a progress bar (percentage =
  `(total - pending) / total`) and one tile per non-zero stat.

```tsx
// CampaignProgress.test.tsx
import { render, screen } from "@testing-library/react";
import { describe, it, expect } from "vitest";
import { CampaignProgress } from "./CampaignProgress";

const stats = {
  total: 10,
  pending: 2,
  sent: 3,
  delivered: 3,
  read: 1,
  failed: 1,
  skipped: 0,
};

describe("CampaignProgress", () => {
  it("shows the processed/total summary", () => {
    render(<CampaignProgress stats={stats} />);
    expect(screen.getByText(/8 \/ 10/)).toBeInTheDocument();
  });
  it("renders a progressbar at 80%", () => {
    render(<CampaignProgress stats={stats} />);
    expect(screen.getByRole("progressbar")).toHaveAttribute(
      "aria-valuenow",
      "80",
    );
  });
  it("shows the delivered count", () => {
    render(<CampaignProgress stats={stats} />);
    expect(screen.getByTestId("stat-delivered")).toHaveTextContent("3");
  });
  it("handles an empty campaign without dividing by zero", () => {
    render(
      <CampaignProgress
        stats={{ total: 0, pending: 0, sent: 0, delivered: 0, read: 0, failed: 0, skipped: 0 }}
      />,
    );
    expect(screen.getByRole("progressbar")).toHaveAttribute(
      "aria-valuenow",
      "0",
    );
  });
});
```

Implement `CampaignProgress.tsx`: compute
`pct = stats.total === 0 ? 0 : Math.round(((stats.total - stats.pending) / stats.total) * 100)`,
render `<div role="progressbar" aria-valuenow={pct} aria-valuemin={0}
aria-valuemax={100}>`, a `{processed} / {total}` label, and a tile
`<div data-testid={`stat-${key}`}>` for each of
`sent|delivered|read|failed|skipped`.

- [ ] **Step 3: `RecipientTable`** — test then implement. Props: `recipients`
  and an optional `statusFilter` + `onStatusFilterChange`. Renders a `<table>`
  row per recipient (waId, status badge, skip reason / error) and a status
  `<select>` filter. Tests: renders N rows; shows a recipient's `skipReason`;
  calls `onStatusFilterChange` when the select changes.

- [ ] **Step 4: `VariableMappingEditor`** — test then implement. The core
  builder control. Props: `variables: string[]` (template variable positions),
  `value: VariableMapping`, `onChange`, and `sampleContact` (a contact-like
  object for the live preview). For each variable it renders: a `kind` toggle
  (`fixed` / `field`), a text input (fixed) or a path `<select>` populated with
  the contact-field paths + an `attributes.<key>` free-text, and a **live
  sample value** computed by resolving the current source against
  `sampleContact`. Tests: changing a fixed value calls `onChange` with
  `{ kind:"fixed", value }`; selecting a field path calls `onChange` with
  `{ kind:"field", path }`; the live preview shows the resolved sample; a
  field with no sample value shows a "missing" hint.

> The editor must not import server code. Re-implement the tiny resolve-for-
> preview inline (fixed -> value; field -> read the path off `sampleContact`),
> OR import `resolveVariableMapping` from `@whatapp/shared` if the web package
> already depends on `@whatapp/shared` (check `apps/web/package.json`). Prefer
> the shared import when available — it keeps preview and send identical.

- [ ] **Step 5: Run the component tests — expect PASS**

Run: `pnpm --filter @whatapp/web test campaigns`
Expected: PASS — every component test green.

- [ ] **Step 6: Commit**

```bash
git add apps/web/src/components/campaigns
git commit -m "feat: add campaign UI components (badge, progress, recipients, mapping)"
```

---

## Task 16: Web — campaigns list and builder routes

**Files:**
- Create: `apps/web/src/routes/Campaigns.tsx` + `.test.tsx`
- Create: `apps/web/src/routes/CampaignBuilder.tsx` + `.test.tsx`

Follow `apps/web/src/routes/Templates.tsx` / `TemplateEditor.tsx` — React Query
for data, the test harness for mocking `campaignsApi`.

- [ ] **Step 1: `Campaigns.tsx`** — test then implement. Uses
  `useQuery(['campaigns'], () => campaignsApi.list())`. Renders a table: name,
  `CampaignStatusBadge`, template name, audience size (`stats.total`), headline
  stats (`sent`/`delivered`/`failed`), schedule. A "New campaign" link to
  `/campaigns/new`; each row links to `/campaigns/:id`. Test: mock
  `campaignsApi.list` to resolve two campaigns, assert both names render and the
  status badges show.

- [ ] **Step 2: `CampaignBuilder.tsx`** — test then implement. The create/edit
  screen (route `/campaigns/new` and `/campaigns/:id/edit`). It:
  - loads approved templates (`templatesApi.list` filtered to `APPROVED`) and
    segments (`segmentsApi.list`) into two `<select>`s;
  - on template select, reads its `variables` and renders a
    `VariableMappingEditor` for them;
  - has a send-now / schedule radio with a datetime input for the scheduled
    case;
  - shows a **live sendable count**: a `useQuery` that calls
    `campaignsApi.audience(id)` once the campaign exists, or — for a brand-new
    unsaved campaign — calls `segmentsApi.preview(definition)` for the picked
    segment so the count appears before first save (AC-05.2). Display
    `matched` and `sendable` and the exclusion delta (`matched - sendable`
    "excluded: blocked / opted-out").
  - On submit: `campaignsApi.create` (or `update`); then if "send now" was
    chosen, `campaignsApi.send(id)`; navigate to `/campaigns/:id`.

  Tests (mock the APIs): renders the template + segment selects; picking a
  template renders one mapping row per variable; the sendable count from a
  mocked `audience`/`preview` is displayed; submitting calls
  `campaignsApi.create` with the assembled body.

- [ ] **Step 3: Run the route tests — expect PASS**

Run: `pnpm --filter @whatapp/web test Campaigns CampaignBuilder`
Expected: PASS.

- [ ] **Step 4: Commit**

```bash
git add apps/web/src/routes/Campaigns.tsx apps/web/src/routes/Campaigns.test.tsx \
  apps/web/src/routes/CampaignBuilder.tsx apps/web/src/routes/CampaignBuilder.test.tsx
git commit -m "feat: add campaigns list and builder screens"
```

---

## Task 17: Web — campaign detail route and route wiring

**Files:**
- Create: `apps/web/src/routes/CampaignDetail.tsx` + `.test.tsx`
- Modify: `apps/web/src/App.tsx`

- [ ] **Step 1: `CampaignDetail.tsx`** — test then implement. Route
  `/campaigns/:id`. It:
  - `useQuery(['campaign', id], () => campaignsApi.get(id))` and
    `useQuery(['campaign-recipients', id, statusFilter], ...)`, both with a
    `refetchInterval` (e.g. 5s) so progress updates live while `sending`;
  - renders the campaign name, `CampaignStatusBadge`, `CampaignProgress` from
    `campaign.stats`, and a `RecipientTable` with the status filter;
  - renders control buttons gated by status: **Pause** when `sending`,
    **Resume** when `paused`, **Cancel** when `sending`/`paused`/`scheduled`,
    **Retry failed** when `stats.failed > 0`. Each button calls the matching
    `campaignsApi` method via a `useMutation` that invalidates the campaign
    query.
  - Tests (mock the APIs): renders the progress bar from stats; a `sending`
    campaign shows Pause + Cancel and not Resume; clicking Pause calls
    `campaignsApi.pause`; a campaign with `stats.failed > 0` shows "Retry
    failed" and clicking it calls `campaignsApi.retryFailed`; the recipient
    table renders the mocked recipients.

- [ ] **Step 2: Wire the routes in `App.tsx`** — replace the placeholder line
  `<Route path="/campaigns" element={<Placeholder name="Campaigns" />} />`
  with:

```tsx
<Route path="/campaigns" element={<Campaigns />} />
<Route path="/campaigns/new" element={<CampaignBuilder />} />
<Route path="/campaigns/:id/edit" element={<CampaignBuilder />} />
<Route path="/campaigns/:id" element={<CampaignDetail />} />
```

Add the three imports at the top of `App.tsx`.

- [ ] **Step 3: Full web verification**

Run: `pnpm --filter @whatapp/web lint && pnpm --filter @whatapp/web typecheck && pnpm --filter @whatapp/web test && pnpm --filter @whatapp/web build`
Expected: PASS. Live click-through in a browser is **"live verification
deferred"** (no live API).

- [ ] **Step 4: Commit**

```bash
git add apps/web/src/routes/CampaignDetail.tsx \
  apps/web/src/routes/CampaignDetail.test.tsx apps/web/src/App.tsx
git commit -m "feat: add campaign detail screen and wire campaign routes"
```

---

## Task 18: Phase verification & roadmap tick

**Files:**
- Modify: `plans/ROADMAP.md`

- [ ] **Step 1: Run the full monorepo gate from the repo root**

Run: `pnpm lint && pnpm typecheck && pnpm test && pnpm build`
Expected: all four PASS. If any fails, fix it under the relevant task's
conventions and re-run before continuing.

- [ ] **Step 2: Walk every acceptance criterion against the suite**

For each AC-05.1..8, point at the test that proves it (see the self-review
table below) and confirm that test is green in the Step 1 run. Anything that
needs a live Meta API, live Redis, or live Postgres is recorded as **"live
verification deferred"** — not a blocker (Docker is broken on this machine).

- [ ] **Step 3: Confirm no secret was committed**

Run: `git log -p -n 20 -- apps packages .env.example` and scan for any token,
key, or password literal. `.env.example` must hold only empty values. Expected:
clean.

- [ ] **Step 4: Tick Phase 4 in `plans/ROADMAP.md`**

Change `- [ ] Phase 4 — Campaigns` to `- [x] Phase 4 — Campaigns`.

- [ ] **Step 5: Commit**

```bash
git add plans/ROADMAP.md
git commit -m "chore: mark Phase 4 (Campaigns) complete"
```

---

## Self-review — FR / AC coverage map

Every functional requirement and acceptance criterion from
`specs/05-campaigns.md`, mapped to the task(s) that implement it.

| Requirement | Covered by |
|---|---|
| FR-05.1 `POST /api/campaigns` create draft, APPROVED template | Task 9 (DTO), Task 10 (`create` + APPROVED check), Task 12 (controller route + test) |
| FR-05.2 per-recipient variable mapping; skip on missing | Task 2 (`CampaignVariableMapping` schema), Task 3 (`resolveVariableMapping`), Task 11 (`sendOneRecipient` skip path) |
| FR-05.3 `PATCH` edit only while draft/scheduled | Task 9 (DTO), Task 10 (`update` guard + test) |
| FR-05.4 `GET /:id/audience` matched + sendable, consent/block exclusion | Task 3 (`buildAudienceWhere`), Task 11 (`resolveAudience`), Task 12 (controller route) |
| FR-05.5 audience resolved at send time (dynamic) | Task 11 (`start`/`resolveAudience` recompile the segment) |
| FR-05.6 `POST /:id/send` immediate or scheduled; `sending`+`startedAt`; one recipient row per sendable | Task 11 (`schedule` + `start`), Task 12 (controller `send`), Task 8 (`campaign-start` processor) |
| FR-05.7 one rate-limited BullMQ job per recipient | Task 5 (`CAMPAIGN_SEND_LIMITER`), Task 11 (`enqueueRecipients`), Task 13 (`Worker` with limiter) |
| FR-05.8 recipient job resolves mapping, sends, updates row; 130429 backoff; permanent error fails one | Task 5 (`CAMPAIGN_SEND_JOB_OPTS`), Task 7 (processor retry decision), Task 11 (`sendOneRecipient` outcomes) |
| FR-05.9 idempotent recipient jobs (no double send) | Task 11 (`sendOneRecipient` status guard), Task 1 (`@@unique([campaignId,contactId])` confirmed) |
| FR-05.10 delivery webhooks reflect onto recipient + stats | Task 4 (`reflectCampaignStatus` in `handleStatus`), Task 12 (webhook-processor wiring) |
| FR-05.11 pause / resume / cancel | Task 10 (`pause`/`resume`/`cancel`), Task 12 (controller routes) |
| FR-05.12 `retry-failed` re-enqueues only failures | Task 10 (`retryFailed`), Task 12 (controller calls `enqueueRecipients`) |
| FR-05.13 `completed` when all recipients terminal; `completedAt` | Task 3 (`isCampaignComplete`), Task 10 (`checkCompletion`) |
| FR-05.14 list / detail / recipients endpoints, recipient status filter | Task 10 (`list`/`get`/`recipients`), Task 12 (controller routes), Tasks 16–17 (UI) |
| FR-05.15 stats kept current | Task 3 (`aggregateCampaignStats`), Task 10 (`recomputeStats`), invoked from Task 11 + Task 4 |
| AC-05.1 campaign only uses APPROVED template | Task 10 `create` test "rejects a non-APPROVED template" |
| AC-05.2 audience excludes blocked + (marketing) opted-out; builder shows sendable count | Task 3 `buildAudienceWhere` tests, Task 11 `resolveAudience` test, Task 16 (builder live count) |
| AC-05.3 sending creates one row per sendable + paces via rate-limited queue | Task 11 `start` test "creates one pending recipient row per sendable contact", Task 5 limiter, Task 13 worker |
| AC-05.4 recipient missing a variable is skipped with a reason | Task 3 resolver skip tests, Task 11 `sendOneRecipient` test "skips a recipient missing a required variable" |
| AC-05.5 re-processing an already-sent recipient does not send twice | Task 11 `sendOneRecipient` test "is idempotent — an already-sent recipient is not re-sent", Task 7 `already_done` case |
| AC-05.6 delivery webhooks update recipient status + stats | Task 4 envelope-processor reflection tests, Task 12 webhook-processor test |
| AC-05.7 pause halts sends; resume continues; retry re-sends only failures | Task 10 control tests, Task 11 `sendOneRecipient` paused-campaign test, Task 10 `retryFailed` test |
| AC-05.8 campaign becomes completed only when all recipients terminal | Task 3 `isCampaignComplete` tests, Task 10 `checkCompletion` tests |

**No unmapped FR or AC.** Every FR-05.1..15 and AC-05.1..8 has at least one
covering task with a concrete test or verification step.