Send alerts to chat and paging services

import type { Message } from "@sdxc/messaging";

export interface SyncRun {
	id: string;
	teamId: string;
	source: string;
	error: string;
	failures: number;
	finishedAt: Date;
	url: string;
}

function incidentKey(run: SyncRun) {
	return `sync:${run.teamId}:${run.source}`;
}

export function syncFailedMessage(run: SyncRun): Message {
	return {
		title: `Sync from ${run.source} failed`,
		text: `The run stopped with \`${run.error}\`. Imported rows are kept.`,
		severity: run.failures > 1 ? "critical" : "warning",
		fields: [
			{ label: "Source", value: run.source, inline: true },
			{ label: "Failures in a row", value: String(run.failures), inline: true },
		],
		links: [{ label: "Open the run", url: run.url }],
		timestamp: run.finishedAt,
		key: incidentKey(run),
		data: { runId: run.id },
	};
}

export function syncRecoveredMessage(run: SyncRun): Message {
	return {
		title: `Sync from ${run.source} recovered`,
		severity: "success",
		links: [{ label: "Open the run", url: run.url }],
		timestamp: run.finishedAt,
		key: incidentKey(run),
		state: "resolved",
	};
}

import type { Destination } from "@sdxc/messaging";

import { DiscordWebhook } from "@sdxc/messaging/discord";
import { Opsgenie } from "@sdxc/messaging/opsgenie";
import { PagerDuty } from "@sdxc/messaging/pagerduty";
import { SlackBot, SlackWebhook } from "@sdxc/messaging/slack";
import { TelegramBot } from "@sdxc/messaging/telegram";
import { Webhook } from "@sdxc/messaging/webhook";
import { env } from "cloudflare:workers";

export type ChannelConfig =
	| { kind: "slack"; url: string }
	| { kind: "slack-bot"; channel: string }
	| { kind: "discord"; url: string; threadId?: string }
	| { kind: "telegram"; chatId: string }
	| { kind: "pagerduty"; routingKey: string }
	| { kind: "opsgenie"; apiKey: string; region: "us" | "eu" }
	| { kind: "webhook"; url: string; secret: string };

export function destinationFor(config: ChannelConfig): Destination {
	switch (config.kind) {
		case "slack":
			return new SlackWebhook({ url: config.url });
		case "slack-bot":
			return new SlackBot({
				token: () => env.SLACK_BOT_TOKEN,
				channel: config.channel,
			});
		case "discord":
			return new DiscordWebhook({ url: config.url, threadId: config.threadId });
		case "telegram":
			return new TelegramBot({
				token: () => env.TELEGRAM_BOT_TOKEN,
				chatId: config.chatId,
			});
		case "pagerduty":
			return new PagerDuty({ routingKey: config.routingKey, source: "sync" });
		case "opsgenie":
			return new Opsgenie({ apiKey: config.apiKey, region: config.region });
		case "webhook":
			return new Webhook({
				url: config.url,
				secret: config.secret,
				resolve: true,
			});
	}
}

import { DiscordWebhook } from "@sdxc/messaging/discord";
import { SlackWebhook } from "@sdxc/messaging/slack";
import { Webhook } from "@sdxc/messaging/webhook";

export function checkChannelUrl(kind: "slack" | "discord" | "webhook", url: string) {
	if (kind === "slack") return SlackWebhook.check(url);
	if (kind === "discord") return DiscordWebhook.check(url);
	return Webhook.check(url);
}

import { job, jobs } from "@sdxc/jobs";
import { MESSAGE_SCHEMA } from "@sdxc/messaging";
import * as s from "remix/data-schema";

export default jobs({
	deliverMessage: job({
		input: s.object({ channelId: s.string(), message: MESSAGE_SCHEMA }),
	}),
});

import type { JobMiddleware } from "@sdxc/jobs";
import type { Destination } from "@sdxc/messaging";
import type { ChannelConfig } from "~/app/notifications/destinations";

import { createContextKey } from "remix/router";

export interface DestinationFactory {
	(config: ChannelConfig): Destination;
}

export const Destinations = createContextKey<DestinationFactory>();

interface DestinationsEffect {
	key: typeof Destinations;
	value: DestinationFactory;
	property: "destinations";
}

export function destinations(
	factory: DestinationFactory,
): JobMiddleware<DestinationsEffect> {
	return async (ctx, next) => {
		ctx.set(Destinations, factory, { property: "destinations" });
		await next();
	};
}

import type { JobQueue } from "@sdxc/jobs/queue";
import type { Database } from "remix/data-table";
import type { DestinationFactory } from "~/app/jobs/middleware/destinations";

import { createJobDispatcher } from "@sdxc/jobs";

import jobs from "~/app/jobs";
import { database } from "~/app/jobs/middleware/database";
import { destinations } from "~/app/jobs/middleware/destinations";
import { queue } from "~/app/jobs/queue";
import { openDatabase } from "~/app/lib/database";
import { logger } from "~/app/logger";
import { destinationFor } from "~/app/notifications/destinations";

interface DispatcherOptions {
	queue?: JobQueue;
	openDb?: () => Database;
	destinations?: DestinationFactory;
}

export function createDispatcher(options: DispatcherOptions = {}) {
	let dispatcher = createJobDispatcher({
		logger,
		queue: options.queue ?? queue,
		middleware: [
			database(options.openDb ?? openDatabase),
			destinations(options.destinations ?? destinationFor),
		] as const,
		timeout: "2 minutes",
	});

	dispatcher.map(jobs.deliverMessage, () => import("~/app/jobs/deliver-message"));
	return dispatcher;
}

export const dispatcher = createDispatcher();

import { createBackoff } from "@sdxc/backoff";
import { createJobHandler } from "@sdxc/jobs";
import { isSuccess } from "@sdxc/result";

import Channels from "~/app/data/channel";
import jobs from "~/app/jobs";

const MAX_ATTEMPTS = 4;

const RETRY_BACKOFF = createBackoff({
	base: "30 seconds",
	max: "30 minutes",
	jitter: 0.2,
});

export default createJobHandler(jobs.deliverMessage, async (ctx) => {
	let channel = await Channels.find(ctx.database, ctx.input.channelId);
	if (!channel) return ctx.exit("The channel was removed");

	let destination = ctx.destinations(channel.config);
	ctx.log.set({ channel: { id: channel.id, provider: destination.provider } });

	let sent = await destination.send(ctx.input.message, { id: ctx.id });
	if (isSuccess(sent)) return;

	let error = sent.error;
	if (error.retryable && ctx.attempts < MAX_ATTEMPTS) {
		let delay = error.retryAfter ?? RETRY_BACKOFF.delay(ctx.attempts);
		return ctx.retry({ delay, reason: error.message, cause: error });
	}

	if (error.code === "gone") {
		await Channels.markBroken(ctx.database, channel.id, error.message);
	}
	return ctx.exit(error.message, { cause: error });
});

import type { JobEnqueuer } from "@sdxc/jobs";
import type { Message } from "@sdxc/messaging";
import type { Database } from "remix/data-table";

import Channels from "~/app/data/channel";
import jobs from "~/app/jobs";

const PAGING_KINDS: ReadonlySet<string> = new Set(["pagerduty", "opsgenie"]);

export async function notifyTeam(
	db: Database,
	enqueuer: JobEnqueuer,
	teamId: string,
	message: Message,
) {
	let pages = message.severity === "critical" || message.state === "resolved";
	let channels = await Channels.forTeam(db, teamId);
	let targets = channels.filter(
		(channel) => pages || !PAGING_KINDS.has(channel.config.kind),
	);

	await enqueuer.enqueueMany(
		jobs.deliverMessage,
		targets.map((channel) => ({ channelId: channel.id, message })),
	);
}

import type { Destination, Message, SendOptions, SentRef } from "@sdxc/messaging";

import { supports } from "@sdxc/messaging";

export async function sendOrUpdate(
	destination: Destination,
	message: Message,
	ref: SentRef | null,
	options: SendOptions,
) {
	if (message.state === "resolved" && ref && supports(destination, "update")) {
		return await destination.update(ref, message, options);
	}
	return await destination.send(message, options);
}

import { MESSAGE_SCHEMA } from "@sdxc/messaging";
import { isFailure } from "@sdxc/result";
import * as Webhooks from "@sdxc/webhooks";
import { env } from "cloudflare:workers";
import { createAction } from "remix/router";

import { recordAlert } from "~/app/data/alerts";
import routes from "~/routes/web";

export default createAction(routes.alerts.inbox, async (ctx) => {
	let verified = await Webhooks.verify(ctx.request, {
		secret: env.ALERTS_WEBHOOK_SECRET,
		schema: MESSAGE_SCHEMA,
	});
	if (isFailure(verified)) return new Response(null, { status: 401 });

	await recordAlert(ctx.db, verified.data.id, verified.data.payload);
	return new Response(null, { status: 204 });
});

import type { Message } from "@sdxc/messaging";

import { SlackWebhook } from "@sdxc/messaging/slack";

export class CompactSlack extends SlackWebhook {
	override render(message: Message) {
		let payload = super.render(message);
		payload.attachments[0].blocks = payload.attachments[0].blocks.filter(
			(block) => block.type !== "context",
		);
		return payload;
	}
}

import * as memory from "@sdxc/jobs/memory";
import { MemoryDestination } from "@sdxc/messaging/memory";
import { expect, test } from "vitest";

import { insertChannel } from "~/app/data/channel";
import jobs from "~/app/jobs";
import { createDispatcher } from "~/app/jobs/dispatcher";
import { createTestDatabase } from "~/app/test/database";

test("waits the delay a rate limit asked for, then delivers", async () => {
	let db = await createTestDatabase();
	let channel = await insertChannel(db, {
		teamId: "team_1",
		config: { kind: "discord", url: "https://discord.com/api/webhooks/1/token" },
	});

	let destination = new MemoryDestination();
	destination.failNext({ code: "rate-limited", retryAfter: 30_000 });

	let now = 0;
	let queue = memory.queue({ now: () => now });
	let dispatcher = createDispatcher({
		queue,
		openDb: () => db,
		destinations: () => destination,
	});

	await dispatcher.enqueue(jobs.deliverMessage, {
		channelId: channel.id,
		message: { title: "Sync from Stripe failed", severity: "warning" },
	});

	let run = () => queue.drain((delivery) => dispatcher.deliver(delivery));
	expect(await run()).toEqual([{ type: "retry", delay: 30_000 }]);
	expect(destination.messages).toEqual([]);

	now += 30_000;
	expect(await run()).toEqual([{ type: "ack" }]);
	expect(destination.last?.message.title).toBe("Sync from Stripe failed");
});

import { describeDestination } from "@sdxc/messaging/conformance";
import { http, HttpResponse } from "msw";
import { setupServer } from "msw/node";
import { afterAll, afterEach, beforeAll } from "vitest";

import { CompactSlack } from "~/app/notifications/compact-slack";

const WEBHOOK_URL = "https://hooks.slack.com/services/T000/B000/test-token";

const server = setupServer();

beforeAll(() => server.listen({ onUnhandledRequest: "error" }));
afterEach(() => server.resetHandlers());
afterAll(() => server.close());

describeDestination({
	name: "CompactSlack",
	capabilities: [],
	create() {
		server.use(http.post(WEBHOOK_URL, () => new HttpResponse("ok")));
		return new CompactSlack({ url: WEBHOOK_URL });
	},
	rateLimitNext(delayMs) {
		let headers = { "Retry-After": String(delayMs / 1000) };
		let limited = () =>
			new HttpResponse("rate_limited", { status: 429, headers });
		server.use(http.post(WEBHOOK_URL, limited, { once: true }));
	},
});