fix(alerts): make the Slack channel picker usable at scale

apps/api/src/services/SlackIntegrationService.test.ts

@@ -1,6 +1,6 @@
 import { createCipheriv, randomBytes } from "node:crypto"
 import { afterEach, assert, describe, it } from "@effect/vitest"
-import { ConfigProvider, Effect, Layer, Schema } from "effect"
+import { ConfigProvider, Effect, Fiber, Layer, Schema } from "effect"
 import { TestClock } from "effect/testing"
 import { FetchHttpClient } from "effect/unstable/http"
 import { OrgId, UserId } from "@maple/domain/http"
@@ -64,6 +64,48 @@ const makeLayer = (
 /** Mirror of the service's (unexported) `SLACK_STATE_TTL_MS` — 10 minutes. */
 const SLACK_STATE_TTL_MS = 10 * 60_000
 
+/** Mirror of the service's (unexported) `SLACK_MAX_CHANNEL_PAGES` runaway guard. */
+const SLACK_MAX_CHANNEL_PAGES = 20
+
+/** Mirror of the service's (unexported) `SLACK_CHANNEL_WALK_BUDGET_MS`. */
+const SLACK_CHANNEL_WALK_BUDGET_MS = 45_000
+
+/**
+ * Run an effect that interleaves real-promise fetch mocks with TestClock sleeps.
+ *
+ * The fetch mock resolves on the real microtask queue while the backoff sleeps
+ * sit on the TestClock, so neither drains the other and both have to be pumped.
+ * Pump until the fiber actually finishes rather than for a fixed number of
+ * turns: a fixed count is coupled to how many macrotask turns PGlite happens to
+ * take, and when it guesses low the failure mode is `Fiber.join` blocking to the
+ * vitest timeout instead of a readable assertion. Running out of steps fails
+ * loudly here instead.
+ */
+const runInterleaved = <A, E, R>(
+	effect: Effect.Effect<A, E, R>,
+	options?: { readonly stepMs?: number; readonly maxSteps?: number },
+) =>
+	Effect.gen(function* () {
+		const stepMs = options?.stepMs ?? 1_000
+		const maxSteps = options?.maxSteps ?? 400
+		let settled = false
+		const fiber = yield* Effect.forkChild(
+			effect.pipe(
+				Effect.onExit(() =>
+					Effect.sync(() => {
+						settled = true
+					}),
+				),
+			),
+		)
+		for (let step = 0; step < maxSteps && !settled; step++) {
+			yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 0)))
+			if (!settled) yield* TestClock.adjust(stepMs)
+		}
+		assert.isTrue(settled, `effect did not settle within ${maxSteps} interleaved steps`)
+		return yield* Fiber.join(fiber)
+	})
+
 /**
  * Wrap the real ApiKeysService so `create` runs `inject` first. `apiKeys.create`
  * is the only seam between completeInstall's cross-org pre-check and its
@@ -1048,6 +1090,7 @@ describe("SlackIntegrationService", () => {
 
 	it.effect("listChannels caps pagination at SLACK_MAX_CHANNEL_PAGES pages", () => {
 		const testDb = createTestDb(trackedDbs)
+		const pageUrls: string[] = []
 		let calls = 0
 		return Effect.gen(function* () {
 			yield* Effect.promise(() =>
@@ -1063,17 +1106,20 @@ describe("SlackIntegrationService", () => {
 			const slack = yield* SlackIntegrationService
 			const channels = yield* slack.listChannels(asOrgId("org_lc"))
 			// The mock ALWAYS hands back a next_cursor — the walk must stop at the
-			// page cap (SLACK_MAX_CHANNEL_PAGES = 3) instead of looping forever.
-			assert.strictEqual(calls, 3)
-			assert.deepStrictEqual(
-				channels.map((c) => c.id),
-				["C-page-0", "C-page-1", "C-page-2"],
-			)
+			// runaway guard (SLACK_MAX_CHANNEL_PAGES) instead of looping forever.
+			assert.strictEqual(calls, SLACK_MAX_CHANNEL_PAGES)
+			assert.strictEqual(channels.length, SLACK_MAX_CHANNEL_PAGES)
+			assert.strictEqual(channels[0]?.id, "C-page-0")
+			assert.strictEqual(channels.at(-1)?.id, `C-page-${SLACK_MAX_CHANNEL_PAGES - 1}`)
+			// 1000 per page (Slack's documented max) keeps a 10k-channel workspace
+			// inside the Tier 2 (~20 req/min) budget.
+			assert.strictEqual(new URL(pageUrls[0]!).searchParams.get("limit"), "1000")
 		}).pipe(
 			Effect.provide(
 				withFetch(
 					testDb,
-					slackApiFetch(CONVERSATIONS_URL, (_url, call) => {
+					slackApiFetch(CONVERSATIONS_URL, (url, call) => {
+						pageUrls.push(url)
 						calls++
 						return jsonResponse({
 							ok: true,
@@ -1086,6 +1132,173 @@ describe("SlackIntegrationService", () => {
 		)
 	})
 
+	it.effect("listChannels waits out a 429 and retries the same page", () => {
+		const testDb = createTestDb(trackedDbs)
+		let calls = 0
+		return Effect.gen(function* () {
+			yield* Effect.promise(() =>
+				insertWorkspace(testDb, {
+					id: "sw_lc429",
+					orgId: "org_lc",
+					teamId: "T-LC",
+					teamName: "LC",
+					botToken: "xoxb-lc-token",
+					apiKey: "maple_ak_lc",
+				}),
+			)
+			const slack = yield* SlackIntegrationService
+			// `conversations.list` is Tier 2 — a 429 must back off, not surface as an
+			// "unexpected payload" error (Slack sends an empty body with the 429).
+			const channels = yield* runInterleaved(slack.listChannels(asOrgId("org_lc")))
+			assert.strictEqual(calls, 2)
+			assert.deepStrictEqual(
+				channels.map((c) => c.id),
+				["C1"],
+			)
+		}).pipe(
+			Effect.provide(
+				withFetch(
+					testDb,
+					slackApiFetch(CONVERSATIONS_URL, (_url, call) => {
+						calls++
+						return call === 0
+							? new Response("", { status: 429, headers: { "retry-after": "2" } })
+							: jsonResponse({ ok: true, channels: [{ id: "C1", name: "one" }] })
+					}),
+				),
+			),
+		)
+	})
+
+	it.effect("listChannels stops at the overall time budget and returns what it collected", () => {
+		const testDb = createTestDb(trackedDbs)
+		let calls = 0
+		return Effect.gen(function* () {
+			yield* Effect.promise(() =>
+				insertWorkspace(testDb, {
+					id: "sw_lcbudget",
+					orgId: "org_lc",
+					teamId: "T-LC",
+					teamName: "LC",
+					botToken: "xoxb-lc-token",
+					apiKey: "maple_ak_lc",
+				}),
+			)
+			const slack = yield* SlackIntegrationService
+			// Page 1 is rate limited for a full Tier 2 window (60s), which is longer
+			// than the whole walk is allowed to take. The walk must give up at the
+			// budget and hand back page 0 rather than sleeping past Cloudflare's
+			// ~100s edge cutoff on a request nobody is listening to any more.
+			const channels = yield* runInterleaved(slack.listChannels(asOrgId("org_lc")), {
+				maxSteps: Math.ceil(SLACK_CHANNEL_WALK_BUDGET_MS / 1_000) + 30,
+			})
+			assert.deepStrictEqual(
+				channels.map((c) => c.id),
+				["C1"],
+			)
+			// Exactly two: the successful page and the 429. A third would mean the
+			// 60s `Retry-After` had been clamped below the budget and retried —
+			// `conversations.list` is Tier 2, so 60 is the honest wait.
+			assert.strictEqual(calls, 2)
+		}).pipe(
+			Effect.provide(
+				withFetch(
+					testDb,
+					slackApiFetch(CONVERSATIONS_URL, (_url, call) => {
+						calls++
+						if (call === 0) {
+							return jsonResponse({
+								ok: true,
+								channels: [{ id: "C1", name: "one" }],
+								response_metadata: { next_cursor: "cursor-2" },
+							})
+						}
+						if (call === 1) {
+							return new Response("", { status: 429, headers: { "retry-after": "60" } })
+						}
+						return jsonResponse({ ok: true, channels: [{ id: "C2", name: "two" }] })
+					}),
+				),
+			),
+		)
+	})
+
+	it.effect("listChannels retries a Slack 5xx and reports the status when it persists", () => {
+		const testDb = createTestDb(trackedDbs)
+		let calls = 0
+		return Effect.gen(function* () {
+			yield* Effect.promise(() =>
+				insertWorkspace(testDb, {
+					id: "sw_lc5xx",
+					orgId: "org_lc",
+					teamId: "T-LC",
+					teamName: "LC",
+					botToken: "xoxb-lc-token",
+					apiKey: "maple_ak_lc",
+				}),
+			)
+			const slack = yield* SlackIntegrationService
+			// A 1000-channel page can time out server-side; Slack answers with an
+			// HTML error page, not JSON. That must not surface as the misleading
+			// "returned a non-JSON response".
+			const error = yield* runInterleaved(Effect.flip(slack.listChannels(asOrgId("org_lc"))))
+			assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsUpstreamError")
+			assert.include(error.message, "HTTP 503")
+			assert.notInclude(error.message, "non-JSON")
+			// One attempt plus SLACK_RATE_LIMIT_RETRIES.
+			assert.strictEqual(calls, 4)
+		}).pipe(
+			Effect.provide(
+				withFetch(
+					testDb,
+					slackApiFetch(CONVERSATIONS_URL, () => {
+						calls++
+						return new Response("<html>upstream timeout</html>", {
+							status: 503,
+							headers: { "content-type": "text/html" },
+						})
+					}),
+				),
+			),
+		)
+	})
+
+	it.effect("listChannels recovers when a transient 5xx clears on retry", () => {
+		const testDb = createTestDb(trackedDbs)
+		let calls = 0
+		return Effect.gen(function* () {
+			yield* Effect.promise(() =>
+				insertWorkspace(testDb, {
+					id: "sw_lc5xxok",
+					orgId: "org_lc",
+					teamId: "T-LC",
+					teamName: "LC",
+					botToken: "xoxb-lc-token",
+					apiKey: "maple_ak_lc",
+				}),
+			)
+			const slack = yield* SlackIntegrationService
+			const channels = yield* runInterleaved(slack.listChannels(asOrgId("org_lc")))
+			assert.deepStrictEqual(
+				channels.map((c) => c.id),
+				["C1"],
+			)
+			assert.strictEqual(calls, 2)
+		}).pipe(
+			Effect.provide(
+				withFetch(
+					testDb,
+					slackApiFetch(CONVERSATIONS_URL, (_url, call) => {
+						calls++
+						return call === 0
+							? new Response("", { status: 500 })
+							: jsonResponse({ ok: true, channels: [{ id: "C1", name: "one" }] })
+					}),
+				),
+			),
+		)
+	})
+
 	it.effect("listChannels maps Slack ok:false to an upstream error", () => {
 		const testDb = createTestDb(trackedDbs)
 		return Effect.gen(function* () {