import { randomUUID } from "crypto" import { and, eq } from "drizzle-orm" import { db } from "@/lib/db" import { webhook_deliveries, webhook_endpoints } from "@/lib/db/schema" import { PING_EVENT, type WebhookEnvelope, type WebhookEvent } from "./events" import { attemptDelivery } from "./deliver" /** Build the delivery envelope for an event. `id` is a unique per-event id. */ function buildEnvelope( event: WebhookEvent | typeof PING_EVENT, data: Record ): WebhookEnvelope { return { id: `evt_${randomUUID().replace(/-/g, "")}`, event, created_at: new Date().toISOString(), data, } } // ============================================================================ // Event emission — the one function domain code calls after a mutation. // // await emitWebhookEvent({ ownerId, event: "tenant.created", data: {...} }) // // It resolves the account's subscribed endpoints, records a pending delivery per // endpoint, and fires an immediate best-effort delivery in the background (this // app runs as a persistent Node server, so post-response work completes). The // webhooks cron is the safety net for retries and process restarts. // ============================================================================ type EndpointRow = typeof webhook_endpoints.$inferSelect async function activeSubscribedEndpoints( ownerId: string, event: WebhookEvent ): Promise { const endpoints = await db.query.webhook_endpoints.findMany({ where: and(eq(webhook_endpoints.user_id, ownerId), eq(webhook_endpoints.status, "active")), }) // Empty `events` means "subscribe to everything". return endpoints.filter((e) => e.events.length === 0 || e.events.includes(event)) } export async function emitWebhookEvent(opts: { ownerId: string event: WebhookEvent data: Record }): Promise { const { ownerId, event, data } = opts try { const endpoints = await activeSubscribedEndpoints(ownerId, event) if (!endpoints.length) return const envelope = buildEnvelope(event, data) // Give the immediate background attempt ~60s before the cron would retry, so // the two never race to double-deliver the same row. const nextAttempt = new Date(Date.now() + 60_000).toISOString() const rows = await db .insert(webhook_deliveries) .values( endpoints.map((e) => ({ user_id: ownerId, endpoint_id: e.id, event, payload: envelope as unknown as Record, next_attempt_at: nextAttempt, })) ) .returning() const byId = new Map(endpoints.map((e) => [e.id, e])) void Promise.allSettled( rows.map((r) => { const ep = byId.get(r.endpoint_id) return ep ? attemptDelivery(r, ep) : Promise.resolve(false) }) ).catch(() => {}) } catch { // The webhook subsystem must never break the originating request. } } /** * Synchronously deliver a one-off `ping` to a single endpoint (the dashboard * "Send test event" button). Awaited so the caller can show the result. */ export async function deliverTestPing( endpoint: EndpointRow ): Promise<{ ok: boolean; responseStatus: number | null; error: string | null }> { const envelope = buildEnvelope(PING_EVENT, { message: "Test event from Property Management Network", endpoint_id: endpoint.id, }) const [row] = await db .insert(webhook_deliveries) .values({ user_id: endpoint.user_id, endpoint_id: endpoint.id, event: PING_EVENT, payload: envelope as unknown as Record, max_attempts: 1, next_attempt_at: new Date().toISOString(), }) .returning() const ok = await attemptDelivery(row, endpoint) const updated = await db.query.webhook_deliveries.findFirst({ where: eq(webhook_deliveries.id, row.id), columns: { response_status: true, error: true }, }) return { ok, responseStatus: updated?.response_status ?? null, error: updated?.error ?? null } }