diff --git a/apps/server/src/routes/slack/resolve-page.ts b/apps/server/src/routes/slack/resolve-page.ts index 7c2e8e13b..6efa62164 100644 --- a/apps/server/src/routes/slack/resolve-page.ts +++ b/apps/server/src/routes/slack/resolve-page.ts @@ -1,4 +1,4 @@ -import { db, eq, or } from "@openstatus/db"; +import { db, eq, or, sql } from "@openstatus/db"; import { page } from "@openstatus/db/src/schema"; const BASE_DOMAIN_SUFFIX = ".openstatus.dev"; @@ -58,7 +58,12 @@ export async function resolvePageFromUrl( customDomain: page.customDomain, }) .from(page) - .where(or(eq(page.customDomain, host), eq(page.customDomain, bareHost))) + .where( + or( + sql`LOWER(${page.customDomain}) = ${host}`, + sql`LOWER(${page.customDomain}) = ${bareHost}`, + ), + ) .get(); return row ?? null; } diff --git a/packages/subscriptions/src/channels/slack-store.ts b/packages/subscriptions/src/channels/slack-store.ts index 66406ed76..77d0e38c4 100644 --- a/packages/subscriptions/src/channels/slack-store.ts +++ b/packages/subscriptions/src/channels/slack-store.ts @@ -16,13 +16,17 @@ export interface SlackAnchorStore { anchor: SlackThreadAnchor, ): Promise; clearAnchor(reportId: number, subscriberId: number): Promise; - isDelivered( + // Atomically claim delivery of (report, subscriber, update). Returns true only + // for the caller that wins the claim; concurrent callers get false and must + // skip. This is the dedupe reservation — a single atomic op, not a + // read-then-write pair, so two dispatchers can't both post the same message. + reserveDelivery( reportId: number, subscriberId: number, updateId: number, ): Promise; - // Recorded only after a successful post, so a failed delivery can be retried. - markDelivered( + // Release a reservation whose post failed, so the delivery can be retried. + releaseDelivery( reportId: number, subscriberId: number, updateId: number, @@ -68,16 +72,16 @@ export function createRedisAnchorStore(): SlackAnchorStore { async clearAnchor(reportId, subscriberId) { await getRedis().del(anchorKey(reportId, subscriberId)); }, - async isDelivered(reportId, subscriberId, updateId) { - const res = await getRedis().get( + async reserveDelivery(reportId, subscriberId, updateId) { + const res = await getRedis().set( deliveredKey(reportId, subscriberId, updateId), + 1, + { ex: TTL_SECONDS, nx: true }, ); - return res !== null; + return res === "OK"; }, - async markDelivered(reportId, subscriberId, updateId) { - await getRedis().set(deliveredKey(reportId, subscriberId, updateId), 1, { - ex: TTL_SECONDS, - }); + async releaseDelivery(reportId, subscriberId, updateId) { + await getRedis().del(deliveredKey(reportId, subscriberId, updateId)); }, }; } @@ -95,11 +99,14 @@ export function createMemoryAnchorStore(): SlackAnchorStore { async clearAnchor(reportId, subscriberId) { anchors.delete(anchorKey(reportId, subscriberId)); }, - async isDelivered(reportId, subscriberId, updateId) { - return delivered.has(deliveredKey(reportId, subscriberId, updateId)); + async reserveDelivery(reportId, subscriberId, updateId) { + const key = deliveredKey(reportId, subscriberId, updateId); + if (delivered.has(key)) return false; + delivered.add(key); + return true; }, - async markDelivered(reportId, subscriberId, updateId) { - delivered.add(deliveredKey(reportId, subscriberId, updateId)); + async releaseDelivery(reportId, subscriberId, updateId) { + delivered.delete(deliveredKey(reportId, subscriberId, updateId)); }, }; } diff --git a/packages/subscriptions/src/channels/slack.test.ts b/packages/subscriptions/src/channels/slack.test.ts index 93898d1ff..3c6f501a3 100644 --- a/packages/subscriptions/src/channels/slack.test.ts +++ b/packages/subscriptions/src/channels/slack.test.ts @@ -1,7 +1,11 @@ import { describe, expect, test } from "bun:test"; import type { PageUpdate, Subscription } from "../types"; -import { type SlackClient, createSlackChannel } from "./slack"; +import { + type SlackClient, + createSlackChannel, + validateSlackConfig, +} from "./slack"; import { createMemoryAnchorStore } from "./slack-store"; interface RecordedCall { @@ -129,6 +133,42 @@ describe("createSlackChannel", () => { expect(calls.filter((c) => c.method === "post").length).toBe(1); }); + test("failed post releases the reservation so a retry re-posts", async () => { + let attempts = 0; + const posts: number[] = []; + const client: SlackClient = { + async postMessage() { + attempts += 1; + posts.push(attempts); + if (attempts === 1) { + // Non-terminal error: runSlack returns null, no soft-unsubscribe. + const error = new Error("slack error") as Error & { + data?: { error?: string }; + }; + error.data = { error: "ratelimited" }; + throw error; + } + return { ts: "1700000000.0001" }; + }, + async update(args) { + return { ts: args.ts }; + }, + }; + const store = createMemoryAnchorStore(); + const channel = createSlackChannel({ + store, + createClient: () => client, + getBotToken: token, + softUnsubscribe: noopUnsub, + }); + + await channel.sendNotifications([makeSub()], makeUpdate({ updateId: 100 })); + await channel.sendNotifications([makeSub()], makeUpdate({ updateId: 100 })); + + // First attempt failed and released; second attempt reserved again and posted. + expect(posts.length).toBe(2); + }); + test("missing bot token delivers nothing", async () => { const { client, calls } = makeClient(); const channel = createSlackChannel({ @@ -178,3 +218,49 @@ describe("createSlackChannel", () => { expect(await store.getAnchor(10, 1)).toBeNull(); }); }); + +describe("validateSlackConfig", () => { + test("accepts non-empty teamId and channelId strings", async () => { + const result = await validateSlackConfig({ + teamId: "T1", + channelId: "C1", + }); + expect(result.valid).toBe(true); + }); + + const invalidConfigs: Array<{ label: string; config: unknown }> = [ + { label: "null config", config: null }, + { label: "non-object", config: "T1" }, + { label: "missing teamId", config: { channelId: "C1" } }, + { label: "missing channelId", config: { teamId: "T1" } }, + { label: "null teamId", config: { teamId: null, channelId: "C1" } }, + { label: "numeric channelId", config: { teamId: "T1", channelId: 42 } }, + { label: "empty teamId", config: { teamId: "", channelId: "C1" } }, + { label: "empty channelId", config: { teamId: "T1", channelId: "" } }, + ]; + + for (const { label, config } of invalidConfigs) { + test(`rejects ${label}`, async () => { + const result = await validateSlackConfig(config); + expect(result.valid).toBe(false); + expect(result.error).toBeDefined(); + }); + } +}); + +describe("reserveDelivery (memory store)", () => { + test("only the first reservation for a key wins", async () => { + const store = createMemoryAnchorStore(); + expect(await store.reserveDelivery(1, 2, 3)).toBe(true); + expect(await store.reserveDelivery(1, 2, 3)).toBe(false); + // A different updateId is an independent reservation. + expect(await store.reserveDelivery(1, 2, 4)).toBe(true); + }); + + test("releaseDelivery makes the key reservable again", async () => { + const store = createMemoryAnchorStore(); + expect(await store.reserveDelivery(1, 2, 3)).toBe(true); + await store.releaseDelivery(1, 2, 3); + expect(await store.reserveDelivery(1, 2, 3)).toBe(true); + }); +}); diff --git a/packages/subscriptions/src/channels/slack.ts b/packages/subscriptions/src/channels/slack.ts index e9b045ca1..e469020af 100644 --- a/packages/subscriptions/src/channels/slack.ts +++ b/packages/subscriptions/src/channels/slack.ts @@ -135,7 +135,9 @@ export function createSlackChannel(deps: SlackChannelDeps) { return; } - if (await deps.store.isDelivered(reportId, sub.id, updateId)) return; + // Atomic dedupe: only the caller that wins this reservation posts. On a + // failed post we release it below so the delivery stays retriable. + if (!(await deps.store.reserveDelivery(reportId, sub.id, updateId))) return; const anchor = await deps.store.getAnchor(reportId, sub.id); const root = buildRootMessage(pageUpdate, sub); @@ -148,11 +150,13 @@ export function createSlackChannel(deps: SlackChannelDeps) { attachments: root.attachments, }), ); - if (!res) return; + if (!res) { + await deps.store.releaseDelivery(reportId, sub.id, updateId); + return; + } if (res.ts) { await deps.store.setAnchor(reportId, sub.id, { ts: res.ts, channelId }); } - await deps.store.markDelivered(reportId, sub.id, updateId); return; } @@ -165,8 +169,10 @@ export function createSlackChannel(deps: SlackChannelDeps) { blocks: reply.blocks, }), ); - if (!replyRes) return; - await deps.store.markDelivered(reportId, sub.id, updateId); + if (!replyRes) { + await deps.store.releaseDelivery(reportId, sub.id, updateId); + return; + } // Re-render the root so its emoji/status track the latest state. await runSlack(sub.id, () => @@ -248,8 +254,10 @@ export async function validateSlackConfig( if ( config !== null && typeof config === "object" && - "teamId" in config && - "channelId" in config + typeof (config as { teamId?: unknown }).teamId === "string" && + (config as { teamId: string }).teamId.length > 0 && + typeof (config as { channelId?: unknown }).channelId === "string" && + (config as { channelId: string }).channelId.length > 0 ) { return { valid: true }; }