property-management-network/lib/webhooks/emit.ts

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<string, unknown>
): 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<EndpointRow[]> {
  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<string, unknown>
}): Promise<void> {
  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<string, unknown>,
          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<string, unknown>,
      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 }
}