Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/openstatusHQ/openstatus. 🫖 Status page with uptime monitoring & API monitoring as code 🫖 openstatus.dev
bun drizzle-orm monitoring monitoring-as-code nextjs observability on-call open-source shadcn-ui status-page statuspage synthetic-monitoring tinybird turso uptime uptime-checker uptime-monitor
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460import { getLogger } from "@logtape/logtape";import { and, db, eq, isNull, sql } from "@openstatus/db";import { integration, pageSubscriber } from "@openstatus/db/src/schema";import { WebClient } from "@slack/web-api";import type { Context } from "hono";import { z } from "zod";
import { runAgent } from "./agent";import { buildConfirmationBlocks, getConfirmationText, type RefResolvers,} from "./blocks";import { findByThread, replace, store } from "./confirmation-store";import type { PendingPayload } from "./confirmation-store";import { publishHomeView } from "./home";import { toMrkdwn } from "./mrkdwn";import { getComponentNames, getPageDashboardLink } from "./page-urls";import { getRegistryTool, isSlackToolDraft } from "./registry-runner";import { resolveWorkspace } from "./workspace-resolver";
function makeRefResolvers(workspaceId: number): RefResolvers { return { page: (pageId) => getPageDashboardLink(workspaceId, pageId), componentNames: (ids) => getComponentNames(workspaceId, ids), };}
const logger = getLogger("api-server");
const processedEvents = new Map<string, number>();
function dedup(eventId: string): boolean { const now = Date.now(); for (const [id, ts] of processedEvents) { if (now - ts > 300_000) processedEvents.delete(id); } if (processedEvents.has(eventId)) return true; processedEvents.set(eventId, now); return false;}
const slackEventSchema = z.object({ type: z.string(), event: z .object({ type: z.string(), subtype: z.string().optional(), text: z.string().optional(), user: z.string().optional(), channel: z.string().optional(), channel_type: z.string().optional(), ts: z.string().optional(), thread_ts: z.string().optional(), bot_id: z.string().optional(), tab: z.string().optional(), }) .optional(), event_id: z.string().optional(), team_id: z.string().optional(), challenge: z.string().optional(),});
type SlackEvent = z.infer<typeof slackEventSchema>;
const threadMessageSchema = z.object({ user: z.string().optional(), bot_id: z.string().optional(), text: z.string().optional(), ts: z.string().optional(),});
type ThreadMessage = z.infer<typeof threadMessageSchema>;
const slackPlatformErrorSchema = z.object({ code: z.literal("slack_webapi_platform_error"), data: z.object({ error: z.string(), }),});
function isSlackPlatformError(err: unknown, errorCode: string): boolean { const parsed = slackPlatformErrorSchema.safeParse(err); return parsed.success && parsed.data.data.error === errorCode;}
// Asking permission to run a write tool ("shall I go ahead and publish this?").const PERMISSION_QUESTION = /\b(shall i|should i|do you want me to|would you like me to|want me to|ready for me to|can i)\b[^?]*\b(go ahead|proceed|publish|create|post|send|schedule|submit|add|resolve|update)\b/i;
// A change written out as message text rather than passed to a tool.const PROSE_DRAFT_FIELDS = /\*{0,2}(title|status|message|impact|components?|from|to)\*{0,2}\s*:/gi;
/** * Heuristic for the "model drafted in prose instead of calling the tool" failure * — logging only, so false positives are cheap. Both signals are required: a * plain answer can end in a question, and a list of reports can look tabular. */export function looksLikeUncardedDraft(text: string): boolean { if (!text) return false; if (!PERMISSION_QUESTION.test(text)) return false; return (text.match(PROSE_DRAFT_FIELDS) ?? []).length >= 2;}
export async function handleSlackEvent(c: Context) { const body = c.get("slackBody") as SlackEvent;
if (body.type === "url_verification") { return c.json({ challenge: body.challenge }); }
if (body.type !== "event_callback") { return c.json({ ok: true }); }
if (body.event_id && dedup(body.event_id)) { return c.json({ ok: true }); }
const promise = processEvent(body); promise.catch((err) => logger.error("slack event processing error", { error: err, teamId: body.team_id, eventId: body.event_id, }), );
return c.json({ ok: true });}
async function processEvent(body: SlackEvent) { const event = body.event; if (!event) return;
if (event.type === "app_uninstalled" || event.type === "tokens_revoked") { const teamId = body.team_id; if (teamId) { await db .delete(integration) .where( and( eq(integration.name, "slack-agent"), eq(integration.externalId, teamId), ), ); await db .update(pageSubscriber) .set({ unsubscribedAt: new Date(), updatedAt: new Date() }) .where( and( eq(pageSubscriber.channelType, "slack"), isNull(pageSubscriber.unsubscribedAt), sql`json_extract(${pageSubscriber.channelConfig}, '$.teamId') = ${teamId}`, ), ); logger.info("slack integration cleaned up", { teamId }); } return; }
if (event.type === "app_home_opened") { if (event.tab && event.tab !== "home") return; const teamId = body.team_id; const userId = event.user; if (!teamId || !userId) return; const resolved = await resolveWorkspace(teamId); if (!resolved) return; try { await publishHomeView(new WebClient(resolved.botToken), userId); } catch (err) { logger.error("slack failed to publish home view", { error: err, teamId }); } return; }
if (event.type !== "app_mention" && event.type !== "message") return; if (event.type === "message" && event.bot_id) return;
const ignoredSubtypes = [ "channel_join", "channel_leave", "channel_topic", "channel_purpose", "channel_name", ]; if (event.subtype && ignoredSubtypes.includes(event.subtype)) return;
const teamId = body.team_id; if (!teamId || !event.channel || !event.ts) return;
// A single mention arrives as BOTH an `app_mention` and a `message.*` event // (distinct event_ids, same message ts), so the event_id dedup above doesn't // catch the pair. Dedup on the message identity so we only respond once. // Runs before the first `await` so concurrent deliveries can't both pass. if (dedup(`msg:${event.channel}:${event.ts}`)) return;
const resolved = await resolveWorkspace(teamId); if (!resolved) { logger.warn("slack integration not found", { teamId }); return; }
const slack = new WebClient(resolved.botToken); const botUserId = resolved.botUserId; const threadTs = event.thread_ts ?? event.ts;
if (event.type === "message" && !event.text?.includes(`<@${botUserId}>`)) { return; }
logger.info("slack event received", { teamId, channel: event.channel, eventType: event.type, threadTs, user: event.user, });
let thinkingTs: string | undefined; try { const thinkingMsg = await slack.chat.postMessage({ channel: event.channel, thread_ts: threadTs, text: ":hourglass_flowing_sand: Thinking...", }); thinkingTs = thinkingMsg.ts; } catch (err) { if (isSlackPlatformError(err, "cannot_reply_to_message")) { logger.warn("slack cannot reply to message, falling back to top-level", { channel: event.channel, teamId, threadTs, }); try { const fallbackMsg = await slack.chat.postMessage({ channel: event.channel, text: ":hourglass_flowing_sand: Thinking...", }); thinkingTs = fallbackMsg.ts; } catch (fallbackErr) { logger.error("slack failed to post fallback thinking message", { error: fallbackErr, channel: event.channel, teamId, }); return; } } else { logger.error("slack failed to post thinking message", { error: err, channel: event.channel, teamId, threadTs, }); return; } }
if (!thinkingTs) { logger.error("slack thinking message returned no ts", { channel: event.channel, teamId, }); return; }
try { let thread: ThreadMessage[] = []; if (event.thread_ts) { const replies = await slack.conversations.replies({ channel: event.channel, ts: event.thread_ts, limit: 100, }); thread = ((replies.messages ?? []) as ThreadMessage[]).filter( (msg) => msg.ts !== thinkingTs, ); } else { thread = [{ user: event.user, text: event.text, ts: event.ts }]; }
logger.info("slack agent invoked", { teamId, channel: event.channel, threadTs, messageCount: thread.length, });
const result = await runAgent( resolved.workspace, thread, botUserId, event.text, { slackUserId: event.user ?? "", teamId }, );
logger.info("slack agent completed", { teamId, channel: event.channel, threadTs, toolCalls: result.toolResults.map((tr) => tr.toolName), finishReason: result.finishReason, stepCount: result.stepCount, hitStepLimit: result.hitStepLimit, });
// One pending action per thread (see findByThread/replace below). // If the model emits multiple destructive drafts in a single step we // only honour the first; the carrier's thread index can't represent // a queue, and forcing the user to confirm twice in a row is worse // UX than asking them to re-issue the second request. const confirmationResult = result.toolResults.find((tr) => isSlackToolDraft(tr.result), );
if (confirmationResult) { logger.info("slack confirmation requested", { teamId, channel: event.channel, threadTs, toolName: confirmationResult.toolName, }); await handleConfirmation( slack, event.channel, threadTs, thinkingTs, event.user ?? "", resolved.workspace.id, resolved.botToken, confirmationResult, ); } else { // No draft means no card. Distinguish a legitimate text answer from the // model drafting a change in prose and asking for permission instead of // calling the tool — the latter strands the user with nothing to click. if (looksLikeUncardedDraft(result.text)) { logger.warn("slack draft proposed without a card", { teamId, channel: event.channel, threadTs, finishReason: result.finishReason, stepCount: result.stepCount, hitStepLimit: result.hitStepLimit, readToolCalls: result.toolResults.map((tr) => tr.toolName), }); } await slack.chat.update({ channel: event.channel, ts: thinkingTs, text: result.text ? toMrkdwn(result.text) : "Done!", }); logger.info("slack response sent", { teamId, channel: event.channel, threadTs, }); } } catch (err) { logger.error("slack agent error", { error: err, channel: event.channel, teamId, threadTs, }); if (thinkingTs) { await slack.chat .update({ channel: event.channel, ts: thinkingTs, text: ":x: Something went wrong. Please try again.", }) .catch((updateErr: unknown) => { logger.error("slack failed to update error message", { error: updateErr, channel: event.channel, thinkingTs, }); }); } }}
async function handleConfirmation( slack: WebClient, channel: string, threadTs: string, thinkingTs: string, userId: string, workspaceId: number, botToken: string, confirmationResult: { toolName: string; result: unknown },) { if (!isSlackToolDraft(confirmationResult.result)) return; const draft = confirmationResult.result; const tool = getRegistryTool(draft.toolName); if (!tool) { logger.error("slack: registry tool not found", { toolName: draft.toolName, }); await slack.chat.update({ channel, ts: thinkingTs, text: ":x: Something went wrong. Please try again.", }); return; }
const payload: PendingPayload = { toolName: draft.toolName, input: draft.input, }; const text = getConfirmationText({ tool, input: draft.displayInput });
// findByThread + replace isn't atomic on its own — two concurrent // events on the same thread could both see `existing` and race on // replace. Atomicity here relies on the `dedup` map at the top of this // file suppressing duplicate event_ids, plus Slack's own per-thread // event throttling. Cross-process dedup is *not* covered; see note in // processedEvents. const existing = await findByThread(threadTs); if (existing) { await replace(existing.id, payload);
const blocks = await buildConfirmationBlocks({ actionId: existing.id, tool, input: draft.displayInput, resolvers: makeRefResolvers(workspaceId), }); await slack.chat.update({ channel, ts: thinkingTs, text, blocks }); await slack.chat.update({ channel, ts: existing.messageTs, text, blocks, }); } else { const actionId = await store({ workspaceId, botToken, channelId: channel, threadTs, messageTs: thinkingTs, userId, payload, });
const blocks = await buildConfirmationBlocks({ actionId, tool, input: draft.displayInput, resolvers: makeRefResolvers(workspaceId), }); await slack.chat.update({ channel, ts: thinkingTs, text, blocks }); }}