feat(09-02): personal channels service + /api/me/channels routes

- lib/services/personal-channels.ts: isValidTeamsWebhookUrl, isValidNtfyTopic,
  mintNtfyTopic, sendChannelTest, TEST_MESSAGE_BODY, isPersonalChannelType,
  PERSONAL_CHANNEL_TYPES
- GET /api/me/channels: returns user's personal channels (owner_user_id scoped)
- PUT /api/me/channels/[type]: WITH-CTE UPSERT + best-effort test send
- DELETE /api/me/channels/[type]: removes user's channel, 404 if missing
- POST /api/me/channels/[type]/test: re-sends test to existing channel
- SSRF mitigation via Teams URL hostname allowlist (T-09-02-06)
- Race window closed by partial unique index from Plan 01 (T-09-02-10)

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 },
+    );
+  }
+}

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 },
+    );
+  }
+}

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 },
+    );
+  }
+}

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';
+}