Reliable Node.js webhook and job processing

/**
 * Discovery example: reliable webhook ingestion and leased background work.
 * Declared runtime: Node.js 24.
 *
 * The store is intentionally expressed as a transactional interface so the
 * concurrency contract is visible without binding the example to one driver.
 */

import { createHmac, randomUUID, timingSafeEqual } from "node:crypto"
import { setTimeout as delay } from "node:timers/promises"

type Json = null | boolean | number | string | Json[] | { [key: string]: Json }
type JobState = "pending" | "running" | "retry_wait" | "succeeded" | "dead"

type WebhookEnvelope = {
  id: string
  type: string
  createdAt: string
  payload: Json
}

type JobRecord = {
  id: string
  tenantId: string
  operationKey: string
  eventId: string
  eventType: string
  payload: Json
  state: JobState
  attempts: number
  maxAttempts: number
  availableAt: Date
  leaseOwner: string | null
  leaseToken: string | null
  leaseExpiresAt: Date | null
  lastError: SerializedFailure | null
  createdAt: Date
  updatedAt: Date
  completedAt: Date | null
}

type JobEvent = {
  id: string
  tenantId: string
  jobId: string
  kind: "created" | "claimed" | "lease_renewed" | "retry_scheduled" | "succeeded" | "dead"
  details: Record<string, Json>
  createdAt: Date
}

type SerializedFailure = {
  name: string
  message: string
  code: string
  retryable: boolean
  stack?: string
}

type Claim = {
  job: JobRecord
  owner: string
  token: string
}

type RetryPolicy = {
  maxAttempts: number
  baseDelayMs: number
  maxDelayMs: number
  leaseMs: number
  heartbeatMs: number
  attemptTimeoutMs: number
}

type Logger = {
  info(event: string, fields?: Record<string, Json>): void
  warn(event: string, fields?: Record<string, Json>): void
  error(event: string, fields?: Record<string, Json>): void
}

class AppError extends Error {
  constructor(
    message: string,
    readonly code: string,
    readonly retryable: boolean,
    readonly status = 500,
    options?: ErrorOptions,
  ) {
    super(message, options)
    this.name = "AppError"
  }
}

class SignatureError extends AppError {
  constructor(message = "Webhook signature verification failed.") {
    super(message, "INVALID_SIGNATURE", false, 401)
    this.name = "SignatureError"
  }
}

class LeaseLostError extends AppError {
  constructor(jobId: string) {
    super(`Lease ownership was lost for job ${jobId}.`, "LEASE_LOST", true, 409)
    this.name = "LeaseLostError"
  }
}

function serializeFailure(error: unknown): SerializedFailure {
  if (error instanceof AppError) {
    return {
      name: error.name,
      message: error.message,
      code: error.code,
      retryable: error.retryable,
      stack: error.stack,
    }
  }
  if (error instanceof Error) {
    return {
      name: error.name,
      message: error.message,
      code: "UNEXPECTED_ERROR",
      retryable: true,
      stack: error.stack,
    }
  }
  return {
    name: "UnknownFailure",
    message: String(error),
    code: "UNKNOWN_FAILURE",
    retryable: false,
  }
}

function jsonLogger(scope: string): Logger {
  const write = (level: string, event: string, fields: Record<string, Json> = {}) => {
    process.stdout.write(`${JSON.stringify({
      at: new Date().toISOString(),
      level,
      scope,
      event,
      ...fields,
    })}\n`)
  }
  return {
    info: (event, fields) => write("info", event, fields),
    warn: (event, fields) => write("warn", event, fields),
    error: (event, fields) => write("error", event, fields),
  }
}

function parseSignatureHeader(value: string | null): { timestamp: number; signatures: string[] } {
  if (!value) throw new SignatureError("The signature header is missing.")
  const fields = value.split(",").map((part) => part.trim().split("=", 2))
  const timestamp = Number(fields.find(([key]) => key === "t")?.[1])
  const signatures = fields.filter(([key]) => key === "v1").map(([, signature]) => signature)
  if (!Number.isSafeInteger(timestamp) || signatures.length === 0) {
    throw new SignatureError("The signature header is malformed.")
  }
  return { timestamp, signatures }
}

function equalHex(left: string, right: string): boolean {
  if (!/^[a-f0-9]+$/i.test(left) || !/^[a-f0-9]+$/i.test(right)) return false
  const leftBytes = Buffer.from(left, "hex")
  const rightBytes = Buffer.from(right, "hex")
  return leftBytes.length === rightBytes.length && timingSafeEqual(leftBytes, rightBytes)
}

function verifyWebhook(input: {
  body: Uint8Array
  signatureHeader: string | null
  secret: string
  now?: Date
  toleranceSeconds?: number
}): void {
  const { timestamp, signatures } = parseSignatureHeader(input.signatureHeader)
  const nowSeconds = Math.floor((input.now ?? new Date()).getTime() / 1_000)
  const tolerance = input.toleranceSeconds ?? 300
  if (Math.abs(nowSeconds - timestamp) > tolerance) {
    throw new SignatureError("The webhook timestamp is outside the allowed window.")
  }
  const signedPayload = Buffer.concat([
    Buffer.from(`${timestamp}.`, "utf8"),
    Buffer.from(input.body),
  ])
  const expected = createHmac("sha256", input.secret).update(signedPayload).digest("hex")
  if (!signatures.some((signature) => equalHex(signature, expected))) {
    throw new SignatureError()
  }
}

function parseEnvelope(body: Uint8Array): WebhookEnvelope {
  let value: unknown
  try {
    value = JSON.parse(Buffer.from(body).toString("utf8"))
  } catch (error) {
    throw new AppError("Webhook JSON is invalid.", "INVALID_JSON", false, 400, { cause: error })
  }
  if (!value || typeof value !== "object" || Array.isArray(value)) {
    throw new AppError("Webhook payload must be an object.", "INVALID_EVENT", false, 400)
  }
  const event = value as Record<string, unknown>
  if (
    typeof event.id !== "string" || event.id.length < 4 ||
    typeof event.type !== "string" || event.type.length < 3 ||
    typeof event.createdAt !== "string" || Number.isNaN(Date.parse(event.createdAt)) ||
    !("payload" in event)
  ) {
    throw new AppError("Webhook fields are invalid.", "INVALID_EVENT", false, 400)
  }
  return {
    id: event.id,
    type: event.type,
    createdAt: event.createdAt,
    payload: event.payload as Json,
  }
}

interface Transaction {
  findJobByOperationKey(tenantId: string, operationKey: string): Promise<JobRecord | null>
  insertJob(job: JobRecord): Promise<void>
  insertEvent(event: JobEvent): Promise<void>
}

interface JobStore {
  transaction<T>(operation: (tx: Transaction) => Promise<T>): Promise<T>
  claimNext(input: {
    tenantId: string
    owner: string
    leaseMs: number
    now: Date
  }): Promise<Claim | null>
  renewLease(input: {
    tenantId: string
    jobId: string
    owner: string
    token: string
    leaseMs: number
    now: Date
  }): Promise<boolean>
  complete(input: {
    tenantId: string
    jobId: string
    owner: string
    token: string
    now: Date
    result: Json
  }): Promise<boolean>
  fail(input: {
    tenantId: string
    jobId: string
    owner: string
    token: string
    now: Date
    failure: SerializedFailure
    nextAvailableAt: Date | null
  }): Promise<boolean>
  releaseExpiredLeases(input: { tenantId: string; now: Date; limit: number }): Promise<number>
}

function operationKey(tenantId: string, event: WebhookEnvelope): string {
  return `${tenantId}:webhook:${event.id}`
}

async function enqueueWebhook(
  store: JobStore,
  input: { tenantId: string; event: WebhookEnvelope; maxAttempts: number; now?: Date },
): Promise<{ job: JobRecord; duplicate: boolean }> {
  const now = input.now ?? new Date()
  const key = operationKey(input.tenantId, input.event)
  return await store.transaction(async (tx) => {
    const existing = await tx.findJobByOperationKey(input.tenantId, key)
    if (existing) return { job: existing, duplicate: true }
    const job: JobRecord = {
      id: randomUUID(),
      tenantId: input.tenantId,
      operationKey: key,
      eventId: input.event.id,
      eventType: input.event.type,
      payload: input.event.payload,
      state: "pending",
      attempts: 0,
      maxAttempts: input.maxAttempts,
      availableAt: now,
      leaseOwner: null,
      leaseToken: null,
      leaseExpiresAt: null,
      lastError: null,
      createdAt: now,
      updatedAt: now,
      completedAt: null,
    }
    await tx.insertJob(job)
    await tx.insertEvent({
      id: randomUUID(),
      tenantId: input.tenantId,
      jobId: job.id,
      kind: "created",
      details: { eventId: input.event.id, eventType: input.event.type },
      createdAt: now,
    })
    return { job, duplicate: false }
  })
}

type TenantResolver = (request: Request, event: WebhookEnvelope) => Promise<string>

function createWebhookHandler(dependencies: {
  secret: string
  store: JobStore
  resolveTenant: TenantResolver
  logger?: Logger
}) {
  const logger = dependencies.logger ?? jsonLogger("webhook")
  return async (request: Request): Promise<Response> => {
    const body = new Uint8Array(await request.arrayBuffer())
    try {
      verifyWebhook({
        body,
        signatureHeader: request.headers.get("webhook-signature"),
        secret: dependencies.secret,
      })
      const event = parseEnvelope(body)
      const tenantId = await dependencies.resolveTenant(request, event)
      const queued = await enqueueWebhook(dependencies.store, {
        tenantId,
        event,
        maxAttempts: 6,
      })
      logger.info("webhook.accepted", {
        tenantId,
        eventId: event.id,
        jobId: queued.job.id,
        duplicate: queued.duplicate,
      })
      return Response.json(
        { accepted: true, duplicate: queued.duplicate, jobId: queued.job.id },
        { status: queued.duplicate ? 200 : 202 },
      )
    } catch (error) {
      const failure = serializeFailure(error)
      logger.warn("webhook.rejected", { code: failure.code, message: failure.message })
      const status = error instanceof AppError ? error.status : 500
      return Response.json({ error: { code: failure.code, message: failure.message } }, { status })
    }
  }
}

function backoffDelay(policy: RetryPolicy, attempt: number): number {
  const exponential = Math.min(policy.baseDelayMs * 2 ** Math.max(0, attempt - 1), policy.maxDelayMs)
  return Math.round(exponential * (0.8 + Math.random() * 0.4))
}

function abortReason(signal: AbortSignal): Error {
  return signal.reason instanceof Error
    ? signal.reason
    : new DOMException("The operation was aborted.", "AbortError")
}

function composeAttemptSignal(parent: AbortSignal, timeoutMs: number): AbortSignal {
  return AbortSignal.any([parent, AbortSignal.timeout(timeoutMs)])
}

type JobProcessor = (job: JobRecord, context: { signal: AbortSignal; logger: Logger }) => Promise<Json>

class LeaseHeartbeat implements AsyncDisposable {
  readonly controller = new AbortController()
  private timer: ReturnType<typeof setInterval> | null = null
  private stopped = false

  constructor(
    private readonly store: JobStore,
    private readonly claim: Claim,
    private readonly policy: RetryPolicy,
    private readonly logger: Logger,
  ) {}

  start(parent: AbortSignal): AbortSignal {
    const combined = AbortSignal.any([parent, this.controller.signal])
    this.timer = setInterval(() => {
      void this.tick().catch((error) => {
        this.logger.error("job.heartbeat_failed", {
          jobId: this.claim.job.id,
          message: error instanceof Error ? error.message : String(error),
        })
        this.controller.abort(error)
      })
    }, this.policy.heartbeatMs)
    this.timer.unref()
    return combined
  }

  private async tick(): Promise<void> {
    if (this.stopped) return
    const renewed = await this.store.renewLease({
      tenantId: this.claim.job.tenantId,
      jobId: this.claim.job.id,
      owner: this.claim.owner,
      token: this.claim.token,
      leaseMs: this.policy.leaseMs,
      now: new Date(),
    })
    if (!renewed) throw new LeaseLostError(this.claim.job.id)
  }

  async [Symbol.asyncDispose](): Promise<void> {
    this.stopped = true
    if (this.timer) clearInterval(this.timer)
    this.timer = null
  }
}

class JobWorker {
  private readonly controller = new AbortController()
  private readonly running = new Set<Promise<void>>()
  private accepting = true

  constructor(
    private readonly tenantId: string,
    private readonly owner: string,
    private readonly store: JobStore,
    private readonly processor: JobProcessor,
    private readonly policy: RetryPolicy,
    private readonly concurrency: number,
    private readonly logger = jsonLogger("worker"),
  ) {
    if (!Number.isInteger(concurrency) || concurrency < 1) {
      throw new RangeError("Worker concurrency must be a positive integer.")
    }
    if (policy.heartbeatMs >= policy.leaseMs / 2) {
      throw new RangeError("Heartbeat interval must be less than half the lease duration.")
    }
  }

  async run(): Promise<void> {
    this.installSignalHandlers()
    await this.store.releaseExpiredLeases({ tenantId: this.tenantId, now: new Date(), limit: 100 })
    try {
      while (this.accepting && !this.controller.signal.aborted) {
        while (this.running.size < this.concurrency && this.accepting) {
          const claim = await this.store.claimNext({
            tenantId: this.tenantId,
            owner: this.owner,
            leaseMs: this.policy.leaseMs,
            now: new Date(),
          })
          if (!claim) break
          const task = this.execute(claim)
          this.running.add(task)
          void task.finally(() => this.running.delete(task))
        }
        if (this.running.size === 0) {
          await delay(250, undefined, { signal: this.controller.signal }).catch(() => undefined)
        } else if (this.running.size >= this.concurrency) {
          await Promise.race(this.running)
        } else {
          await delay(50, undefined, { signal: this.controller.signal }).catch(() => undefined)
        }
      }
    } finally {
      this.accepting = false
      await Promise.allSettled(this.running)
      this.removeSignalHandlers()
    }
  }

  stop(reason = new AppError("Worker shutdown requested.", "WORKER_STOPPING", true, 503)): void {
    if (!this.accepting) return
    this.accepting = false
    this.controller.abort(reason)
  }

  private readonly onSignal = () => this.stop()

  private installSignalHandlers(): void {
    process.once("SIGTERM", this.onSignal)
    process.once("SIGINT", this.onSignal)
  }

  private removeSignalHandlers(): void {
    process.off("SIGTERM", this.onSignal)
    process.off("SIGINT", this.onSignal)
  }

  private async execute(claim: Claim): Promise<void> {
    const heartbeat = new LeaseHeartbeat(this.store, claim, this.policy, this.logger)
    await using lease = heartbeat
    const attemptSignal = composeAttemptSignal(lease.start(this.controller.signal), this.policy.attemptTimeoutMs)
    this.logger.info("job.started", {
      jobId: claim.job.id,
      attempt: claim.job.attempts,
      owner: claim.owner,
    })
    try {
      const result = await this.processor(claim.job, { signal: attemptSignal, logger: this.logger })
      if (attemptSignal.aborted) throw abortReason(attemptSignal)
      const completed = await this.store.complete({
        tenantId: claim.job.tenantId,
        jobId: claim.job.id,
        owner: claim.owner,
        token: claim.token,
        now: new Date(),
        result,
      })
      if (!completed) throw new LeaseLostError(claim.job.id)
      this.logger.info("job.succeeded", { jobId: claim.job.id })
    } catch (error) {
      await this.recordFailure(claim, error)
    }
  }

  private async recordFailure(claim: Claim, error: unknown): Promise<void> {
    const failure = serializeFailure(error)
    const attemptsExhausted = claim.job.attempts >= claim.job.maxAttempts
    const retryable = failure.retryable && !attemptsExhausted
    const nextAvailableAt = retryable
      ? new Date(Date.now() + backoffDelay(this.policy, claim.job.attempts))
      : null
    const recorded = await this.store.fail({
      tenantId: claim.job.tenantId,
      jobId: claim.job.id,
      owner: claim.owner,
      token: claim.token,
      now: new Date(),
      failure,
      nextAvailableAt,
    })
    if (!recorded) {
      this.logger.warn("job.failure_not_recorded", { jobId: claim.job.id, code: failure.code })
      return
    }
    this.logger.warn(retryable ? "job.retry_scheduled" : "job.dead", {
      jobId: claim.job.id,
      code: failure.code,
      retryable,
      nextAvailableAt: nextAvailableAt?.toISOString() ?? null,
    })
  }
}

type Delivery = {
  destination: URL
  body: Json
  idempotencyKey: string
}

async function deliverWebhook(delivery: Delivery, signal: AbortSignal): Promise<Json> {
  const response = await fetch(delivery.destination, {
    method: "POST",
    headers: {
      "content-type": "application/json",
      "idempotency-key": delivery.idempotencyKey,
    },
    body: JSON.stringify(delivery.body),
    signal,
  })
  if (!response.ok) {
    const retryable = response.status === 408 || response.status === 429 || response.status >= 500
    throw new AppError(
      `Delivery returned HTTP ${response.status}.`,
      `DELIVERY_HTTP_${response.status}`,
      retryable,
      response.status,
    )
  }
  return { status: response.status, deliveredAt: new Date().toISOString() }
}

function createProcessor(destinationForTenant: (tenantId: string) => Promise<URL>): JobProcessor {
  return async (job, context) => {
    context.signal.throwIfAborted()
    const destination = await destinationForTenant(job.tenantId)
    context.signal.throwIfAborted()
    return await deliverWebhook({
      destination,
      body: {
        sourceEventId: job.eventId,
        sourceEventType: job.eventType,
        payload: job.payload,
      },
      idempotencyKey: `${job.operationKey}:delivery`,
    }, context.signal)
  }
}

export function startTenantWorker(input: {
  tenantId: string
  store: JobStore
  destinationForTenant(tenantId: string): Promise<URL>
}): JobWorker {
  const policy: RetryPolicy = {
    maxAttempts: 6,
    baseDelayMs: 1_000,
    maxDelayMs: 60_000,
    leaseMs: 30_000,
    heartbeatMs: 10_000,
    attemptTimeoutMs: 20_000,
  }
  return new JobWorker(
    input.tenantId,
    `worker-${process.pid}-${randomUUID()}`,
    input.store,
    createProcessor(input.destinationForTenant),
    policy,
    4,
  )
}

export function webhookRoute(input: {
  secret: string
  store: JobStore
  resolveTenant: TenantResolver
}): (request: Request) => Promise<Response> {
  return createWebhookHandler(input)
}