chore: merge 09-02 worktree commits (Wave 2)

2026-05-10 07:36:15 -04:00 · 2026-05-10 07:36:15 -04:00 · 1bce661648
commit 1bce661648
parent 486f27eb77 17a189e564
7 changed files with 810 additions and 0 deletions
--- a/.planning/phases/09-user-profile-preferences-new/09-02-SUMMARY.md
+++ b/.planning/phases/09-user-profile-preferences-new/09-02-SUMMARY.md
@ -0,0 +1,144 @@
+---
+phase: 09-user-profile-preferences-new
+plan: "02"
+subsystem: api-routes
+tags: [api, auth, channels, theme, notifications, phase-9]
+dependency_graph:
+  requires:
+    - "user.theme column (Plan 01 — migration 084)"
+    - "notification_channels.owner_user_id + partial unique index (Plan 01 — migration 085)"
+    - "notify_event_keys + user_event_subscriptions tables (Plan 01 — migration 086)"
+    - "NotifyEventKey, UserEventSubscription, NotificationChannel types (Plan 01)"
+  provides:
+    - "GET + PUT /api/me/theme — { theme, source } and validated write"
+    - "GET /api/me/channels — user's personal channels, camelCase"
+    - "PUT /api/me/channels/[type] — UPSERT + best-effort test send"
+    - "DELETE /api/me/channels/[type] — remove personal channel"
+    - "POST /api/me/channels/[type]/test — re-send test to existing channel"
+    - "GET + PUT /api/me/notification-subscriptions — full matrix + single row UPSERT"
+    - "lib/services/personal-channels.ts — shared validation + test-send helpers"
+  affects:
+    - "Plans 03/04/05/06 — consume these endpoints and the personal-channels service"
+tech_stack:
+  added: []
+  patterns:
+    - "WITH-CTE UPSERT pattern scoped to (owner_user_id, channel_type) with 23505 → 409 fallback"
+    - "Default-enabled matrix (row absence = true) for subscription opt-out model"
+    - "Best-effort test send on every channel save — never blocks success"
+    - "SSRF mitigation: Teams URL hostname allowlist via URL() parsing"
+    - "Snake_case column names in DB, camelCase in API responses (manual transform)"
+key_files:
+  created:
+    - app/api/me/theme/route.ts
+    - app/api/me/channels/route.ts
+    - app/api/me/channels/[type]/route.ts
+    - app/api/me/channels/[type]/test/route.ts
+    - app/api/me/notification-subscriptions/route.ts
+    - lib/services/personal-channels.ts
+  modified: []
+decisions:
+  - "theme source='default' when stored value is 'system' (matches /api/me/timezone semantic)"
+  - "WITH-CTE UPSERT (not INSERT ... ON CONFLICT) because the partial unique index only fires on INSERT — the CTE update arm is the preferred path, with INSERT as fallback when no existing row"
+  - "sendChannelTest does NOT log channel.config to avoid leaking webhook URLs / auth tokens in server logs"
+  - "Custom ntfy topic 400 response body carries both error and message keys so Plan 05 UI can render inline error below the input"
+metrics:
+  duration_minutes: 15
+  completed_date: "2026-05-10"
+  tasks_completed: 3
+  files_created: 6
+  files_modified: 0
+---
+
+# Phase 9 Plan 02: Per-User API Surface Summary
+
+One-liner: Six new files deliver the complete /api/me/* API surface for theme, personal channels (Teams + ntfy), and notification subscription matrix — all gated by requireAuth(), writing only session.user.id, with Teams SSRF prevention and default-enabled subscription fallback.
+
+## Tasks Completed
+
+| Task | Name | Commit | Files |
+|------|------|--------|-------|
+| 1 | GET + PUT /api/me/theme | dc5dc91 | app/api/me/theme/route.ts |
+| 2 | Personal channels service + /api/me/channels routes | c35b968 | lib/services/personal-channels.ts, app/api/me/channels/route.ts, app/api/me/channels/[type]/route.ts, app/api/me/channels/[type]/test/route.ts |
+| 3 | GET + PUT /api/me/notification-subscriptions | 55a80a0 | app/api/me/notification-subscriptions/route.ts |
+
+## Route Files
+
+### app/api/me/theme/route.ts
+- **Exports:** `GET`, `PUT`
+- **GET:** `SELECT theme FROM "user" WHERE id = $1` → `{ theme: 'light'|'dark'|'system', source: 'user'|'default' }`. Source is `'user'` when stored value differs from `'system'` (the default).
+- **PUT:** Validates `body.theme` against `ALLOWED_THEMES = new Set(['light', 'dark', 'system'])`. Returns 400 on invalid value. `UPDATE "user" SET theme = $1, updated_at = NOW() WHERE id = $2 RETURNING theme` — uses `updated_at` (snake_case, unquoted, matches migration 012). Does NOT contain `"updatedAt"`.
+- **Auth:** `requireAuth()` on both handlers. Write target: `session!.user.id` only.
+
+### app/api/me/channels/route.ts
+- **Exports:** `GET`
+- **GET:** `SELECT ... FROM notification_channels WHERE owner_user_id = $1 ORDER BY channel_type ASC` → `{ channels: [...camelCase rows] }`. Scoped strictly to calling user.
+
+### app/api/me/channels/[type]/route.ts
+- **Exports:** `PUT`, `DELETE`
+- **PUT teams:** Validates `webhook_url` via `isValidTeamsWebhookUrl`. Name: `"Personal Teams (<email>)"`. Config: `{ webhook_url }`.
+- **PUT ntfy:** Validates custom topic via `isValidNtfyTopic` if supplied; otherwise mints via `mintNtfyTopic()`. Name: `"Personal ntfy (<email>)"`. Config: `{ server_url: 'https://ntfy.sh', topic }`.
+- **PUT UPSERT SQL:** WITH-CTE pattern — UPDATE existing row if present, INSERT if not; params `[$1=owner_user_id, $2=channel_type, $3=name, $4=config]`. On `23505` unique_violation → 409 `{ error: 'Conflict', message: '...' }`.
+- **PUT response:** `{ channel: <camelCase row>, test: <ChannelTestResult> }`.
+- **DELETE SQL:** `DELETE FROM notification_channels WHERE owner_user_id = $1 AND channel_type = $2 RETURNING id` → 404 if rowCount=0, else `{ deleted: true, channelType }`.
+- **Auth:** `requireAuth()` on both. No `userId`/`user_id`/`owner_user_id` accepted from body.
+
+### app/api/me/channels/[type]/test/route.ts
+- **Exports:** `POST`
+- **POST:** `SELECT ... WHERE owner_user_id = $1 AND channel_type = $2` → 404 if missing. Calls `sendChannelTest(channel)` → `{ test: ChannelTestResult }`. Returns 200 even on test failure.
+
+### app/api/me/notification-subscriptions/route.ts
+- **Exports:** `GET`, `PUT`
+- **GET implementation:** Three sequential queries:
+  1. `SELECT key, display_label, description, sort_order FROM notify_event_keys WHERE is_active = true ORDER BY sort_order ASC, key ASC`
+  2. `SELECT channel_type FROM notification_channels WHERE owner_user_id = $1 AND is_active = true`
+  3. `SELECT event_key, channel_type, enabled FROM user_event_subscriptions WHERE user_id = $1`
+  - Builds matrix defaulting to `true` when no row exists (D-15 opt-out model).
+- **GET response shape:** `{ eventKeys: [{key, displayLabel, description, sortOrder}], channelTypes: string[], matrix: { [event_key]: { [channel_type]: boolean } } }`
+- **PUT validation:** `event_key` non-empty string ≤ 128 chars; `channel_type` via `isPersonalChannelType`; `enabled` via `typeof === 'boolean'` (400 on failure).
+- **PUT SQL:** `INSERT INTO user_event_subscriptions ... ON CONFLICT (user_id, event_key, channel_type) DO UPDATE SET enabled = EXCLUDED.enabled, updated_at = NOW() RETURNING ...`
+- **PUT response:** `{ subscription: { eventKey, channelType, enabled } }`
+
+## lib/services/personal-channels.ts — Exported Symbols
+
+| Symbol | Type | Description |
+|--------|------|-------------|
+| `TEST_MESSAGE_BODY` | `const string` | `'Pulse channel verified — you can ignore this message.'` |
+| `isValidTeamsWebhookUrl` | `(input: unknown) => input is string` | https:// + hostname must match *.webhook.office.com or *.logic.azure.com |
+| `isValidNtfyTopic` | `(input: unknown) => input is string` | Regex `^[A-Za-z0-9_-]{6,64}$` |
+| `mintNtfyTopic` | `() => string` | Returns `pulse-XXXXXXXX` (8 hex chars from crypto.randomUUID()) |
+| `sendChannelTest` | `(channel: NotificationChannel) => Promise<ChannelTestResult>` | Best-effort test send to teams or ntfy |
+| `isPersonalChannelType` | `(t: unknown) => t is 'teams' \| 'ntfy'` | Type guard for allowed personal channel types |
+| `PERSONAL_CHANNEL_TYPES` | `ChannelType[]` | `['teams', 'ntfy']` |
+
+## Security Confirmations
+
+- **No route accepts userId/user_id/owner_user_id from request input.** All writes are scoped to `session!.user.id` from `requireAuth()`.
+- **Theme route uses `updated_at` (NOT `"updatedAt"`).** Verified: the file contains `updated_at = NOW()` and does not contain the quoted camelCase identifier.
+- **Teams SSRF prevention:** `isValidTeamsWebhookUrl` uses `new URL(input)` to parse then checks `u.protocol === 'https:'` and hostname against three host patterns. Internal hosts cannot match.
+- **Custom ntfy topic 400 shape:** `{ error: 'Invalid topic', message: 'topic must match ^[A-Za-z0-9_-]{6,64}$' }` — Plan 05 UI renders `message` inline below the custom-topic Input.
+- **sendChannelTest does not log channel.config** — only logs channel id/type on failure paths.
+
+## Deviations from Plan
+
+None — plan executed exactly as written.
+
+## Threat Flags
+
+No new network endpoints, auth paths, file access patterns, or schema changes beyond what was declared in the plan's threat model. All ten threats documented (T-09-02-01 through T-09-02-10) are addressed by the implementation.
+
+## Self-Check: PASSED
+
+Files exist:
+- app/api/me/theme/route.ts: FOUND
+- app/api/me/channels/route.ts: FOUND
+- app/api/me/channels/[type]/route.ts: FOUND
+- app/api/me/channels/[type]/test/route.ts: FOUND
+- app/api/me/notification-subscriptions/route.ts: FOUND
+- lib/services/personal-channels.ts: FOUND
+
+Commits exist:
+- dc5dc91: FOUND (Task 1 — theme route)
+- c35b968: FOUND (Task 2 — channels service + routes)
+- 55a80a0: FOUND (Task 3 — notification-subscriptions route)
+
+TypeScript: `npx tsc --noEmit --pretty` exit 0 — no errors.
--- a/app/api/me/channels/[type]/route.ts
+++ b/app/api/me/channels/[type]/route.ts
@ -0,0 +1,180 @@
+// PUT  /api/me/channels/[type] — upsert a personal channel for the calling user
+// DELETE /api/me/channels/[type] — remove the calling user's channel of that type
+//
+// CHAN-02, CHAN-03, CHAN-04, CHAN-05, CHAN-07 / D-02..D-06
+// Auth: requireAuth(). Write target is always session.user.id.
+// T-09-02-01: no userId/user_id accepted from body or query.
+// T-09-02-06: Teams URLs validated via isValidTeamsWebhookUrl (SSRF mitigation).
+// T-09-02-07: ntfy topics minted server-side; custom topics validated.
+// T-09-02-10: partial unique index closes race window; 23505 → 409.
+
+import { NextRequest, NextResponse } from 'next/server';
+import { requireAuth } from '@/lib/auth-utils';
+import { postgresClient } from '@/lib/services/postgres-client';
+import { NotificationChannel } from '@/lib/types/pipeline';
+import {
+  isPersonalChannelType,
+  isValidTeamsWebhookUrl,
+  isValidNtfyTopic,
+  mintNtfyTopic,
+  sendChannelTest,
+} from '@/lib/services/personal-channels';
+
+type RouteParams = { params: Promise<{ type: string }> };
+
+export async function PUT(request: NextRequest, { params }: RouteParams): Promise<NextResponse> {
+  const { session, error } = await requireAuth();
+  if (error) return error;
+
+  const { type } = await params;
+  if (!isPersonalChannelType(type)) {
+    return NextResponse.json(
+      { error: 'Unsupported channel type', message: `type must be one of: teams, ntfy` },
+      { status: 400 },
+    );
+  }
+
+  let body: Record<string, unknown>;
+  try {
+    body = await request.json();
+  } catch {
+    return NextResponse.json(
+      { error: 'Invalid JSON', message: 'Request body must be JSON' },
+      { status: 400 },
+    );
+  }
+
+  let name: string;
+  let config: Record<string, unknown>;
+
+  if (type === 'teams') {
+    const webhook_url = body.webhook_url;
+    if (!isValidTeamsWebhookUrl(webhook_url)) {
+      return NextResponse.json(
+        {
+          error: 'Invalid webhook URL',
+          message: 'webhook_url must be https://*.webhook.office.com or https://*.logic.azure.com',
+        },
+        { status: 400 },
+      );
+    }
+    config = { webhook_url };
+    name = `Personal Teams (${session!.user.email})`;
+  } else {
+    // type === 'ntfy'
+    const customTopic = typeof body.topic === 'string' ? body.topic : null;
+    let topic: string;
+    if (customTopic !== null) {
+      if (!isValidNtfyTopic(customTopic)) {
+        return NextResponse.json(
+          {
+            error: 'Invalid topic',
+            message: 'topic must match ^[A-Za-z0-9_-]{6,64}$',
+          },
+          { status: 400 },
+        );
+      }
+      topic = customTopic;
+    } else {
+      topic = mintNtfyTopic();
+    }
+    config = { server_url: 'https://ntfy.sh', topic };
+    name = `Personal ntfy (${session!.user.email})`;
+  }
+
+  try {
+    // UPSERT scoped to calling user + channel type.
+    // The partial unique index notification_channels_owner_user_id_channel_type_uniq
+    // (Plan 01) closes the race window — concurrent INSERT raises 23505 → 409.
+    const result = await postgresClient.query<NotificationChannel>(
+      `WITH existing AS (
+        SELECT id FROM notification_channels
+        WHERE owner_user_id = $1 AND channel_type = $2
+        LIMIT 1
+      ),
+      updated AS (
+        UPDATE notification_channels
+           SET name = $3, config = $4, is_active = true, updated_at = NOW()
+         WHERE id = (SELECT id FROM existing)
+         RETURNING *
+      ),
+      inserted AS (
+        INSERT INTO notification_channels (name, channel_type, config, is_active, owner_user_id)
+        SELECT $3, $2, $4, true, $1
+        WHERE NOT EXISTS (SELECT 1 FROM existing)
+        RETURNING *
+      )
+      SELECT * FROM updated UNION ALL SELECT * FROM inserted`,
+      [session!.user.id, type, name, JSON.stringify(config)],
+    );
+
+    const row = result.rows[0];
+    // Best-effort test send — never blocks save success (D-06).
+    const testResult = await sendChannelTest(row);
+
+    return NextResponse.json({
+      channel: {
+        id: row.id,
+        name: row.name,
+        channelType: row.channel_type,
+        config: row.config,
+        isActive: row.is_active,
+        ownerUserId: row.owner_user_id,
+        createdAt: row.created_at,
+        updatedAt: row.updated_at,
+      },
+      test: testResult,
+    });
+  } catch (e: unknown) {
+    // Partial unique index race: concurrent PUT already inserted the row.
+    if (
+      e &&
+      typeof e === 'object' &&
+      'code' in e &&
+      (e as { code: string }).code === '23505'
+    ) {
+      return NextResponse.json(
+        { error: 'Conflict', message: 'Channel already exists for this user; retry the save' },
+        { status: 409 },
+      );
+    }
+    console.error('PUT /api/me/channels/[type] failed:', e);
+    return NextResponse.json(
+      { error: 'Failed to save channel', message: e instanceof Error ? e.message : 'unknown' },
+      { status: 500 },
+    );
+  }
+}
+
+export async function DELETE(_request: NextRequest, { params }: RouteParams): Promise<NextResponse> {
+  const { session, error } = await requireAuth();
+  if (error) return error;
+
+  const { type } = await params;
+  if (!isPersonalChannelType(type)) {
+    return NextResponse.json(
+      { error: 'Unsupported channel type', message: `type must be one of: teams, ntfy` },
+      { status: 400 },
+    );
+  }
+
+  try {
+    const result = await postgresClient.query<{ id: number }>(
+      'DELETE FROM notification_channels WHERE owner_user_id = $1 AND channel_type = $2 RETURNING id',
+      [session!.user.id, type],
+    );
+    if (result.rowCount === 0) {
+      return NextResponse.json(
+        { error: 'Not found', message: `No ${type} channel found for this user` },
+        { status: 404 },
+      );
+    }
+    return NextResponse.json({ deleted: true, channelType: type });
+  } catch (e) {
+    console.error('DELETE /api/me/channels/[type] failed:', e);
+    return NextResponse.json(
+      { error: 'Failed to delete channel', message: e instanceof Error ? e.message : 'unknown' },
+      { status: 500 },
+    );
+  }
+}
--- a/app/api/me/channels/[type]/test/route.ts
+++ b/app/api/me/channels/[type]/test/route.ts
@ -0,0 +1,55 @@
+// POST /api/me/channels/[type]/test
+// Re-sends a test message to the calling user's existing channel of the given type.
+// Returns { test: ChannelTestResult } — 200 even on test failure (mirrors save semantics).
+//
+// CHAN-07 / D-06. Auth: requireAuth(). Read target is always session.user.id.
+// T-09-02-04: cannot trigger test for another user's channel.
+
+import { NextRequest, NextResponse } from 'next/server';
+import { requireAuth } from '@/lib/auth-utils';
+import { postgresClient } from '@/lib/services/postgres-client';
+import { NotificationChannel } from '@/lib/types/pipeline';
+import { isPersonalChannelType, sendChannelTest } from '@/lib/services/personal-channels';
+
+type RouteParams = { params: Promise<{ type: string }> };
+
+export async function POST(_req: NextRequest, { params }: RouteParams): Promise<NextResponse> {
+  const { session, error } = await requireAuth();
+  if (error) return error;
+
+  const { type } = await params;
+  if (!isPersonalChannelType(type)) {
+    return NextResponse.json(
+      { error: 'Unsupported channel type', message: `type must be one of: teams, ntfy` },
+      { status: 400 },
+    );
+  }
+
+  try {
+    const result = await postgresClient.query<NotificationChannel>(
+      `SELECT id, name, channel_type, config, is_active, owner_user_id, created_at, updated_at
+       FROM notification_channels
+       WHERE owner_user_id = $1 AND channel_type = $2`,
+      [session!.user.id, type],
+    );
+
+    if (result.rows.length === 0) {
+      return NextResponse.json(
+        { error: 'Not found', message: `No ${type} channel configured for this user` },
+        { status: 404 },
+      );
+    }
+
+    const channel = result.rows[0];
+    const testResult = await sendChannelTest(channel);
+
+    // Return 200 even on test failure — the test result is informational.
+    return NextResponse.json({ test: testResult });
+  } catch (e) {
+    console.error('POST /api/me/channels/[type]/test failed:', e);
+    return NextResponse.json(
+      { error: 'Failed to send test', message: e instanceof Error ? e.message : 'unknown' },
+      { status: 500 },
+    );
+  }
+}
--- a/app/api/me/channels/route.ts
+++ b/app/api/me/channels/route.ts
@ -0,0 +1,42 @@
+// GET /api/me/channels
+// Returns the calling user's personal channels (Teams + ntfy), camelCase.
+// Auth: requireAuth(). Reads only owner_user_id = session.user.id.
+// CHAN-02 / T-09-02-04 — users cannot read other users' channels.
+
+import { NextResponse } from 'next/server';
+import { requireAuth } from '@/lib/auth-utils';
+import { postgresClient } from '@/lib/services/postgres-client';
+import { NotificationChannel } from '@/lib/types/pipeline';
+
+export async function GET(): Promise<NextResponse> {
+  const { session, error } = await requireAuth();
+  if (error) return error;
+
+  try {
+    const result = await postgresClient.query<NotificationChannel>(
+      `SELECT id, name, channel_type, config, is_active, owner_user_id, created_at, updated_at
+       FROM notification_channels
+       WHERE owner_user_id = $1
+       ORDER BY channel_type ASC`,
+      [session!.user.id],
+    );
+    // camelCase transform — snake_case DB → camelCase API
+    const channels = result.rows.map((r) => ({
+      id: r.id,
+      name: r.name,
+      channelType: r.channel_type,
+      config: r.config,
+      isActive: r.is_active,
+      ownerUserId: r.owner_user_id,
+      createdAt: r.created_at,
+      updatedAt: r.updated_at,
+    }));
+    return NextResponse.json({ channels });
+  } catch (e) {
+    console.error('GET /api/me/channels failed:', e);
+    return NextResponse.json(
+      { error: 'Failed to read channels', message: e instanceof Error ? e.message : 'unknown' },
+      { status: 500 },
+    );
+  }
+}
--- a/app/api/me/notification-subscriptions/route.ts
+++ b/app/api/me/notification-subscriptions/route.ts
@ -0,0 +1,159 @@
+// GET  /api/me/notification-subscriptions -> { eventKeys, channelTypes, matrix }
+// PUT  /api/me/notification-subscriptions -> body { event_key, channel_type, enabled } -> { subscription }
+//
+// SUB-04. Authentication: requireAuth(). Write target is always session.user.id.
+// T-09-02-01: no user_id accepted from body or query.
+// T-09-02-03: event_key length-bounded; channel_type via isPersonalChannelType;
+//             enabled typeof === 'boolean'.
+//
+// GET matrix shape: { event_key → { channel_type → boolean } }
+// Default: true when no row exists in user_event_subscriptions (D-15 opt-out model).
+
+import { NextRequest, NextResponse } from 'next/server';
+import { requireAuth } from '@/lib/auth-utils';
+import { postgresClient } from '@/lib/services/postgres-client';
+import { NotifyEventKey, UserEventSubscription } from '@/lib/types/pipeline';
+import { isPersonalChannelType } from '@/lib/services/personal-channels';
+
+export async function GET(): Promise<NextResponse> {
+  const { session, error } = await requireAuth();
+  if (error) return error;
+
+  try {
+    // Query A: all active event keys ordered by sort_order
+    const eventKeysResult = await postgresClient.query<NotifyEventKey>(
+      `SELECT key, display_label, description, sort_order
+       FROM notify_event_keys
+       WHERE is_active = true
+       ORDER BY sort_order ASC, key ASC`,
+      [],
+    );
+
+    // Query B: channel types configured for this user
+    const channelTypesResult = await postgresClient.query<{ channel_type: string }>(
+      `SELECT channel_type
+       FROM notification_channels
+       WHERE owner_user_id = $1 AND is_active = true
+       ORDER BY channel_type ASC`,
+      [session!.user.id],
+    );
+
+    // Query C: stored subscription rows for this user
+    const subscriptionsResult = await postgresClient.query<UserEventSubscription>(
+      `SELECT event_key, channel_type, enabled
+       FROM user_event_subscriptions
+       WHERE user_id = $1`,
+      [session!.user.id],
+    );
+
+    const channelTypes = channelTypesResult.rows.map((r) => r.channel_type);
+
+    // Build stored subscription lookup for O(1) access
+    const storedMap = new Map<string, boolean>();
+    for (const row of subscriptionsResult.rows) {
+      storedMap.set(`${row.event_key}:${row.channel_type}`, row.enabled);
+    }
+
+    // Build matrix: event_key → channel_type → boolean.
+    // Default is true (D-15 / T-09-02-03: row absence means enabled).
+    const matrix: Record<string, Record<string, boolean>> = {};
+    for (const ek of eventKeysResult.rows) {
+      matrix[ek.key] = {};
+      for (const ct of channelTypes) {
+        const stored = storedMap.get(`${ek.key}:${ct}`);
+        matrix[ek.key][ct] = stored !== undefined ? stored : true;
+      }
+    }
+
+    // camelCase transform for event keys
+    const eventKeys = eventKeysResult.rows.map((r) => ({
+      key: r.key,
+      displayLabel: r.display_label,
+      description: r.description,
+      sortOrder: r.sort_order,
+    }));
+
+    return NextResponse.json({ eventKeys, channelTypes, matrix });
+  } catch (e) {
+    console.error('GET /api/me/notification-subscriptions failed:', e);
+    return NextResponse.json(
+      { error: 'Failed to read subscriptions', message: e instanceof Error ? e.message : 'unknown' },
+      { status: 500 },
+    );
+  }
+}
+
+export async function PUT(request: NextRequest): Promise<NextResponse> {
+  const { session, error } = await requireAuth();
+  if (error) return error;
+
+  let body: unknown;
+  try {
+    body = await request.json();
+  } catch {
+    return NextResponse.json(
+      { error: 'Invalid JSON', message: 'Request body must be JSON' },
+      { status: 400 },
+    );
+  }
+
+  if (!body || typeof body !== 'object') {
+    return NextResponse.json(
+      { error: 'Invalid body', message: 'Request body must be a JSON object' },
+      { status: 400 },
+    );
+  }
+
+  const { event_key, channel_type, enabled } = body as Record<string, unknown>;
+
+  // Validate event_key: non-empty string ≤ 128 chars
+  if (typeof event_key !== 'string' || event_key.length === 0 || event_key.length > 128) {
+    return NextResponse.json(
+      { error: 'Invalid event_key', message: 'event_key must be a non-empty string of at most 128 characters' },
+      { status: 400 },
+    );
+  }
+
+  // Validate channel_type via isPersonalChannelType allowlist
+  if (!isPersonalChannelType(channel_type)) {
+    return NextResponse.json(
+      { error: 'Invalid channel_type', message: 'channel_type must be one of: teams, ntfy' },
+      { status: 400 },
+    );
+  }
+
+  // Validate enabled: must be a boolean
+  if (typeof enabled !== 'boolean') {
+    return NextResponse.json(
+      { error: 'Invalid enabled', message: 'enabled must be a boolean' },
+      { status: 400 },
+    );
+  }
+
+  try {
+    // Composite PK UPSERT — write target is always session.user.id (T-09-02-01).
+    const result = await postgresClient.query<UserEventSubscription>(
+      `INSERT INTO user_event_subscriptions (user_id, event_key, channel_type, enabled, updated_at)
+       VALUES ($1, $2, $3, $4, NOW())
+       ON CONFLICT (user_id, event_key, channel_type) DO UPDATE
+         SET enabled = EXCLUDED.enabled, updated_at = NOW()
+       RETURNING user_id, event_key, channel_type, enabled, updated_at`,
+      [session!.user.id, event_key, channel_type, enabled],
+    );
+
+    const row = result.rows[0];
+    return NextResponse.json({
+      subscription: {
+        eventKey: row.event_key,
+        channelType: row.channel_type,
+        enabled: row.enabled,
+      },
+    });
+  } catch (e) {
+    console.error('PUT /api/me/notification-subscriptions failed:', e);
+    return NextResponse.json(
+      { error: 'Failed to update subscription', message: e instanceof Error ? e.message : 'unknown' },
+      { status: 500 },
+    );
+  }
+}
--- a/app/api/me/theme/route.ts
+++ b/app/api/me/theme/route.ts
@ -0,0 +1,96 @@
+import { NextRequest, NextResponse } from 'next/server';
+import { requireAuth } from '@/lib/auth-utils';
+import { postgresClient } from '@/lib/services/postgres-client';
+
+// GET  /api/me/theme  -> { theme: 'light' | 'dark' | 'system', source: 'user' | 'default' }
+// PUT  /api/me/theme  -> body { theme: 'light' | 'dark' | 'system' } -> { theme: string }
+//
+// THEME-02. Authentication: requireAuth(). The PUT handler updates ONLY the
+// calling user's row — there is no `userId` query param or body field. The
+// write target is always `session.user.id`.
+//
+// Validation: the input theme must be one of: 'light', 'dark', 'system'.
+// Anything else is rejected with 400 before touching the database.
+//
+// NOTE: The UPDATE statement uses `updated_at` (snake_case, unquoted) — the
+// actual column in migration 012_create_auth_tables.sql. This matches the
+// working precedent in app/api/settings/profile/route.ts (not the timezone route).
+
+const ALLOWED_THEMES = new Set(['light', 'dark', 'system'] as const);
+
+function isValidTheme(t: unknown): t is 'light' | 'dark' | 'system' {
+  return typeof t === 'string' && ALLOWED_THEMES.has(t as 'light' | 'dark' | 'system');
+}
+
+export async function GET(): Promise<NextResponse> {
+  const { session, error } = await requireAuth();
+  if (error) return error;
+
+  try {
+    const result = await postgresClient.query<{ theme: string | null }>(
+      'SELECT theme FROM "user" WHERE id = $1',
+      [session!.user.id],
+    );
+    const stored = result.rows[0]?.theme;
+    // source is 'user' when a non-default value is explicitly stored;
+    // 'default' when value equals fallback ('system') or no row found.
+    const theme = stored ?? 'system';
+    const source: 'user' | 'default' = stored && stored !== 'system' ? 'user' : 'default';
+    return NextResponse.json({ theme, source });
+  } catch (e) {
+    console.error('GET /api/me/theme failed:', e);
+    return NextResponse.json(
+      { error: 'Failed to read theme', message: e instanceof Error ? e.message : 'unknown' },
+      { status: 500 },
+    );
+  }
+}
+
+export async function PUT(request: NextRequest): Promise<NextResponse> {
+  const { session, error } = await requireAuth();
+  if (error) return error;
+
+  let body: unknown;
+  try {
+    body = await request.json();
+  } catch {
+    return NextResponse.json(
+      { error: 'Invalid JSON', message: 'Request body must be JSON' },
+      { status: 400 },
+    );
+  }
+
+  const candidate =
+    body && typeof body === 'object' && 'theme' in body
+      ? (body as { theme: unknown }).theme
+      : undefined;
+
+  if (!isValidTheme(candidate)) {
+    return NextResponse.json(
+      { error: 'Invalid theme', message: 'theme must be one of: light, dark, system' },
+      { status: 400 },
+    );
+  }
+
+  try {
+    // Authoritative write target: session.user.id. NO userId from body.
+    // updated_at is unquoted snake_case — matches migration 012 schema.
+    const result = await postgresClient.query<{ theme: string }>(
+      'UPDATE "user" SET theme = $1, updated_at = NOW() WHERE id = $2 RETURNING theme',
+      [candidate, session!.user.id],
+    );
+    if (result.rowCount === 0) {
+      return NextResponse.json(
+        { error: 'User not found', message: 'No user row matched the session' },
+        { status: 404 },
+      );
+    }
+    return NextResponse.json({ theme: result.rows[0].theme });
+  } catch (e) {
+    console.error('PUT /api/me/theme failed:', e);
+    return NextResponse.json(
+      { error: 'Failed to update theme', message: e instanceof Error ? e.message : 'unknown' },
+      { status: 500 },
+    );
+  }
+}
--- a/lib/services/personal-channels.ts
+++ b/lib/services/personal-channels.ts
@ -0,0 +1,134 @@
+/**
+ * Personal Channels Service — shared validation + test-send helpers.
+ *
+ * Used by:
+ *   - app/api/me/channels/[type]/route.ts  (PUT upsert + DELETE)
+ *   - app/api/me/channels/[type]/test/route.ts  (POST test)
+ *   - lib/services/pipeline-steps/notify.ts  (route_to_user, Plan 03)
+ *   - app/api/admin/channels/route.ts  (admin extension, Plan 06)
+ *
+ * CHAN-02, CHAN-03, CHAN-04, CHAN-05, CHAN-07 / D-02..D-06
+ */
+
+import { randomUUID } from 'crypto';
+import { NotificationChannel, ChannelType } from '@/lib/types/pipeline';
+
+/** Test-send copy used by every personal-channel save and POST :test. */
+export const TEST_MESSAGE_BODY =
+  'Pulse channel verified — you can ignore this message.';
+
+/** Allowed Teams webhook hosts (CHAN-04 / D-05). */
+const TEAMS_HOST_PATTERNS = [
+  /^[a-z0-9-]+\.webhook\.office\.com$/i,
+  /^[a-z0-9-]+\.logic\.azure\.com$/i,
+  /^[a-z0-9-]+\.[a-z]+\.logic\.azure\.com$/i, // regional subdomains, e.g. prod-12.eastus.logic.azure.com
+];
+
+/**
+ * Validate a Teams incoming webhook URL (CHAN-04 / D-05).
+ * Must be https:// and hostname must match *.webhook.office.com or *.logic.azure.com.
+ * Prevents SSRF to internal hosts (T-09-02-06 mitigation).
+ */
+export function isValidTeamsWebhookUrl(input: unknown): input is string {
+  if (typeof input !== 'string' || input.length === 0 || input.length > 2048) return false;
+  try {
+    const u = new URL(input);
+    if (u.protocol !== 'https:') return false;
+    return TEAMS_HOST_PATTERNS.some((re) => re.test(u.hostname));
+  } catch {
+    return false;
+  }
+}
+
+/** ntfy topic format guard (CHAN-03 / D-04). Used when a user supplies a
+ *  custom topic via "Edit advanced". Default flow mints via mintNtfyTopic. */
+const NTFY_TOPIC_RE = /^[A-Za-z0-9_-]{6,64}$/;
+
+export function isValidNtfyTopic(input: unknown): input is string {
+  return typeof input === 'string' && NTFY_TOPIC_RE.test(input);
+}
+
+/**
+ * Mint a Pulse-namespaced ntfy topic (CHAN-03 / D-04).
+ * 8 hex chars from a UUID gives 32 bits of entropy — sufficient collision-
+ * resistance for ntfy public tier at expected user counts (T-09-02-07 mitigation).
+ */
+export function mintNtfyTopic(): string {
+  const id = randomUUID().replace(/-/g, '').slice(0, 8);
+  return `pulse-${id}`;
+}
+
+export type ChannelTestResult =
+  | { ok: true }
+  | { ok: false; status?: number; error: string };
+
+/**
+ * Issue a single best-effort test send to a personal channel. Mirrors the
+ * existing app/api/notification-channels/[id]/test/route.ts flow but with the
+ * Phase 9 test-message body.
+ *
+ * IMPORTANT: Does NOT log channel.config (which may contain webhook URLs or
+ * auth tokens). Only logs channel id and channel_type. (T-09-02-05 mitigation)
+ */
+export async function sendChannelTest(channel: NotificationChannel): Promise<ChannelTestResult> {
+  try {
+    switch (channel.channel_type) {
+      case 'teams': {
+        const url = channel.config.webhook_url;
+        if (!url) return { ok: false, error: 'Teams channel missing webhook_url' };
+        const card = {
+          type: 'message',
+          attachments: [{
+            contentType: 'application/vnd.microsoft.card.adaptive',
+            content: {
+              type: 'AdaptiveCard',
+              $schema: 'http://adaptivecards.io/schemas/adaptive-card.json',
+              version: '1.4',
+              body: [
+                { type: 'TextBlock', text: 'Pulse channel verified', weight: 'bolder', size: 'medium' },
+                { type: 'TextBlock', text: TEST_MESSAGE_BODY, wrap: true },
+              ],
+            },
+          }],
+        };
+        const resp = await fetch(url, {
+          method: 'POST',
+          headers: { 'Content-Type': 'application/json' },
+          body: JSON.stringify(card),
+        });
+        if (!resp.ok) return { ok: false, status: resp.status, error: (await resp.text()).slice(0, 200) };
+        return { ok: true };
+      }
+
+      case 'ntfy': {
+        const serverUrl = channel.config.server_url || 'https://ntfy.sh';
+        const topic = channel.config.topic;
+        if (!topic) return { ok: false, error: 'ntfy channel missing topic' };
+        const headers: Record<string, string> = {
+          'Content-Type': 'text/plain',
+          'Title': 'Pulse channel verified',
+        };
+        if (channel.config.auth_token) headers['Authorization'] = `Bearer ${channel.config.auth_token}`;
+        const resp = await fetch(`${serverUrl}/${topic}`, {
+          method: 'POST',
+          headers,
+          body: TEST_MESSAGE_BODY,
+        });
+        if (!resp.ok) return { ok: false, status: resp.status, error: (await resp.text()).slice(0, 200) };
+        return { ok: true };
+      }
+
+      default:
+        return { ok: false, error: `Unsupported personal channel type: ${channel.channel_type}` };
+    }
+  } catch (e) {
+    return { ok: false, error: e instanceof Error ? e.message : 'unknown error' };
+  }
+}
+
+/** ChannelType values that personal channels may take in Phase 9 (D-02). */
+export const PERSONAL_CHANNEL_TYPES: ChannelType[] = ['teams', 'ntfy'];
+
+export function isPersonalChannelType(t: unknown): t is 'teams' | 'ntfy' {
+  return t === 'teams' || t === 'ntfy';
+}