Receive signed message notifications and queue reliable work

{
  "id": "8ALpC7nK2dQ4sE6rT9vX1Z",
  "type": "message.received",
  "data": {
    "messageId": "8ALpC7nK2dQ4sE6rT9vX1Z",
    "numberId": "4JfW8bN2mR6pT1qV9xK3cH",
    "receivedAt": "2026-09-30T12:00:00.000Z"
  }
}

import { createHmac, timingSafeEqual } from "node:crypto"
import { z } from "zod"
 
const eventSchema = z
  .object({
    id: z.string().min(1).max(128),
    type: z.literal("message.received"),
    data: z
      .object({
        messageId: z.string().min(1).max(128),
        numberId: z.string().min(1).max(128),
        receivedAt: z.iso.datetime(),
      })
      .strict(),
  })
  .strict()
  .refine((event) => event.id === event.data.messageId)
export type MessageReceivedEvent = z.infer<typeof eventSchema>
 
export interface ReceiverOptions {
  signingSecret: string
  /** Resolve only after an atomic, durable enqueue/dedup transaction commits. */
  enqueueOnce(event: MessageReceivedEvent): Promise<"queued" | "duplicate">
}
 
function validTimestamp(timestamp: string): boolean {
  return (
    /^\d{1,12}$/.test(timestamp) &&
    Math.abs(Date.now() / 1000 - Number(timestamp)) <= 300
  )
}
 
export function verifySignature(
  raw: Buffer,
  timestamp: string,
  signature: string,
  secret: string
): boolean {
  if (!validTimestamp(timestamp) || !/^[a-f0-9]{64}$/.test(signature))
    return false
  const expected = createHmac("sha256", secret)
    .update(timestamp + ".")
    .update(raw)
    .digest()
  return timingSafeEqual(expected, Buffer.from(signature, "hex"))
}
 
class BodyTooLarge extends Error {}
async function readRawBody(request: Request): Promise<Buffer> {
  const limit = 16_384
  const declared = Number(request.headers.get("content-length"))
  if (Number.isFinite(declared) && declared > limit) throw new BodyTooLarge()
  const reader = request.body?.getReader()
  if (!reader) return Buffer.alloc(0)
  const parts: Buffer[] = []
  let total = 0
  try {
    for (;;) {
      const { done, value } = await reader.read()
      if (done) return Buffer.concat(parts, total)
      total += value.byteLength
      if (total > limit) {
        await reader.cancel()
        throw new BodyTooLarge()
      }
      parts.push(Buffer.from(value))
    }
  } finally {
    reader.releaseLock()
  }
}
 
/** Mount the returned POST function in a Next.js Node-runtime Route Handler. */
export function createPhonekitReceiver(
  options: ReceiverOptions
): (request: Request) => Promise<Response> {
  if (!options.signingSecret) throw new Error("Set PHONEKIT_WEBHOOK_SECRET")
  return async function POST(request: Request): Promise<Response> {
    const timestamp = request.headers.get("phonekit-timestamp") ?? ""
    const signature = request.headers.get("phonekit-signature") ?? ""
    if (!validTimestamp(timestamp) || !/^[a-f0-9]{64}$/.test(signature))
      return new Response(null, { status: 401 })
    let raw: Buffer
    try {
      raw = await readRawBody(request)
    } catch (error) {
      return new Response(null, {
        status: error instanceof BodyTooLarge ? 413 : 400,
      })
    }
    if (!verifySignature(raw, timestamp, signature, options.signingSecret))
      return new Response(null, { status: 401 })
    let payload: unknown
    try {
      payload = JSON.parse(raw.toString("utf8")) as unknown
    } catch {
      return new Response(null, { status: 400 })
    }
    const parsed = eventSchema.safeParse(payload)
    if (!parsed.success) return new Response(null, { status: 400 })
    try {
      // Await persistence before acknowledging. No detached promise after response.
      await options.enqueueOnce(parsed.data)
      return new Response(null, { status: 204 })
    } catch {
      // Let the sender retry when the job could not be committed.
      return new Response(null, { status: 503 })
    }
  }
}

// Copy to app/api/phonekit/route.ts, keeping receiver.ts and jobs.ts alongside it.
import { createPhonekitReceiver } from "./receiver"
import { enqueueOnce } from "./jobs"
 
export const runtime = "nodejs"
const signingSecret = process.env["PHONEKIT_WEBHOOK_SECRET"]
if (!signingSecret) throw new Error("Set PHONEKIT_WEBHOOK_SECRET")
 
export const POST = createPhonekitReceiver({ signingSecret, enqueueOnce })

-- Run in your application's own database, not the Phonekit database.
CREATE TABLE phonekit_message_jobs (
  event_id text PRIMARY KEY,
  message_id text NOT NULL,
  number_id text NOT NULL,
  received_at timestamptz NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now(),
  completed_at timestamptz
);

import postgres from "postgres"
import type { MessageReceivedEvent } from "./receiver"
 
const databaseUrl = process.env["DATABASE_URL"]
if (!databaseUrl) throw new Error("Set your application's DATABASE_URL")
export const jobsDb = postgres(databaseUrl, { max: 5 })
 
/** The insert is both deduplication and durable queueing, in one atomic statement. */
export async function enqueueOnce(
  event: MessageReceivedEvent
): Promise<"queued" | "duplicate"> {
  const inserted = await jobsDb<{ event_id: string }[]>`
    INSERT INTO phonekit_message_jobs (event_id, message_id, number_id, received_at)
    VALUES (${event.id}, ${event.data.messageId}, ${event.data.numberId}, ${event.data.receivedAt})
    ON CONFLICT (event_id) DO NOTHING
    RETURNING event_id
  `
  return inserted.length ? "queued" : "duplicate"
}

import { PhonekitClient } from "@phonekit/sdk"
import type { PhonekitMessage } from "@phonekit/sdk"
import { jobsDb } from "./jobs"
 
/**
 * Invoke from a separately scheduled, single worker. A restart can repeat an
 * unfinished job: consume must be idempotent by eventId. No SMS is saved here.
 */
export async function processNextJob(
  consume: (input: {
    eventId: string
    message: PhonekitMessage
  }) => Promise<void>
): Promise<boolean> {
  const apiKey = process.env["PHONEKIT_API_KEY"]
  if (!apiKey) throw new Error("Set PHONEKIT_API_KEY in the worker")
  const [job] = await jobsDb<{ event_id: string; message_id: string }[]>`
    SELECT event_id, message_id FROM phonekit_message_jobs
    WHERE completed_at IS NULL
    ORDER BY created_at, event_id
    LIMIT 1
  `
  if (!job) return false
  const client = new PhonekitClient(apiKey)
  const message = await client.message(job.message_id)
  await consume({ eventId: job.event_id, message })
  await jobsDb`
    UPDATE phonekit_message_jobs SET completed_at = now()
    WHERE event_id = ${job.event_id}
  `
  return true
}