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.
6.0 kB · 178 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179import { and, eq, inArray, isNull, sql } from "@openstatus/db";import { pageComponent, pageSubscriber, pageSubscriberToPageComponent, selectPageSubscriberSchema,} from "@openstatus/db/src/schema";import { assertSafeUrl } from "@openstatus/utils";
import { emitAudit } from "../audit";import { requireScope } from "../auth";import { type ServiceContext, withTransaction } from "../context";import { ConflictError, ValidationError } from "../errors";import { assertSubscribersAllowed, loadPageForWorkspace, parseWorkspaceForContext,} from "./internal";import { CreatePageSubscriberInput } from "./schemas";
export type CreatePageSubscriberResult = { id: number; row: typeof pageSubscriber.$inferSelect; componentIds: number[];};
/** * Create a **vendor-added** page subscriber — a workspace member adds * someone (typically a partner) from the dashboard / RPC. Always * inserts with `source: "vendor"`, skips the verification flow, stamps * `acceptedAt` immediately. A `token` is still generated so the * partner can self-manage via `/manage/{token}` and * `/unsubscribe/{token}`. * * NOT for self-serve subscriptions originating from the public status * page — those go through `upsertEmailSubscription` in * `@openstatus/subscriptions`, which inserts with the default * `source: "self_signup"`, runs the email verification flow, and * intentionally emits no audit row (visitor action, not a workspace * mutation). * * The dedup queries below are deliberately source-agnostic: an active * row of *either* source for the same (page, email|webhookUrl) blocks * the vendor add. That prevents duplicate notification fanout when a * visitor self-subscribed before the vendor add. * * Single transaction: plan-gate, dedup, INSERT, join inserts, and the * audit emit all run in one `withTransaction` block. A failed audit * row rolls the subscriber back — fail-closed audit guarantee from * CLAUDE.md is preserved. */export async function createPageSubscriber(args: { ctx: ServiceContext; input: CreatePageSubscriberInput;}): Promise<CreatePageSubscriberResult> { const { ctx } = args; requireScope(ctx, "write"); const input = CreatePageSubscriberInput.parse(args.input); const componentIds = input.componentIds ?? [];
// String-only check (no DNS resolution) — a public name pointing at a // private address still passes. Delivery and test sends never follow redirects. if (input.channelType === "webhook") { await assertSafeUrl(input.webhookUrl); }
return withTransaction(ctx, async (tx) => { const pageWithWorkspace = await loadPageForWorkspace({ tx, pageId: input.pageId, workspaceId: ctx.workspace.id, }); assertSubscribersAllowed( parseWorkspaceForContext(pageWithWorkspace.workspace), );
if (componentIds.length > 0) { const valid = await tx .select({ id: pageComponent.id }) .from(pageComponent) .where( and( eq(pageComponent.pageId, input.pageId), inArray(pageComponent.id, componentIds), ), ) .all(); if (valid.length !== componentIds.length) { throw new ValidationError("Some components do not belong to this page"); } }
let emailLower: string | null = null; let webhookUrl: string | null = null; let channelConfig: string | null = null;
if (input.channelType === "email") { emailLower = input.email.toLowerCase(); const dup = await tx.query.pageSubscriber.findFirst({ where: and( eq(pageSubscriber.pageId, input.pageId), eq(pageSubscriber.email, emailLower), eq(pageSubscriber.channelType, "email"), isNull(pageSubscriber.unsubscribedAt), ), }); if (dup) { throw new ConflictError( "A subscriber with this email already exists for this page.", ); } } else { webhookUrl = input.webhookUrl; const dup = await tx.query.pageSubscriber.findFirst({ where: and( eq(pageSubscriber.pageId, input.pageId), sql`LOWER(${pageSubscriber.webhookUrl}) = ${webhookUrl.toLowerCase()}`, eq(pageSubscriber.channelType, "webhook"), isNull(pageSubscriber.unsubscribedAt), ), }); if (dup) { throw new ConflictError( "A subscriber with this webhook URL already exists for this page.", ); } // Empty array is truthy in JS, so check `length > 0` — otherwise // we'd persist `{"headers":[]}` instead of leaving the column null. channelConfig = input.headers && input.headers.length > 0 ? JSON.stringify({ headers: input.headers }) : null; }
const row = await tx .insert(pageSubscriber) .values({ channelType: input.channelType, email: emailLower, webhookUrl, channelConfig, pageId: input.pageId, source: "vendor", name: input.name ?? null, token: crypto.randomUUID(), acceptedAt: new Date(), expiresAt: null, }) .returning() .get();
if (componentIds.length > 0) { await tx .insert(pageSubscriberToPageComponent) .values( componentIds.map((compId) => ({ pageSubscriberId: row.id, pageComponentId: compId, })), ) .run(); }
// Strip the management `token` — capability for the self-manage / // unsubscribe URLs, must never land in the audit log. const parsed = selectPageSubscriberSchema.parse(row); const { token: _token, ...after } = parsed; await emitAudit(tx, ctx, { action: "page_subscriber.create", entityType: "page_subscriber", entityId: parsed.id, after, metadata: componentIds.length ? { componentIds } : undefined, });
return { id: row.id, row, componentIds }; });}