import { Hash, Hex } from 'ox'

import type * as app_Webhooks from '../apps/data/routes/webhooks.js'
import type * as Db from '../db/Db.js'
import * as WebhookDeliveries from '../db/tables/webhookDeliveries.js'
import * as WebhookQueueCompletions from '../db/tables/webhookQueueCompletions.js'
import * as WebhookQueueEvents from '../db/tables/webhookQueueEvents.js'
import * as WebhookSubscriptions from '../db/tables/webhookSubscriptions.js'
import type * as Metrics from '../Metrics.js'
import * as Cursor from './Cursor.js'
import * as Id from './Id.js'
import * as MetricSink from './MetricSink.js'
import * as WebhookDestination from './WebhookDestination.js'

// The destination layer lives in `./WebhookDestination.js`. Re-export only the
// pieces that belong on the public `Webhooks.*` surface: the `Destination` and
// `Result` types, the `InvalidUrlError` thrown by subscription writes, and the
// signature-verification API a consumer uses to authenticate a delivery
// (`verify` + `signatureHeader`, with `sign` as its pair). The transport-specific
// validators/log helpers stay internal to `WebhookDestination`.
export type { Destination, Result } from './WebhookDestination.js'
export { InvalidUrlError, sign, signatureHeader, verify } from './WebhookDestination.js'

/**
 * Event types a subscription can listen to. The canonical set lives in the apps
 * layer ({@link app_Webhooks.eventTypes}), co-located with the per-type schemas
 * and descriptions that reference apps-layer read schemas; this storage layer
 * only consumes the type.
 */
export type EventType = app_Webhooks.EventType

/** Lifecycle status of a subscription. */
export type Status = 'active' | 'disabled' | 'paused'

/**
 * Authenticated owner of a subscription. Derived from the request principal:
 * API keys are org-scoped; MPP payers are identified by their stable DID.
 */
export type Owner =
  | {
      /** Owning organization id (any key in the org manages its webhooks). */
      orgId: string
      /** Owner kind. */
      type: 'api_key'
    }
  | {
      /** Verified MPP payer identifier (e.g. `did:pkh:…`). */
      payer: string
      /** Owner kind. */
      type: 'mpp'
    }

/** API-key scope applied when managing private-resource subscriptions. */
export type Access = {
  /** API-key environment. */
  environment: 'production' | 'sandbox'
  /** Project attribution, when the key is project-scoped. */
  projectId?: string | undefined
  /** API-key scopes used to authorize private resource reads. */
  scopes: readonly string[]
}

/**
 * Optional human context describing what a subscription is for. Carried onto
 * every delivered {@link Envelope} and surfaced by destinations (Slack header +
 * description, Better Stack log fields, and the raw `url` payload).
 */
export type Context = {
  /** Longer description of the subscription's purpose. */
  description?: string | undefined
  /** Arbitrary key/value labels echoed on every delivery and rendered by destinations. */
  metadata?: Record<string, string> | undefined
  /** Short label for the subscription (e.g. `Prod USDC large transfers`). */
  title?: string | undefined
}

/** A durable webhook subscription record. */
export type Subscription = {
  /** Chain the subscription listens on. */
  chainId: number
  /** Optional human context describing what the subscription is for. */
  context?: Context | undefined
  /** ISO timestamp of creation. */
  createdAt: string
  /** Where matched events are delivered (URL or Slack channel). */
  destination: WebhookDestination.Destination
  /** API-key environment captured when the subscription was created. */
  environment?: 'production' | 'sandbox' | undefined
  /** Event type the subscription listens to. */
  eventType: EventType
  /** ISO expiry timestamp; set for MPP-owned subscriptions (TTL-bounded). */
  expiresAt?: string | undefined
  /** Consecutive delivery failures; drives auto-disable. */
  failureCount: number
  /** Event-type-specific filter predicates. */
  filters: Record<string, unknown>
  /** Subscription id (`wh_…`). */
  id: string
  /** ISO timestamp of the last successful delivery, if any. Display only:
   * refreshed at most once per precision window. */
  lastDeliveryAt?: string | undefined
  /** Authenticated owner. */
  owner: Owner
  /** Project attribution captured from the creating API key. */
  projectId?: string | undefined
  /** HMAC signing secret. Returned once at creation; never exposed by reads. */
  secret: string
  /** Lifecycle status. */
  status: Status
  /** ISO timestamp of the last mutation. May lag repeat deliveries; see
   * `lastDeliveryAt`. */
  updatedAt: string
}

/** Fields accepted when creating a subscription. */
export type CreateInput = {
  /** Chain to listen on. */
  chainId: number
  /** Optional human context describing what the subscription is for. */
  context?: Context | undefined
  /** Where matched events are delivered. `url` destinations are SSRF-validated. */
  destination: WebhookDestination.Destination
  /** API-key environment used to isolate private resource events. */
  environment?: 'production' | 'sandbox' | undefined
  /** Event type to listen to. */
  eventType: EventType
  /** Event-type-specific filter predicates. */
  filters?: Record<string, unknown> | undefined
  /** Owner derived from the request principal. */
  owner: Owner
  /** Project attribution used to isolate private resource events. */
  projectId?: string | undefined
  /** Time-to-live in milliseconds; set the subscription to expire (MPP TTL). */
  ttl?: number | undefined
}

/** Mutable fields accepted when patching a subscription. */
export type PatchInput = {
  /** New human context, or `null` to clear it. */
  context?: Context | null | undefined
  /** New delivery destination. `url` destinations are re-validated for SSRF. */
  destination?: WebhookDestination.Destination | undefined
  /** New filter predicates. */
  filters?: Record<string, unknown> | undefined
  /** New lifecycle status (`active`/`paused`/`disabled`). */
  status?: Status | undefined
}

/** Outcome status of a delivery-log row. */
export type DeliveryStatus = 'failed' | 'pending' | 'succeeded'

/**
 * A durable delivery-log row recording one delivery attempt. Persisted by
 * {@link deliverAndRecord} so deliveries are observable and replayable (Phase
 * C); the read endpoint returns these verbatim.
 */
export type Delivery = {
  /** Consecutive attempt number (1-based) at the time of this delivery. */
  attempt: number
  /** ISO timestamp the attempt was recorded. */
  createdAt: string
  /**
   * The exact envelope that was sent, persisted so the delivery can be replayed
   * verbatim (manual retry). Internal-only: the public delivery row omits it.
   */
  envelope: Envelope
  /** Failure reason when `status` is `failed`. */
  error?: string | undefined
  /** Idempotent event id this delivery carried (`evt_…`). */
  eventId: string
  /** Delivery id (`whd_…`); time-sortable so keys order chronologically. */
  id: string
  /** Subscriber callback URL the attempt targeted. */
  requestUrl: string
  /** Wall-clock duration of the attempt in ms, when a request was made. */
  responseMs?: number | undefined
  /** HTTP response status, when a response was received. */
  responseStatus?: number | undefined
  /** Outcome status. */
  status: DeliveryStatus
  /** Owning subscription id. */
  subscriptionId: string
}

/**
 * Creates a subscription, validating its URL and (optionally) enforcing a
 * per-owner cap, then persists it. Capped writes serialize per owner so
 * concurrent requests cannot exceed the limit.
 */
export async function createSubscription(
  db: Db.Db,
  input: CreateInput,
  options: createSubscription.Options = {},
): Promise<Subscription> {
  WebhookDestination.assertDestination(input.destination)
  if (
    options.startBlockNumber !== undefined &&
    (!Number.isSafeInteger(options.startBlockNumber) || options.startBlockNumber < 0)
  )
    throw new RangeError('startBlockNumber must be a non-negative safe integer.')

  const now = (options.now ?? (() => new Date()))()
  const iso = now.toISOString()
  // A one-field checkpoint records the live creation head without claiming it
  // was scanned; it anchors the fanout boundary for the new subscription.
  const pollerCursor =
    options.startBlockNumber === undefined ? undefined : Cursor.encode([options.startBlockNumber])

  const subscription: Subscription = {
    chainId: input.chainId,
    ...(input.context === undefined ? {} : { context: input.context }),
    createdAt: iso,
    destination: input.destination,
    ...(input.environment === undefined ? {} : { environment: input.environment }),
    eventType: input.eventType,
    failureCount: 0,
    filters: input.filters ?? {},
    id: generateId(now),
    owner: input.owner,
    ...(input.projectId === undefined ? {} : { projectId: input.projectId }),
    secret: generateSecret(),
    status: 'active',
    updatedAt: iso,
    ...(input.ttl === undefined
      ? {}
      : { expiresAt: new Date(now.getTime() + input.ttl).toISOString() }),
  }

  if (options.maxPerOwner !== undefined) {
    const inserted = await WebhookSubscriptions.insertWithinLimit(db, subscription, {
      maxPerOwner: options.maxPerOwner,
      now: iso,
      ...(pollerCursor === undefined ? {} : { pollerCursor }),
    })
    if (!inserted) throw new LimitExceededError(options.maxPerOwner)
  } else
    await WebhookSubscriptions.insert(
      db,
      subscription,
      pollerCursor === undefined ? {} : { pollerCursor },
    )
  return subscription
}

export declare namespace createSubscription {
  /** Options for {@link createSubscription}. */
  type Options = {
    /** Reject creation once the owner has this many live subscriptions. */
    maxPerOwner?: number | undefined
    /** Clock used for timestamps (injectable for tests). */
    now?: (() => Date) | undefined
    /** Chain head observed immediately before the subscription becomes visible. */
    startBlockNumber?: number | undefined
  }
}

/** Reads a single subscription scoped to its owner. Returns null when absent. */
export async function getSubscription(
  db: Db.Db,
  owner: Owner,
  id: string,
  options: getSubscription.Options = {},
): Promise<Subscription | null> {
  return WebhookSubscriptions.get(db, owner, id, new Date().toISOString(), options.access)
}

export declare namespace getSubscription {
  /** Private-resource visibility options. */
  type Options = { access?: Access | undefined }
}

/**
 * Lists an owner's subscriptions, newest first, with optional keyset paging.
 * Subscription ids embed a timestamp, so lexical id order is chronological.
 */
export async function listSubscriptions(
  db: Db.Db,
  owner: Owner,
  options: listSubscriptions.Options = {},
): Promise<Subscription[]> {
  return WebhookSubscriptions.list(db, owner, { ...options, now: new Date().toISOString() })
}

export declare namespace listSubscriptions {
  /** Options for {@link listSubscriptions}. */
  type Options = {
    /** Private-resource visibility scope. */
    access?: Access | undefined
    /** Return subscriptions older than this id (keyset paging, newest first). */
    cursor?: string | undefined
    /** Maximum subscriptions to return. */
    limit?: number | undefined
    /** Rows to skip from the head (positional pagination; exclusive with `cursor`). */
    offset?: number | undefined
  }
}

/**
 * Counts an owner's subscriptions — a single indexed count, exact and cheap,
 * since the per-owner set is bounded by `maxPerOwner`. Feeds the opt-in
 * `meta.totalCount` on `GET /webhooks`.
 */
export async function countSubscriptions(
  db: Db.Db,
  owner: Owner,
  options: countSubscriptions.Options = {},
): Promise<number> {
  return WebhookSubscriptions.count(db, owner, new Date().toISOString(), options.access)
}

export declare namespace countSubscriptions {
  /** Private-resource visibility options. */
  type Options = { access?: Access | undefined }
}

/**
 * Patches an owner's subscription, re-validating the URL when changed.
 * Returns null when absent. The status column drives the partial active
 * index, so no secondary-index maintenance is needed.
 */
export async function updateSubscription(
  db: Db.Db,
  owner: Owner,
  id: string,
  patch: PatchInput,
  options: updateSubscription.Options = {},
): Promise<Subscription | null> {
  const current = await getSubscription(
    db,
    owner,
    id,
    options.access === undefined ? {} : { access: options.access },
  )
  if (!current) return null

  if (current.destination.type !== 'url' && patch.destination?.type === 'url')
    throw new InvalidDestinationTransitionError()
  if (patch.destination !== undefined) WebhookDestination.assertDestination(patch.destination)

  const now = (options.now ?? (() => new Date()))()
  const next = await WebhookSubscriptions.update(db, id, {
    // `context: null` clears it (stored as a null column; drops from reads).
    ...(patch.context === undefined ? {} : { context: patch.context }),
    ...(patch.destination === undefined ? {} : { destination: patch.destination }),
    ...(patch.filters === undefined ? {} : { filters: patch.filters }),
    ...(patch.status === undefined ? {} : { status: patch.status }),
    updatedAt: now.toISOString(),
  })
  return next ?? null
}

export declare namespace updateSubscription {
  /** Options for {@link updateSubscription}. */
  type Options = {
    /** Private-resource visibility scope. */
    access?: Access | undefined
    /** Clock used for timestamps (injectable for tests). */
    now?: (() => Date) | undefined
  }
}

/** Deletes an owner's subscription; its delivery rows cascade via the FK. */
export async function deleteSubscription(
  db: Db.Db,
  owner: Owner,
  id: string,
  options: deleteSubscription.Options = {},
): Promise<boolean> {
  return WebhookSubscriptions.remove(db, owner, id, new Date().toISOString(), options.access)
}

export declare namespace deleteSubscription {
  /** Private-resource visibility options. */
  type Options = { access?: Access | undefined }
}

/** Reads every live active subscription on a chain in one query. */
export async function listActiveForChain(
  db: Db.Db,
  options: listActiveForChain.Options,
): Promise<{ cursor: string | null; subscription: Subscription }[]> {
  return WebhookSubscriptions.listActiveForChain(db, {
    chainId: options.chainId,
    eventTypes: options.eventTypes,
    now: new Date().toISOString(),
  })
}

export declare namespace listActiveForChain {
  /** Filters for the chain-level subscription read. */
  type Options = Pick<WebhookSubscriptions.listActiveForChain.Options, 'chainId' | 'eventTypes'>
}

/** Decodes the block component of a subscription cursor, if any. */
export function cursorBlock(cursor: string | null): number | undefined {
  if (cursor === null) return undefined
  const decoded = Cursor.decode(cursor, ['int', 'int']) ?? Cursor.decode(cursor, ['int'])
  const block = decoded?.[0]
  return typeof block === 'number' ? block : undefined
}

/** Reads a subscription's keyset cursor. Returns null when unset. */
export async function getCursor(db: Db.Db, subscriptionId: string): Promise<string | null> {
  return WebhookSubscriptions.getCursor(db, subscriptionId)
}

/**
 * Deletes expired MPP-owned subscriptions (delivery rows cascade) and
 * deliveries past retention. Reads already filter expiry at query time; this
 * is the janitor reclaiming space, run best-effort on a schedule.
 */
export async function pruneExpired(
  db: Db.Db,
  options: pruneExpired.Options = {},
): Promise<pruneExpired.Result> {
  const iso = (options.now ?? (() => new Date()))().toISOString()
  // Sequential on purpose: the subscription delete cascades into the other
  // tables, so overlapping it with their prunes risks deadlocks.
  return {
    completions: await WebhookQueueCompletions.pruneExpired(db, iso),
    deliveries: await WebhookDeliveries.pruneExpired(db, iso),
    subscriptions: await WebhookSubscriptions.pruneExpired(db, iso),
  }
}

export declare namespace pruneExpired {
  /** Options for {@link pruneExpired}. */
  type Options = {
    /** Clock used for expiry evaluation; injectable for tests. */
    now?: (() => Date) | undefined
  }

  /** Rows deleted per table; a count pinned at the batch bound means expiry is outpacing pruning. */
  type Result = {
    /** Deleted `webhook_queue_completions` rows. */
    completions: number
    /** Deleted `webhook_deliveries` rows. */
    deliveries: number
    /** Deleted `webhook_subscriptions` rows. */
    subscriptions: number
  }
}

/**
 * Default delivery-log retention. Rows are written with this TTL so a busy
 * subscription's log doesn't grow unbounded; tune per deployment.
 */
export const deliveryRetentionMs = 7 * 24 * 60 * 60 * 1_000

/**
 * Completions dedupe window: replays of a completed obligation are suppressed
 * this long. Deliberately its own literal so per-deployment log-retention
 * tuning cannot move it.
 */
export const dedupeRetentionMs = 7 * 24 * 60 * 60 * 1_000

/** A compact Cloudflare Queue message referencing a durable staged envelope. */
export type QueueReference = {
  /** Idempotent event id of the staged envelope. */
  eventId: string
  /** Subscription that owns the staged envelope. */
  subscriptionId: string
}

/** A full envelope staged outside Cloudflare Queues. */
export type QueueEvent = QueueReference & {
  /** ISO timestamp when the envelope was staged. */
  createdAt: string
  /** Full envelope loaded by the Queue consumer. */
  envelope: Envelope
}

/** A claim newer than this counts as an in-flight attempt; above the 10s destination timeout default. */
export const deliveryAttemptStaleMs = 60_000

/**
 * Retry backoff by attempt count: 5s, 25s, ~2m, ~10m, ~52m, then hourly.
 * Endpoint outages retry on this schedule until delivery or auto-disable.
 */
export function retryDelayMs(attemptCount: number): number {
  return Math.min(5_000 * 5 ** Math.max(0, attemptCount - 1), 3_600_000)
}

/**
 * Stages full envelopes as delivery obligations and returns compact
 * references safe for Queue admission. Existing rows are left untouched, so
 * the first staged envelope wins and terminal outcomes never resurrect.
 */
export async function ensureQueueEvents(
  db: Db.Db,
  dispatchables: readonly ensureQueueEvents.Dispatchable[],
  options: ensureQueueEvents.Options = {},
): Promise<ensureQueueEvents.Result> {
  const now = (options.now ?? (() => new Date()))()
  // Carried so the delivery worker can measure observation-to-attempt without
  // re-reading the block; replays have no observation and stay null.
  const observedAt =
    options.observedAt === undefined ? null : new Date(options.observedAt).toISOString()
  const events = dispatchables.map(({ envelope, subscription }) => ({
    createdAt: now.toISOString(),
    envelope,
    eventId: envelope.id,
    // Due immediately; the value orders the sweep so it is never null here.
    nextAttemptAt: now.toISOString(),
    observedAt,
    subscriptionId: subscription.id,
  }))
  return {
    created: await WebhookQueueEvents.ensure(db, events),
    references: events.map(({ eventId, subscriptionId }) => ({ eventId, subscriptionId })),
  }
}

export declare namespace ensureQueueEvents {
  /** One full envelope and the subscription that produced it. */
  type Dispatchable = {
    /** Webhook envelope. */
    envelope: Envelope
    /** Subscription that owns the envelope. */
    subscription: Subscription
  }

  /** Staging options. */
  type Options = {
    /** Clock used for staging timestamps. */
    now?: (() => Date) | undefined
    /** Epoch ms the head owing these events was observed; absent on replays. */
    observedAt?: number | undefined
  }

  /** Created-row count plus references for every requested obligation. */
  type Result = {
    /** Newly staged obligations. */
    created: number
    /** Compact references for all inputs, including pre-existing rows. */
    references: QueueReference[]
  }
}

/** Claims one due pending obligation; misses classify why the claim is unavailable. */
export function claimQueueEvent(
  db: Db.Db,
  reference: QueueReference,
): Promise<WebhookQueueEvents.claim.Result> {
  const now = new Date()
  return WebhookQueueEvents.claim(db, reference, {
    now: now.toISOString(),
    staleBefore: new Date(now.getTime() - deliveryAttemptStaleMs).toISOString(),
  })
}

/**
 * Marks a claimed obligation terminal and restarts its retention window, so
 * the dedupe row outlives replays. Fenced by the claim marker: a stale
 * claimant's completion returns false instead of clobbering a newer claim.
 */
export function completeQueueEvent(
  db: Db.Db,
  reference: QueueReference,
  options: completeQueueEvent.Options,
): Promise<boolean> {
  return WebhookQueueEvents.complete(db, reference, {
    ...options,
    expiresAt: new Date(Date.now() + dedupeRetentionMs).toISOString(),
  })
}

export declare namespace completeQueueEvent {
  /** Fenced terminal transition. */
  type Options = {
    /** Claim marker from the claimed record's `attemptingAt`. */
    claimedAt: string
    /** Terminal obligation outcome. */
    status: WebhookQueueCompletions.TerminalStatus
  }
}

/**
 * Releases a claim and schedules the next attempt on the retry backoff.
 * Fenced by the claim marker like {@link completeQueueEvent}.
 */
export function scheduleQueueEventRetry(
  db: Db.Db,
  reference: QueueReference,
  options: scheduleQueueEventRetry.Options,
): Promise<boolean> {
  return WebhookQueueEvents.schedule(db, reference, {
    claimedAt: options.claimedAt,
    nextAttemptAt: new Date(Date.now() + retryDelayMs(options.attemptCount)).toISOString(),
  })
}

export declare namespace scheduleQueueEventRetry {
  /** Fenced retry scheduling. */
  type Options = {
    /** Attempts claimed so far; drives the backoff position. */
    attemptCount: number
    /** Claim marker from the claimed record's `attemptingAt`. */
    claimedAt: string
  }
}

/** Lists due pending references for the sweeper, oldest first. */
export function dueQueueEvents(db: Db.Db, limit: number): Promise<QueueReference[]> {
  const now = new Date()
  return WebhookQueueEvents.due(db, {
    limit,
    now: now.toISOString(),
    staleBefore: new Date(now.getTime() - deliveryAttemptStaleMs).toISOString(),
  })
}

/** Returns the authoritative pending-obligation snapshot. */
export function pendingQueueEvents(db: Db.Db): Promise<WebhookQueueEvents.pending.Result> {
  const now = new Date()
  return WebhookQueueEvents.pending(db, {
    now: now.toISOString(),
    staleBefore: new Date(now.getTime() - deliveryAttemptStaleMs).toISOString(),
  })
}

/** Reads a full staged envelope for a compact Queue reference. */
export async function getQueueEvent(
  db: Db.Db,
  reference: QueueReference,
): Promise<QueueEvent | null> {
  return WebhookQueueEvents.get(db, reference)
}

/** Loads a staged envelope and delivers it with the subscription's current configuration. */
export async function deliverQueueEventAndRecord(
  db: Db.Db,
  reference: QueueReference,
  options: deliverAndRecord.Options = {},
): Promise<deliverQueueEventAndRecord.Result> {
  const dequeuedAt = Date.now()
  const event = await getQueueEvent(db, reference)
  if (!event) return { status: 'missing' }
  return deliverClaimedAndRecord(db, event, { ...options, dequeuedAt })
}

/**
 * Delivers an already-loaded obligation envelope with the subscription's
 * current configuration, skipping the staged-row read a claim already paid.
 */
export async function deliverClaimedAndRecord(
  db: Db.Db,
  claimed: deliverClaimedAndRecord.Claimed,
  options: deliverClaimedAndRecord.Options = {},
): Promise<deliverClaimedAndRecord.Result> {
  const dequeuedAt = options.dequeuedAt ?? Date.now()
  const subscriptionReadStartedAt = Date.now()
  const subscription = await WebhookSubscriptions.getById(
    db,
    claimed.subscriptionId,
    new Date().toISOString(),
  )
  if (!subscription || subscription.status !== 'active') return { status: 'skipped' }
  // A staged row predating the observation column, or a replay, reads null.
  const observedAt = claimed.observedAt ? Date.parse(claimed.observedAt) : Number.NaN
  return {
    result: await deliverAndRecord(db, subscription, claimed.envelope, {
      ...options,
      dequeuedAt,
      ...(Number.isNaN(observedAt) ? {} : { observedAt }),
      subscriptionReadMs: Date.now() - subscriptionReadStartedAt,
    }),
    status: 'delivered',
  }
}

export declare namespace deliverClaimedAndRecord {
  /** Envelope and owning subscription id from a claimed obligation. */
  type Claimed = {
    /** Immutable event envelope to deliver. */
    envelope: Envelope
    /** ISO head-observation stamp carried by the staged row; null on replays. */
    observedAt?: string | null | undefined
    /** Subscription re-resolved for current status and secret. */
    subscriptionId: string
  }

  /** Delivery options plus the dequeue timestamp of the claiming path. */
  type Options = deliverAndRecord.Options & {
    /** Epoch ms the claiming message was dequeued; defaults to now. */
    dequeuedAt?: number | undefined
  }

  /** A delivered event or an inactive subscription. */
  type Result =
    | {
        /** Transport result from the current subscription state. */
        result: WebhookDestination.Result
        /** The envelope was delivered. */
        status: 'delivered'
      }
    | {
        /** The current subscription is inactive or deleted. */
        status: 'skipped'
      }
}

export declare namespace deliverQueueEventAndRecord {
  /** A delivered event, inactive subscription, or stale duplicate reference. */
  type Result =
    | {
        /** Transport result from the current subscription state. */
        result: WebhookDestination.Result
        /** The staged event was delivered. */
        status: 'delivered'
      }
    | {
        /** The staged event or current subscription was not deliverable. */
        status: 'missing' | 'skipped'
      }
}

/** Records one delivery-log row (stateful, owner-agnostic; scoped per subscription). */
export async function recordDelivery(
  db: Db.Db,
  delivery: Delivery,
  options: recordDelivery.Options = {},
): Promise<void> {
  const ttl = options.ttl ?? deliveryRetentionMs
  // The retention deadline replaces the old store TTL: reads filter on it at
  // query time and pruning reclaims past it. Anchored to write time (not the
  // row's `createdAt`, which tests pin) — exactly like the old `put` TTL.
  const expiresAt = new Date(Date.now() + ttl).toISOString()
  await WebhookDeliveries.insert(db, delivery, expiresAt)
}

export declare namespace recordDelivery {
  /** Options for {@link recordDelivery}. */
  type Options = {
    /** Row time-to-live in ms; defaults to {@link deliveryRetentionMs}. */
    ttl?: number | undefined
  }
}

/**
 * Lists a subscription's delivery log, newest first, with optional keyset
 * paging. Delivery ids embed a timestamp, so lexical key order is chronological.
 */
export async function listDeliveries(
  db: Db.Db,
  subscriptionId: string,
  options: listDeliveries.Options = {},
): Promise<Delivery[]> {
  return WebhookDeliveries.list(db, subscriptionId, { ...options, now: new Date().toISOString() })
}

export declare namespace listDeliveries {
  /** Options for {@link listDeliveries}. */
  type Options = {
    /** Return deliveries whose id sorts before this one (older entries). */
    cursor?: string | undefined
    /** Maximum deliveries to return. */
    limit?: number | undefined
    /** Rows to skip from the head (positional pagination; exclusive with `cursor`). */
    offset?: number | undefined
  }
}

/**
 * Counts a subscription's delivery-log entries — a single indexed count,
 * exact, and bounded by the delivery retention window. Feeds the opt-in
 * `meta.totalCount` on `GET /webhooks/:id/deliveries`.
 */
export async function countDeliveries(db: Db.Db, subscriptionId: string): Promise<number> {
  return WebhookDeliveries.count(db, subscriptionId, new Date().toISOString())
}

/** Loads a single delivery-log row (`null` when absent), scoped to its subscription. */
export async function getDelivery(
  db: Db.Db,
  subscriptionId: string,
  deliveryId: string,
): Promise<Delivery | null> {
  return WebhookDeliveries.get(db, subscriptionId, deliveryId, new Date().toISOString())
}

/**
 * Default number of consecutive delivery failures after which a subscription
 * auto-disables (and is dropped from the partial active index).
 */
export const maxFailures = 10

/**
 * How stale `lastDeliveryAt` may get before a success rewrites it. Read only
 * for display, so exactness is traded for not rewriting the row per delivery.
 */
const lastDeliveryPrecisionMs = 60_000

/**
 * Records a successful delivery: resets the consecutive-failure counter and
 * refreshes `lastDeliveryAt` at most once per precision window. Returns the
 * written row, or the caller's snapshot when the write was skipped or the row
 * deleted.
 */
export async function recordSuccess(
  db: Db.Db,
  subscription: Subscription,
  options: recordSuccess.Options = {},
): Promise<Subscription> {
  const now = (options.now ?? (() => new Date()))()
  const next = await WebhookSubscriptions.recordSuccess(db, subscription.id, {
    now: now.toISOString(),
    staleBefore: new Date(now.getTime() - lastDeliveryPrecisionMs).toISOString(),
  })
  // No row means the write was skipped or the row was deleted; neither
  // resurrects, so keep the caller's timestamps.
  return next ?? { ...subscription, failureCount: 0 }
}

export declare namespace recordSuccess {
  /** Options for {@link recordSuccess}. */
  type Options = {
    /** Clock used for timestamps; injectable for tests. */
    now?: (() => Date) | undefined
  }
}

/**
 * Records a failed delivery: atomically increments the consecutive-failure
 * counter and, once it reaches `maxFailures`, sets `status: 'disabled'` so the
 * subscription drops out of the partial active index (surfaced via the API).
 * Once disabled at the cap the row is left alone, so an outage's in-flight
 * failures stop contending on its lock. Returns the written row, or the
 * caller's snapshot advanced optimistically.
 */
export async function recordFailure(
  db: Db.Db,
  subscription: Subscription,
  options: recordFailure.Options = {},
): Promise<Subscription> {
  const now = (options.now ?? (() => new Date()))()
  const iso = now.toISOString()
  const max = options.maxFailures ?? maxFailures
  const next = await WebhookSubscriptions.recordFailure(db, subscription.id, max, iso)
  if (next) return next
  // No row means skipped (already disabled at the cap) or deleted; neither
  // resurrects, so keep the caller's timestamps. Capped so chained skips
  // cannot grow the counter past what the store would hold.
  const failureCount = Math.min(subscription.failureCount + 1, max)
  return {
    ...subscription,
    failureCount,
    ...(failureCount >= max ? { status: 'disabled' as const } : {}),
  }
}

export declare namespace recordFailure {
  /** Options for {@link recordFailure}. */
  type Options = {
    /** Consecutive failures before auto-disabling (default {@link maxFailures}). */
    maxFailures?: number | undefined
    /** Clock used for timestamps; injectable for tests. */
    now?: (() => Date) | undefined
  }
}

/**
 * Delivers an envelope to its subscription and records the outcome: success
 * resets the failure counter and stamps `lastDeliveryAt`; failure increments it
 * and auto-disables after {@link maxFailures}. This is the single delivery entry
 * point shared by the Cloudflare Queue consumer and the self-host inline path.
 */
export async function deliverAndRecord(
  db: Db.Db,
  subscription: Subscription,
  envelope: Envelope,
  options: deliverAndRecord.Options = {},
): Promise<WebhookDestination.Result> {
  const startedAt = Date.now()
  // Snapshot the attempt before recordSuccess/recordFailure mutate the counter.
  const attempt = options.attempt ?? subscription.failureCount + 1
  const result = await (async () => {
    try {
      return await WebhookDestination.from(subscription.destination).deliver({
        envelope,
        secret: subscription.secret,
        ...(options.fetch === undefined ? {} : { fetch: options.fetch }),
        ...(options.timeoutMs === undefined ? {} : { timeoutMs: options.timeoutMs }),
      })
    } catch {
      // A formatting failure is a failed attempt, not an exception that bypasses
      // subscription failure state and delivery metrics on every queue retry.
      return { error: 'delivery failed before request', ok: false }
    }
  })()
  // Overlapped: the two writes touch different rows and the log carries the
  // immutable pre-attempt subscription, so it never reads the state write's
  // result. Sequentially they cost two Postgres round trips per delivery.
  const writeStartedAt = Date.now()
  // Append a delivery-log row. Best-effort: a logging failure must never change
  // the delivery outcome the caller (Queue consumer / inline path) acts on.
  const deliveryLog = (async () => {
    try {
      await recordDelivery(
        db,
        buildDelivery({ attempt, envelope, result, subscription }, options),
        options.deliveryRetentionMs === undefined ? {} : { ttl: options.deliveryRetentionMs },
      )
    } catch {
      // delivery logging is best-effort
    }
    return Date.now() - writeStartedAt
  })()
  const stateWrite = (async () => {
    if (result.ok) await recordSuccess(db, subscription, options)
    else await recordFailure(db, subscription, options)
    return Date.now() - writeStartedAt
  })()
  // A state-write rejection still propagates to the caller's retry path; the
  // log promise never rejects, so it cannot mask it or go unhandled.
  const [stateWriteMs, deliveryLogMs] = await Promise.all([stateWrite, deliveryLog])
  const completedAt = Date.now()

  // Emit after all delivery work so the timings cover every stage. The
  // subscription is immutable, preserving the pre-attempt failure count.
  const metrics = options.metrics ? MetricSink.webhooks(options.metrics) : undefined
  if (metrics) {
    const eventAt = eventTimestamp(envelope)
    const envelopeAt = Date.parse(envelope.createdAt)
    const dequeuedAt = options.dequeuedAt ?? startedAt
    metrics.record({
      envelope,
      queueAttempt: options.queueAttempt,
      result,
      subscription,
      attempt,
      timings: {
        deliveryLogMs,
        envelopeToQueueMs:
          options.queuedAt === undefined || Number.isNaN(envelopeAt)
            ? undefined
            : Math.max(0, options.queuedAt - envelopeAt),
        endToEndMs: eventAt === undefined ? undefined : Math.max(0, completedAt - eventAt),
        // The attempt leaving, not bookkeeping finishing: everything after
        // `startedAt` is either the subscriber's time or our own recording,
        // neither of which delays the event reaching them. Gated on a request
        // actually being made, so a failure before send (formatting, invalid
        // destination) does not report an attempt that never left.
        eventToAttemptMs:
          eventAt === undefined || result.durationMs === undefined
            ? undefined
            : Math.max(0, startedAt - eventAt),
        eventToEnvelopeMs:
          eventAt === undefined || Number.isNaN(envelopeAt)
            ? undefined
            : Math.max(0, envelopeAt - eventAt),
        observedToAttemptMs:
          options.observedAt === undefined || result.durationMs === undefined
            ? undefined
            : Math.max(0, startedAt - options.observedAt),
        preflightMs: Math.max(0, startedAt - dequeuedAt),
        processingMs: Math.max(0, completedAt - dequeuedAt),
        queueWaitMs:
          options.queuedAt === undefined ? undefined : Math.max(0, dequeuedAt - options.queuedAt),
        settleMs: Math.max(0, completedAt - writeStartedAt),
        stateWriteMs,
        subscriptionReadMs: options.subscriptionReadMs,
      },
      trigger: options.trigger ?? 'queue',
      type: 'webhook:delivery',
    })
    metrics.flush()
  }

  return result
}

/** Reads an event timestamp from an envelope, falling back to its creation time. */
function eventTimestamp(envelope: Envelope): number | undefined {
  const value =
    typeof envelope.data === 'object' &&
    envelope.data !== null &&
    'timestamp' in envelope.data &&
    typeof envelope.data.timestamp === 'string'
      ? envelope.data.timestamp
      : envelope.createdAt
  const timestamp = Date.parse(value)
  return Number.isNaN(timestamp) ? undefined : timestamp
}

export declare namespace deliverAndRecord {
  /** Options for {@link deliverAndRecord}. */
  type Options = {
    /**
     * Delivery-log attempt number; the ledger claim count on the queue path.
     * Defaults to the subscription failure counter, which concurrent
     * deliveries can read stale.
     */
    attempt?: number | undefined
    /** Delivery-log row TTL in ms; defaults to {@link deliveryRetentionMs}. */
    deliveryRetentionMs?: number | undefined
    /** Queue dequeue time as epoch milliseconds, captured before subscription lookup. */
    dequeuedAt?: number | undefined
    /** `fetch` implementation passed through to {@link deliver}. */
    fetch?: typeof globalThis.fetch | undefined
    /** Consecutive failures before auto-disabling (default {@link maxFailures}). */
    maxFailures?: number | undefined
    /** Metrics backend; emits one `webhook:delivery` event per attempt. */
    metrics?: Metrics.Metrics | undefined
    /** Clock used for timestamps; injectable for tests. */
    now?: (() => Date) | undefined
    /** Epoch ms the head owing this event was observed; absent on replays. */
    observedAt?: number | undefined
    /** Cloudflare Queue delivery attempt, starting at one; absent for inline delivery paths. */
    queueAttempt?: number | undefined
    /** Queue enqueue time as epoch milliseconds, used to isolate queue wait. */
    queuedAt?: number | undefined
    /** Authoritative subscription lookup duration, supplied by the queue path. */
    subscriptionReadMs?: number | undefined
    /** Delivery timeout in ms passed through to {@link deliver}. */
    timeoutMs?: number | undefined
    /** Where this attempt originated; tags the `webhook:delivery` event (default `queue`). */
    trigger?: MetricSink.Trigger | undefined
  }
}

/**
 * Re-resolves an enqueued subscription before delivery. Missing or inactive
 * subscriptions are skipped; delivery uses the current destination and secret.
 */
export async function deliverCurrentAndRecord(
  db: Db.Db,
  queued: Subscription,
  envelope: Envelope,
  options: deliverAndRecord.Options = {},
): Promise<deliverCurrentAndRecord.Result> {
  const dequeuedAt = Date.now()
  const subscriptionReadStartedAt = dequeuedAt
  const subscription = await getSubscription(db, queued.owner, queued.id)
  if (!subscription || subscription.status !== 'active') return { status: 'skipped' }
  return {
    result: await deliverAndRecord(db, subscription, envelope, {
      ...options,
      dequeuedAt,
      subscriptionReadMs: Date.now() - subscriptionReadStartedAt,
    }),
    status: 'delivered',
  }
}

export declare namespace deliverCurrentAndRecord {
  /** Authoritative delivery outcome, or a skipped stale queue message. */
  type Result =
    | {
        /** Transport result from the current subscription state. */
        result: WebhookDestination.Result
        /** The current subscription was delivered. */
        status: 'delivered'
      }
    | {
        /** The queued subscription is absent or inactive. */
        status: 'skipped'
      }
}

/** Builds a delivery-log row from an attempt's envelope and classified result. */
function buildDelivery(
  input: {
    attempt: number
    envelope: Envelope
    result: WebhookDestination.Result
    subscription: Subscription
  },
  options: { now?: (() => Date) | undefined } = {},
): Delivery {
  const { attempt, envelope, result, subscription } = input
  const now = (options.now ?? (() => new Date()))()
  return {
    attempt,
    createdAt: now.toISOString(),
    envelope,
    eventId: envelope.id,
    id: generateDeliveryId(now),
    requestUrl: WebhookDestination.destinationLabel(subscription.destination),
    status: result.ok ? 'succeeded' : 'failed',
    subscriptionId: subscription.id,
    ...(result.error === undefined ? {} : { error: result.error }),
    ...(result.durationMs === undefined ? {} : { responseMs: result.durationMs }),
    ...(result.status === undefined ? {} : { responseStatus: result.status }),
  }
}

/**
 * Computes a stable, idempotent event id from its on-chain coordinates so
 * re-deliveries are dedupable by the receiver.
 */
export function eventId(options: eventId.Options): string {
  return eventIdFromKey({
    chainId: options.chainId,
    eventType: options.eventType,
    key: `${options.blockNumber}:${options.logIndex}`,
  })
}

export declare namespace eventId {
  /** Options for {@link eventId}. */
  type Options = {
    /** Block number the event was indexed in. */
    blockNumber: bigint | number | string
    /** Chain the event occurred on. */
    chainId: number
    /** Event type. */
    eventType: EventType
    /** Log index within the block (or transaction index for transaction events). */
    logIndex: bigint | number | string
  }
}

/** Computes a stable event id from an event-type-specific identity key. */
export function eventIdFromKey(options: eventIdFromKey.Options): string {
  const input = `${options.chainId}:${options.eventType}:${options.key}`
  return `evt_${Hash.sha256(Hex.fromString(input)).slice(2)}`
}

export declare namespace eventIdFromKey {
  /** Stable identity inputs for a webhook event. */
  type Options = {
    /** Chain the event belongs to. */
    chainId: number
    /** Event type. */
    eventType: EventType
    /** Stable event-type-specific identity. */
    key: string
  }
}

/** The JSON envelope delivered to subscribers. */
export type Envelope = {
  /** Chain the event occurred on. */
  chainId: number
  /** Human context copied from the subscription, when set. */
  context?: Context | undefined
  /** ISO timestamp the envelope was built. */
  createdAt: string
  /** The same row shape the corresponding read endpoint returns. */
  data: unknown
  /** Stable, idempotent event id (`evt_…`). */
  id: string
  /** Originating subscription id. */
  subscriptionId: string
  /** Event type; `ping` marks a synthetic test delivery (see {@link buildPingEnvelope}). */
  type: EventType | 'ping'
}

/** Builds a delivery envelope for an event matched against a subscription. */
export function buildEnvelope(options: buildEnvelope.Options): Envelope {
  const { subscription } = options
  return {
    chainId: subscription.chainId,
    ...(subscription.context === undefined ? {} : { context: subscription.context }),
    createdAt: (options.createdAt ?? new Date()).toISOString(),
    data: options.data,
    id: eventId({
      blockNumber: options.blockNumber,
      chainId: subscription.chainId,
      eventType: subscription.eventType,
      logIndex: options.logIndex,
    }),
    subscriptionId: subscription.id,
    type: subscription.eventType,
  }
}

export declare namespace buildEnvelope {
  /** Options for {@link buildEnvelope}. */
  type Options = {
    /** Block number the event was indexed in. */
    blockNumber: bigint | number | string
    /** Timestamp to stamp on the envelope; defaults to now. */
    createdAt?: Date | undefined
    /** Decoded event row (same shape the read endpoint returns). */
    data: unknown
    /** Log index within the block. */
    logIndex: bigint | number | string
    /** Subscription the event matched. */
    subscription: Subscription
  }
}

/** Builds a delivery envelope from a stable application-event identity. */
export function buildKeyedEnvelope(options: buildKeyedEnvelope.Options): Envelope {
  const { subscription } = options
  return {
    chainId: subscription.chainId,
    ...(subscription.context === undefined ? {} : { context: subscription.context }),
    createdAt: (options.createdAt ?? new Date()).toISOString(),
    data: options.data,
    id: eventIdFromKey({
      chainId: subscription.chainId,
      eventType: subscription.eventType,
      key: options.key,
    }),
    subscriptionId: subscription.id,
    type: subscription.eventType,
  }
}

export declare namespace buildKeyedEnvelope {
  /** Application-event envelope fields. */
  type Options = {
    /** Timestamp to stamp on the envelope; defaults to now. */
    createdAt?: Date | undefined
    /** Public event payload. */
    data: unknown
    /** Stable event-type-specific identity. */
    key: string
    /** Subscription the event matched. */
    subscription: Subscription
  }
}

/**
 * Builds a synthetic `ping` envelope so an owner can test a subscription's
 * endpoint (connectivity, TLS, signature verification) without waiting for a real
 * on-chain event. The id is randomized per call so repeated pings are never
 * deduped by the receiver; `type` is `ping` and `data` is a recognizable
 * `{ ping: true }` marker rather than an event row, so receivers can tell a test
 * delivery apart from a real event.
 */
export function buildPingEnvelope(
  subscription: Subscription,
  options: buildPingEnvelope.Options = {},
): Envelope {
  const now = options.createdAt ?? new Date()
  const nonce = options.nonce ?? Hex.random(16).slice(2)
  return {
    chainId: subscription.chainId,
    ...(subscription.context === undefined ? {} : { context: subscription.context }),
    createdAt: now.toISOString(),
    data: { ping: true },
    id: `evt_${Hash.sha256(Hex.fromString(`ping:${subscription.id}:${nonce}`)).slice(2)}`,
    subscriptionId: subscription.id,
    type: 'ping',
  }
}

export declare namespace buildPingEnvelope {
  /** Options for {@link buildPingEnvelope}. */
  type Options = {
    /** Timestamp to stamp on the envelope; defaults to now. */
    createdAt?: Date | undefined
    /** Deterministic nonce for the event id; defaults to random (tests only). */
    nonce?: string | undefined
  }
}

/**
 * Delivers a synthetic {@link buildPingEnvelope} to a subscription's endpoint and
 * logs the attempt. Unlike {@link deliverAndRecord} it deliberately does NOT
 * touch the subscription's `failureCount`/`status`: a test ping must never
 * auto-disable a healthy subscription or reset a real failure streak. The
 * delivery row is still appended (best-effort) so the ping is observable via the
 * delivery log.
 */
export async function ping(
  db: Db.Db,
  subscription: Subscription,
  options: ping.Options = {},
): Promise<ping.Result> {
  const now = options.now ?? (() => new Date())
  const envelope = buildPingEnvelope(subscription, {
    createdAt: now(),
    ...(options.nonce === undefined ? {} : { nonce: options.nonce }),
  })
  const result = await WebhookDestination.from(subscription.destination).deliver({
    envelope,
    secret: subscription.secret,
    ...(options.fetch === undefined ? {} : { fetch: options.fetch }),
    ...(options.timeoutMs === undefined ? {} : { timeoutMs: options.timeoutMs }),
  })

  // Best-effort: a logging failure must never change the result the caller sees.
  const delivery = buildDelivery({ attempt: 1, envelope, result, subscription }, { now })
  try {
    await recordDelivery(
      db,
      delivery,
      options.deliveryRetentionMs === undefined ? {} : { ttl: options.deliveryRetentionMs },
    )
  } catch {
    // delivery logging is best-effort
  }

  return { delivery, envelope, result }
}

export declare namespace ping {
  /** Options for {@link ping}. */
  type Options = {
    /** Delivery-log row TTL in ms; defaults to {@link deliveryRetentionMs}. */
    deliveryRetentionMs?: number | undefined
    /** `fetch` implementation passed through to {@link deliver}. */
    fetch?: typeof globalThis.fetch | undefined
    /** Deterministic event-id nonce; defaults to random (tests only). */
    nonce?: string | undefined
    /** Clock used for timestamps; injectable for tests. */
    now?: (() => Date) | undefined
    /** Delivery timeout in ms passed through to {@link deliver}. */
    timeoutMs?: number | undefined
  }
  /** Outcome of a ping: the classified delivery result plus the logged row. */
  type Result = {
    /** The delivery-log row that was recorded. */
    delivery: Delivery
    /** The synthetic envelope that was sent. */
    envelope: Envelope
    /** The classified transport result. */
    result: WebhookDestination.Result
  }
}

function generateId(now: Date): string {
  return Id.generateSortable('wh', now)
}

function generateDeliveryId(now: Date): string {
  return Id.generateSortable('whd', now)
}

function generateSecret(): string {
  return `whsec_${Hex.random(32).slice(2)}`
}

export class LimitExceededError extends Error {
  /** The cap that was reached. */
  limit: number

  constructor(limit: number) {
    super(`Webhook subscription limit reached (${limit}).`)
    this.name = 'Webhooks.LimitExceededError'
    this.limit = limit
  }
}

/** Thrown when an update would activate URL signing with an unrecoverable secret. */
export class InvalidDestinationTransitionError extends Error {
  constructor() {
    super('Create a new webhook to change a provider-managed destination to an HTTPS endpoint.')
    this.name = 'Webhooks.InvalidDestinationTransitionError'
  }
}

/**
 * Thrown when a subscription's `filters` fail to validate against the schema for
 * its `eventType`. The detection path treats stored filters as **fail-closed**:
 * an invalid filter never degrades to a firehose, so the API rejects bad filters
 * at create/patch time rather than silently widening the match set later.
 */
export class InvalidFilterError extends Error {
  /** Event type whose filters failed validation. */
  eventType: string
  /** Per-field validation issues, suitable for an API error `details` array. */
  details: readonly InvalidFilterError.Detail[]

  constructor(eventType: string, details: readonly InvalidFilterError.Detail[] = []) {
    super(`Invalid filters for "${eventType}" webhook subscription.`)
    this.name = 'Webhooks.InvalidFilterError'
    this.eventType = eventType
    this.details = details
  }
}

export declare namespace InvalidFilterError {
  /** A single validation issue: a message plus the path to the invalid value. */
  type Detail = {
    /** Human-readable validation message. */
    message: string
    /** Path to the invalid value within the `filters` object. */
    path?: readonly (string | number)[] | undefined
  }
}

/** Thrown when webhook operations run without a configured state store. */
export class UnconfiguredError extends Error {
  constructor() {
    super('Webhooks are not configured for this deployment.')
    this.name = 'Webhooks.UnconfiguredError'
  }
}