import { Webhooks } from 'tapimo'
import { sql } from 'kysely'
import * as TestApp from '../../test/App.js'
import * as WebhookSubscriptions from '../db/tables/webhookSubscriptions.js'
import * as Metrics from '../Metrics.js'
import * as Cursor from './Cursor.js'

const apiKeyOwner = { orgId: 'org_1', type: 'api_key' } as const
const mppOwner = { payer: 'did:pkh:eip155:1:0xabc', type: 'mpp' } as const

/** Fixed clock so created/updated timestamps snapshot deterministically. */
const fixedNow = () => new Date('2026-01-01T00:00:00.000Z')

/** Asserts the random id/secret shape, then redacts them for a stable snapshot. */
function redact(subscription: Webhooks.Subscription | null) {
  expect(subscription).not.toBeNull()
  expect(subscription?.id).toMatch(/^wh_\d{15}_[A-Za-z0-9]{24}$/)
  expect(subscription?.secret).toMatch(/^whsec_[0-9a-f]{64}$/)
  return { ...subscription, id: '<id>', secret: '<secret>' }
}

function createInput(overrides: Partial<Webhooks.CreateInput> = {}): Webhooks.CreateInput {
  return {
    chainId: 4217,
    destination: { type: 'url', url: 'https://hooks.example.com/endpoint' },
    eventType: 'token:transfer',
    owner: apiKeyOwner,
    ...overrides,
  }
}

describe('createSubscription', () => {
  test('persists a subscription with generated id and secret', async () => {
    const db = TestApp.database()
    const subscription = await Webhooks.createSubscription(db, createInput(), { now: fixedNow })

    expect(redact(subscription)).toMatchInlineSnapshot(`
      {
        "chainId": 4217,
        "createdAt": "2026-01-01T00:00:00.000Z",
        "destination": {
          "type": "url",
          "url": "https://hooks.example.com/endpoint",
        },
        "eventType": "token:transfer",
        "failureCount": 0,
        "filters": {},
        "id": "<id>",
        "owner": {
          "orgId": "org_1",
          "type": "api_key",
        },
        "secret": "<secret>",
        "status": "active",
        "updatedAt": "2026-01-01T00:00:00.000Z",
      }
    `)

    const read = await Webhooks.getSubscription(db, apiKeyOwner, subscription.id)
    expect(read).toEqual(subscription)
  })

  test('commits an initial poller cursor through both insert paths', async () => {
    const db = TestApp.database()
    const direct = await Webhooks.createSubscription(db, createInput(), {
      startBlockNumber: 123,
    })
    const capped = await Webhooks.createSubscription(db, createInput({ owner: mppOwner }), {
      maxPerOwner: 1,
      startBlockNumber: 456,
    })

    expect(Cursor.decode((await Webhooks.getCursor(db, direct.id))!, ['int']))
      .toMatchInlineSnapshot(`
      [
        123,
      ]
    `)
    expect(Cursor.decode((await Webhooks.getCursor(db, capped.id))!, ['int']))
      .toMatchInlineSnapshot(`
      [
        456,
      ]
    `)
  })

  test('rejects an invalid initial block number before persisting', async () => {
    const db = TestApp.database()

    for (const startBlockNumber of [-1, 1.5, Number.MAX_SAFE_INTEGER + 1])
      await expect(
        Webhooks.createSubscription(db, createInput(), { startBlockNumber }),
      ).rejects.toThrowErrorMatchingInlineSnapshot(
        `[RangeError: startBlockNumber must be a non-negative safe integer.]`,
      )
    expect(await Webhooks.listSubscriptions(db, apiKeyOwner)).toEqual([])
  })

  test('sets expiresAt from ttl for MPP-owned subscriptions', async () => {
    const db = TestApp.database()
    const subscription = await Webhooks.createSubscription(
      db,
      createInput({ owner: mppOwner, ttl: 60_000 }),
      { now: fixedNow },
    )

    expect(subscription.expiresAt).toMatchInlineSnapshot(`"2026-01-01T00:01:00.000Z"`)
  })

  test('enforces a per-owner cap', async () => {
    const db = TestApp.database()
    await Webhooks.createSubscription(db, createInput(), { maxPerOwner: 1 })

    await expect(
      Webhooks.createSubscription(db, createInput(), { maxPerOwner: 1 }),
    ).rejects.toThrowErrorMatchingInlineSnapshot(
      `[Webhooks.LimitExceededError: Webhook subscription limit reached (1).]`,
    )
  })

  test('counts paused subscriptions toward the live subscription cap', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput(), { maxPerOwner: 1 })
    await Webhooks.updateSubscription(db, apiKeyOwner, created.id, { status: 'paused' })

    await expect(
      Webhooks.createSubscription(db, createInput(), { maxPerOwner: 1 }),
    ).rejects.toBeInstanceOf(Webhooks.LimitExceededError)
    await Webhooks.deleteSubscription(db, apiKeyOwner, created.id)
    await expect(
      Webhooks.createSubscription(db, createInput(), { maxPerOwner: 1 }),
    ).resolves.toMatchObject({ status: 'active' })
  })

  test('enforces the per-owner cap across concurrent creates', async () => {
    const getDb = TestApp.databaseFactory()
    const results = await Promise.allSettled(
      Array.from({ length: 8 }, () =>
        Webhooks.createSubscription(getDb(), createInput(), { maxPerOwner: 3 }),
      ),
    )

    expect(results.map((result) => result.status).sort()).toMatchInlineSnapshot(`
      [
        "fulfilled",
        "fulfilled",
        "fulfilled",
        "rejected",
        "rejected",
        "rejected",
        "rejected",
        "rejected",
      ]
    `)
    for (const result of results)
      if (result.status === 'rejected') expect(result.reason).toBeInstanceOf(Webhooks.LimitExceededError) // prettier-ignore
    expect(await Webhooks.listSubscriptions(getDb(), apiKeyOwner)).toHaveLength(3)
  })

  test('rejects an undeliverable URL before persisting', async () => {
    const db = TestApp.database()
    await expect(
      Webhooks.createSubscription(
        db,
        createInput({ destination: { type: 'url', url: 'http://localhost/x' } }),
      ),
    ).rejects.toThrowErrorMatchingInlineSnapshot(
      `[Webhooks.InvalidUrlError: Webhook URL rejected (protocol): http://localhost/x]`,
    )

    expect(await Webhooks.listSubscriptions(db, apiKeyOwner)).toMatchInlineSnapshot(`[]`)
  })
})

describe('listActiveForFundingEvent', () => {
  test('keeps selected subscriptions until the caller transaction commits', async () => {
    const getDb = TestApp.databaseFactory()
    const owner = { orgId: 'org_funding_subscription_lock', type: 'api_key' } as const
    const subscription = await Webhooks.createSubscription(
      getDb(),
      createInput({
        environment: 'production',
        eventType: 'funding:transfer.updated',
        owner,
      }),
    )
    const locked = Promise.withResolvers<void>()
    const release = Promise.withResolvers<void>()
    const read = getDb().transaction(async (tx) => {
      const subscriptions = await WebhookSubscriptions.listActiveForFundingEvent(tx, {
        chainId: 4217,
        environment: 'production',
        eventType: 'funding:transfer.updated',
        now: new Date().toISOString(),
        orgId: owner.orgId,
      })
      locked.resolve()
      await release.promise
      return subscriptions
    })
    await locked.promise

    const deletion = await getDb()
      .transaction(async (tx) => {
        await sql`select set_config('lock_timeout', '100ms', true)`.execute(tx.kysely)
        return Webhooks.deleteSubscription(tx, owner, subscription.id)
      })
      .catch((cause: unknown) => cause)
    release.resolve()

    await expect(read).resolves.toEqual([subscription])
    expect(deletion).toBeInstanceOf(Error)
    expect(String(deletion)).toContain('lock timeout')
    await expect(Webhooks.deleteSubscription(getDb(), owner, subscription.id)).resolves.toBe(true)
  })
})

describe('owner isolation', () => {
  test('an owner cannot read or list another owner subscriptions', async () => {
    const db = TestApp.database()
    const mine = await Webhooks.createSubscription(db, createInput({ owner: apiKeyOwner }))
    await Webhooks.createSubscription(db, createInput({ owner: mppOwner }))

    expect(await Webhooks.getSubscription(db, mppOwner, mine.id)).toMatchInlineSnapshot(`null`)

    const apiKeyList = await Webhooks.listSubscriptions(db, apiKeyOwner)
    expect(apiKeyList.map((s) => s.id)).toEqual([mine.id])

    const mppList = await Webhooks.listSubscriptions(db, mppOwner)
    expect(mppList).toHaveLength(1)
    expect(mppList[0]?.id).not.toBe(mine.id)
  })

  test('an owner cannot delete another owner subscription', async () => {
    const db = TestApp.database()
    const mine = await Webhooks.createSubscription(db, createInput({ owner: apiKeyOwner }))

    expect(await Webhooks.deleteSubscription(db, mppOwner, mine.id)).toMatchInlineSnapshot(`false`)
    expect(await Webhooks.getSubscription(db, apiKeyOwner, mine.id)).not.toBeNull()
  })
})

describe('listSubscriptions', () => {
  test('lists subscriptions newest-first with keyset paging', async () => {
    const db = TestApp.database()
    const created = []
    for (const minute of [0, 1, 2]) {
      const now = () => new Date(`2026-01-01T00:0${minute}:00.000Z`)
      created.push(await Webhooks.createSubscription(db, createInput(), { now }))
    }
    const [first, second, third] = created

    // Newest first: most recently created subscription leads.
    const all = await Webhooks.listSubscriptions(db, apiKeyOwner)
    expect(all.map((s) => s.id)).toEqual([third!.id, second!.id, first!.id])

    // First page of two, then resume after the last id for the older remainder.
    const page = await Webhooks.listSubscriptions(db, apiKeyOwner, { limit: 2 })
    expect(page.map((s) => s.id)).toEqual([third!.id, second!.id])

    const next = await Webhooks.listSubscriptions(db, apiKeyOwner, {
      cursor: page.at(-1)?.id,
      limit: 2,
    })
    expect(next.map((s) => s.id)).toEqual([first!.id])
  })
})

describe('updateSubscription', () => {
  test('patches fields and bumps updatedAt', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput(), { now: fixedNow })

    const updated = await Webhooks.updateSubscription(
      db,
      apiKeyOwner,
      created.id,
      {
        destination: { type: 'url', url: 'https://hooks.example.com/v2' },
        filters: { token: `0x${'aa'.repeat(20)}` },
      },
      { now: () => new Date('2026-01-02T00:00:00.000Z') },
    )

    expect(redact(updated)).toMatchInlineSnapshot(`
      {
        "chainId": 4217,
        "createdAt": "2026-01-01T00:00:00.000Z",
        "destination": {
          "type": "url",
          "url": "https://hooks.example.com/v2",
        },
        "eventType": "token:transfer",
        "failureCount": 0,
        "filters": {
          "token": "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
        },
        "id": "<id>",
        "owner": {
          "orgId": "org_1",
          "type": "api_key",
        },
        "secret": "<secret>",
        "status": "active",
        "updatedAt": "2026-01-02T00:00:00.000Z",
      }
    `)
  })

  test('returns null for a missing subscription', async () => {
    const db = TestApp.database()
    expect(
      await Webhooks.updateSubscription(db, apiKeyOwner, 'wh_missing', { status: 'paused' }),
    ).toMatchInlineSnapshot(`null`)
  })

  test('rejects a patched URL that fails validation', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())

    await expect(
      Webhooks.updateSubscription(db, apiKeyOwner, created.id, {
        destination: { type: 'url', url: 'http://10.0.0.1' },
      }),
    ).rejects.toThrowErrorMatchingInlineSnapshot(
      `[Webhooks.InvalidUrlError: Webhook URL rejected (protocol): http://10.0.0.1]`,
    )
  })

  test('requires recreation before activating URL signing for a provider destination', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(
      db,
      createInput({
        destination: {
          type: 'slack',
          url: 'https://hooks.slack.com/services/T000/B000/xxxx',
        },
      }),
    )

    await expect(
      Webhooks.updateSubscription(db, apiKeyOwner, created.id, {
        destination: { type: 'url', url: 'https://hooks.example.com/v2' },
      }),
    ).rejects.toThrowErrorMatchingInlineSnapshot(
      `[Webhooks.InvalidDestinationTransitionError: Create a new webhook to change a provider-managed destination to an HTTPS endpoint.]`,
    )
  })
})

describe('listActiveForChain', () => {
  test('returns active subscriptions across event types in one read', async () => {
    const db = TestApp.database()
    const transfer = await Webhooks.createSubscription(db, createInput())
    const transaction = await Webhooks.createSubscription(
      db,
      createInput({ eventType: 'transaction:included' }),
    )
    await Webhooks.createSubscription(
      db,
      createInput({ eventType: 'block:created', owner: mppOwner }),
    )
    await Webhooks.createSubscription(db, createInput({ chainId: 42431, owner: mppOwner }))
    await Webhooks.updateSubscription(db, apiKeyOwner, transfer.id, { status: 'paused' })

    const pairs = await Webhooks.listActiveForChain(db, {
      chainId: 4217,
      eventTypes: ['transaction:included'],
    })
    expect(pairs.map(({ subscription }) => subscription.id)).toEqual([transaction.id])
  })
})

describe('recordSuccess / recordFailure', () => {
  test('recordSuccess resets failures and stamps lastDeliveryAt', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())
    const failed = await Webhooks.recordFailure(db, created, { now: fixedNow })
    expect(failed.failureCount).toMatchInlineSnapshot(`1`)

    const ok = await Webhooks.recordSuccess(db, failed, { now: fixedNow })
    expect({ failureCount: ok.failureCount, lastDeliveryAt: ok.lastDeliveryAt })
      .toMatchInlineSnapshot(`
      {
        "failureCount": 0,
        "lastDeliveryAt": "2026-01-01T00:00:00.000Z",
      }
    `)

    // Persisted, not just returned.
    const read = await Webhooks.getSubscription(db, apiKeyOwner, created.id)
    expect(read?.lastDeliveryAt).toBe('2026-01-01T00:00:00.000Z')
  })

  test('recordSuccess refreshes the stamp at most once per precision window', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())
    const stamped = await Webhooks.recordSuccess(db, created, { now: fixedNow })
    expect(stamped.lastDeliveryAt).toBe(fixedNow().toISOString())

    // A success moments later has nothing to move: the counter is already zero
    // and the stamp is inside its window, so the row is never rewritten.
    const soon = new Date(fixedNow().getTime() + 1_000)
    await Webhooks.recordSuccess(db, stamped, { now: () => soon })
    const unmoved = await Webhooks.getSubscription(db, apiKeyOwner, created.id)
    expect(unmoved?.lastDeliveryAt, 'the stored stamp must not move').toBe(fixedNow().toISOString())

    // Past any plausible window; the exact bound is internal.
    const later = new Date(fixedNow().getTime() + 3_600_000)
    await Webhooks.recordSuccess(db, stamped, { now: () => later })
    const moved = await Webhooks.getSubscription(db, apiKeyOwner, created.id)
    expect(moved?.lastDeliveryAt).toBe(later.toISOString())
  })

  test('recordSuccess always resets a non-zero failure counter', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())
    // Stamp first, so only the counter is out of date: the reset must not be
    // suppressed by a fresh stamp.
    const stamped = await Webhooks.recordSuccess(db, created, { now: fixedNow })
    const failed = await Webhooks.recordFailure(db, stamped, { now: fixedNow })
    expect(failed.failureCount).toBe(1)

    await Webhooks.recordSuccess(db, failed, { now: fixedNow })

    const read = await Webhooks.getSubscription(db, apiKeyOwner, created.id)
    expect(read?.failureCount, 'a pending reset must never be skipped').toBe(0)
  })

  test('recordFailure auto-disables at the threshold and drops from the active index', async () => {
    const db = TestApp.database()
    let subscription = await Webhooks.createSubscription(db, createInput())

    // Below the threshold: stays active.
    subscription = await Webhooks.recordFailure(db, subscription, { maxFailures: 3 })
    subscription = await Webhooks.recordFailure(db, subscription, { maxFailures: 3 })
    expect(subscription.status).toMatchInlineSnapshot(`"active"`)
    expect(
      (
        await Webhooks.listActiveForChain(db, { chainId: 4217, eventTypes: ['token:transfer'] })
      ).map((pair) => pair.subscription.id),
    ).toEqual([subscription.id])

    // Hitting the threshold disables it and removes it from the active index.
    subscription = await Webhooks.recordFailure(db, subscription, { maxFailures: 3 })
    expect({ failureCount: subscription.failureCount, status: subscription.status })
      .toMatchInlineSnapshot(`
      {
        "failureCount": 3,
        "status": "disabled",
      }
    `)
    expect(
      await Webhooks.listActiveForChain(db, { chainId: 4217, eventTypes: ['token:transfer'] }),
    ).toMatchInlineSnapshot(`[]`)
  })

  test('recordFailure leaves an already-disabled subscription alone at the cap', async () => {
    const db = TestApp.database()
    let subscription = await Webhooks.createSubscription(db, createInput())
    subscription = await Webhooks.recordFailure(db, subscription, { maxFailures: 2, now: fixedNow })
    subscription = await Webhooks.recordFailure(db, subscription, { maxFailures: 2, now: fixedNow })
    expect(subscription.status).toBe('disabled')

    // A straggler failing after the disable has nothing to move: the row is
    // never rewritten.
    const later = new Date(fixedNow().getTime() + 1_000)
    await Webhooks.recordFailure(db, subscription, { maxFailures: 2, now: () => later })

    const read = await Webhooks.getSubscription(db, apiKeyOwner, subscription.id)
    expect(read?.failureCount, 'the stored counter must not move').toBe(2)
    expect(read?.updatedAt, 'the stored stamp must not move').toBe(fixedNow().toISOString())
  })

  test('recordFailure re-disables a re-enabled subscription still at the cap', async () => {
    const db = TestApp.database()
    let subscription = await Webhooks.createSubscription(db, createInput())
    subscription = await Webhooks.recordFailure(db, subscription, { maxFailures: 1, now: fixedNow })
    expect(subscription.status).toBe('disabled')

    await Webhooks.updateSubscription(db, apiKeyOwner, subscription.id, { status: 'active' })

    // The counter still sits at the cap, so this failure must write again.
    await Webhooks.recordFailure(db, subscription, { maxFailures: 1, now: fixedNow })
    const read = await Webhooks.getSubscription(db, apiKeyOwner, subscription.id)
    expect(read?.status).toBe('disabled')
  })
})

describe('sign / verify', () => {
  test('signs a body deterministically', () => {
    const signature = Webhooks.sign({
      body: JSON.stringify({ hello: 'world' }),
      secret: 'whsec_test',
      timestamp: 1_000,
    })

    expect(signature).toMatchInlineSnapshot(
      `"t=1000,v1=b345b1f6668bbb1b5f1a02cc5a2de10334d1281ba860bbdeadf4681c4719563d"`,
    )
  })

  test('verifies a freshly signed body', () => {
    const secret = 'whsec_test'
    const body = JSON.stringify({ hello: 'world' })
    const signature = Webhooks.sign({ body, secret, timestamp: 1_000 })

    expect(Webhooks.verify({ body, now: 1_000, secret, signature })).toMatchInlineSnapshot(`true`)
  })

  test('rejects a tampered body', () => {
    const secret = 'whsec_test'
    const signature = Webhooks.sign({ body: 'original', secret, timestamp: 1_000 })

    expect(
      Webhooks.verify({ body: 'tampered', now: 1_000, secret, signature }),
    ).toMatchInlineSnapshot(`false`)
  })

  test('rejects the wrong secret', () => {
    const signature = Webhooks.sign({ body: 'x', secret: 'a', timestamp: 1_000 })
    expect(
      Webhooks.verify({ body: 'x', now: 1_000, secret: 'b', signature }),
    ).toMatchInlineSnapshot(`false`)
  })

  test('rejects a stale signature beyond tolerance', () => {
    const secret = 'whsec_test'
    const signature = Webhooks.sign({ body: 'x', secret, timestamp: 1_000 })

    expect(
      Webhooks.verify({ body: 'x', now: 2_000, secret, signature, tolerance: 300 }),
    ).toMatchInlineSnapshot(`false`)
    expect(
      Webhooks.verify({ body: 'x', now: 1_200, secret, signature, tolerance: 300 }),
    ).toMatchInlineSnapshot(`true`)
  })

  test('rejects a malformed signature', () => {
    expect(
      Webhooks.verify({ body: 'x', secret: 'a', signature: 'nonsense' }),
    ).toMatchInlineSnapshot(`false`)
  })
})

describe('eventId / buildEnvelope', () => {
  test('eventId is stable for the same coordinates', () => {
    const a = Webhooks.eventId({
      blockNumber: 100,
      chainId: 4217,
      eventType: 'token:transfer',
      logIndex: 3,
    })
    const b = Webhooks.eventId({
      blockNumber: 100,
      chainId: 4217,
      eventType: 'token:transfer',
      logIndex: 3,
    })
    const different = Webhooks.eventId({
      blockNumber: 100,
      chainId: 4217,
      eventType: 'token:transfer',
      logIndex: 4,
    })

    expect(a).toMatchInlineSnapshot(
      `"evt_d7c75afd0abccdc90a391d9313762a957c7e4383cac1fae76e5de56e49c9d3f8"`,
    )
    expect(a).toBe(b)
    expect(a).not.toBe(different)
  })

  test('buildEnvelope embeds the data row and a stable id', async () => {
    const db = TestApp.database()
    const subscription = await Webhooks.createSubscription(db, createInput())

    const envelope = Webhooks.buildEnvelope({
      blockNumber: 100,
      createdAt: new Date('2026-01-01T00:00:00.000Z'),
      data: { amount: '1', to: `0x${'aa'.repeat(20)}` },
      logIndex: 3,
      subscription,
    })

    expect(envelope.subscriptionId).toBe(subscription.id)
    expect({ ...envelope, subscriptionId: '<subscriptionId>' }).toMatchInlineSnapshot(`
      {
        "chainId": 4217,
        "createdAt": "2026-01-01T00:00:00.000Z",
        "data": {
          "amount": "1",
          "to": "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
        },
        "id": "evt_d7c75afd0abccdc90a391d9313762a957c7e4383cac1fae76e5de56e49c9d3f8",
        "subscriptionId": "<subscriptionId>",
        "type": "token:transfer",
      }
    `)
  })
})

describe('delivery log', () => {
  /** Asserts the time-sortable delivery id shape and redacts non-deterministic fields. */
  function redactDelivery(delivery: Webhooks.Delivery) {
    expect(delivery.id).toMatch(/^whd_\d{15}_[A-Za-z0-9]{24}$/)
    expect(delivery.subscriptionId).toMatch(/^wh_\d{15}_[A-Za-z0-9]{24}$/)
    return {
      ...delivery,
      envelope: { ...delivery.envelope, subscriptionId: '<subscriptionId>' },
      id: '<id>',
      subscriptionId: '<subscriptionId>',
      ...(delivery.responseMs === undefined ? {} : { responseMs: '<ms>' }),
    }
  }

  /** A delivery row with an explicit, lexically-sortable id for ordering tests. */
  function row(subscriptionId: string, n: number): Webhooks.Delivery {
    return {
      attempt: n,
      createdAt: new Date(1_700_000_000_000 + n * 1_000).toISOString(),
      envelope: {
        chainId: 4217,
        createdAt: new Date(1_700_000_000_000 + n * 1_000).toISOString(),
        data: { amount: String(n) },
        id: `evt_${n}`,
        subscriptionId,
        type: 'token:transfer',
      },
      eventId: `evt_${n}`,
      id: `whd_${String(1_700_000_000_000 + n * 1_000).padStart(15, '0')}_${'0'.repeat(16)}`,
      requestUrl: 'https://hooks.example.com/endpoint',
      status: 'succeeded',
      subscriptionId,
    }
  }

  test('lists rows newest-first with keyset paging', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())
    for (const n of [1, 2, 3]) await Webhooks.recordDelivery(db, row(created.id, n))

    // Newest first.
    const all = await Webhooks.listDeliveries(db, created.id)
    expect(all.map((d) => d.eventId)).toMatchInlineSnapshot(`
      [
        "evt_3",
        "evt_2",
        "evt_1",
      ]
    `)

    // First page of two, then resume after the last id for the older remainder.
    const page = await Webhooks.listDeliveries(db, created.id, { limit: 2 })
    expect(page.map((d) => d.eventId)).toEqual(['evt_3', 'evt_2'])

    const next = await Webhooks.listDeliveries(db, created.id, {
      cursor: page.at(-1)?.id,
      limit: 2,
    })
    expect(next.map((d) => d.eventId)).toEqual(['evt_1'])
  })

  test('deliverAndRecord appends a succeeded row', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())
    const envelope = {
      chainId: 4217,
      createdAt: '2026-01-01T00:00:00.000Z',
      data: { amount: '1' },
      id: 'evt_ok',
      subscriptionId: created.id,
      type: 'token:transfer',
    } satisfies Webhooks.Envelope

    const fetch = (async () =>
      new Response(null, { status: 200 })) satisfies typeof globalThis.fetch
    await Webhooks.deliverAndRecord(db, created, envelope, { fetch, now: fixedNow })

    const [delivery] = await Webhooks.listDeliveries(db, created.id)
    expect(redactDelivery(delivery!)).toMatchInlineSnapshot(`
      {
        "attempt": 1,
        "createdAt": "2026-01-01T00:00:00.000Z",
        "envelope": {
          "chainId": 4217,
          "createdAt": "2026-01-01T00:00:00.000Z",
          "data": {
            "amount": "1",
          },
          "id": "evt_ok",
          "subscriptionId": "<subscriptionId>",
          "type": "token:transfer",
        },
        "eventId": "evt_ok",
        "id": "<id>",
        "requestUrl": "https://hooks.example.com/endpoint",
        "responseMs": "<ms>",
        "responseStatus": 200,
        "status": "succeeded",
        "subscriptionId": "<subscriptionId>",
      }
    `)
  })

  test('deliverAndRecord partitions event, enqueue, and queue latency', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())
    const histograms = new Map<string, number>()
    const metrics = Metrics.from({
      count() {},
      flush() {},
      gauge() {},
      histogram(name, value) {
        histograms.set(name, value)
      },
    })
    const envelope = {
      chainId: 4217,
      createdAt: '1970-01-01T00:00:00.500Z',
      data: { timestamp: '1970-01-01T00:00:00.000Z' },
      id: 'evt_null',
      subscriptionId: created.id,
      type: 'token:transfer',
    } satisfies Webhooks.Envelope
    const fetch = (async () =>
      new Response(null, { status: 200 })) satisfies typeof globalThis.fetch

    await expect(
      Webhooks.deliverAndRecord(db, created, envelope, {
        dequeuedAt: 1_250,
        fetch,
        metrics,
        now: fixedNow,
        queueAttempt: 2,
        queuedAt: 1_000,
      }),
    ).resolves.toMatchObject({ ok: true })
    expect(histograms.get('webhook_delivery_envelope_to_queue_ms')).toBe(500)
    expect(histograms.get('webhook_delivery_event_to_envelope_ms')).toBe(500)
    expect(histograms.get('webhook_delivery_queue_wait_ms')).toBe(250)
  })

  test('deliverAndRecord appends a failed row with the error and status', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())
    const envelope = {
      chainId: 4217,
      createdAt: '2026-01-01T00:00:00.000Z',
      data: { amount: '1' },
      id: 'evt_fail',
      subscriptionId: created.id,
      type: 'token:transfer',
    } satisfies Webhooks.Envelope

    const fetch = (async () =>
      new Response(null, { status: 500 })) satisfies typeof globalThis.fetch
    await Webhooks.deliverAndRecord(db, created, envelope, { fetch, now: fixedNow })

    const [delivery] = await Webhooks.listDeliveries(db, created.id)
    expect(redactDelivery(delivery!)).toMatchInlineSnapshot(`
      {
        "attempt": 1,
        "createdAt": "2026-01-01T00:00:00.000Z",
        "envelope": {
          "chainId": 4217,
          "createdAt": "2026-01-01T00:00:00.000Z",
          "data": {
            "amount": "1",
          },
          "id": "evt_fail",
          "subscriptionId": "<subscriptionId>",
          "type": "token:transfer",
        },
        "error": "non-2xx response (500)",
        "eventId": "evt_fail",
        "id": "<id>",
        "requestUrl": "https://hooks.example.com/endpoint",
        "responseMs": "<ms>",
        "responseStatus": 500,
        "status": "failed",
        "subscriptionId": "<subscriptionId>",
      }
    `)
  })

  test('deliverAndRecord records envelope serialization failures', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())
    const envelope = {
      chainId: 4217,
      createdAt: '2026-01-01T00:00:00.000Z',
      data: { amount: 1n },
      id: 'evt_invalid',
      subscriptionId: created.id,
      type: 'token:transfer',
    } satisfies Webhooks.Envelope

    const result = await Webhooks.deliverAndRecord(db, created, envelope, { now: fixedNow })

    expect(result).toEqual({ error: 'delivery failed before request', ok: false })
    const after = await Webhooks.getSubscription(db, apiKeyOwner, created.id)
    expect(after?.failureCount).toBe(1)
  })

  test('deliverCurrentAndRecord skips paused and deleted queued subscriptions', async () => {
    const db = TestApp.database()
    const paused = await Webhooks.createSubscription(db, createInput())
    const removed = await Webhooks.createSubscription(db, createInput())
    const pausedEnvelope = Webhooks.buildEnvelope({
      blockNumber: 1,
      data: { amount: '1' },
      logIndex: 0,
      subscription: paused,
    })
    const removedEnvelope = Webhooks.buildEnvelope({
      blockNumber: 1,
      data: { amount: '1' },
      logIndex: 1,
      subscription: removed,
    })
    await Webhooks.updateSubscription(db, apiKeyOwner, paused.id, { status: 'paused' })
    await Webhooks.deleteSubscription(db, apiKeyOwner, removed.id)
    let requests = 0
    const fetch = (async () => {
      requests++
      return new Response(null, { status: 200 })
    }) satisfies typeof globalThis.fetch

    await expect(
      Webhooks.deliverCurrentAndRecord(db, paused, pausedEnvelope, { fetch }),
    ).resolves.toEqual({ status: 'skipped' })
    await expect(
      Webhooks.deliverCurrentAndRecord(db, removed, removedEnvelope, { fetch }),
    ).resolves.toEqual({ status: 'skipped' })
    expect(requests).toBe(0)
  })

  test('deliverCurrentAndRecord uses the current destination and secret', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())
    const envelope = Webhooks.buildEnvelope({
      blockNumber: 1,
      data: { amount: '1' },
      logIndex: 0,
      subscription: created,
    })
    const current = await Webhooks.updateSubscription(db, apiKeyOwner, created.id, {
      destination: { type: 'url', url: 'https://hooks.example.com/current' },
    })
    const queued = { ...created, secret: `whsec_${'ff'.repeat(32)}` }
    const captured: { body: string; signature: string | null; url: string } = {
      body: '',
      signature: null,
      url: '',
    }
    const fetch = (async (input, init) => {
      captured.body = typeof init?.body === 'string' ? init.body : ''
      captured.signature = new Headers(init?.headers).get(Webhooks.signatureHeader)
      captured.url = input instanceof Request ? input.url : input.toString()
      return new Response(null, { status: 200 })
    }) satisfies typeof globalThis.fetch

    const delivery = await Webhooks.deliverCurrentAndRecord(db, queued, envelope, { fetch })

    expect(delivery.status).toBe('delivered')
    expect(captured.url).toBe('https://hooks.example.com/current')
    expect(
      Webhooks.verify({
        body: captured.body,
        secret: current!.secret,
        signature: captured.signature ?? '',
        tolerance: Number.POSITIVE_INFINITY,
      }),
    ).toBe(true)
    expect(
      Webhooks.verify({
        body: captured.body,
        secret: queued.secret,
        signature: captured.signature ?? '',
        tolerance: Number.POSITIVE_INFINITY,
      }),
    ).toBe(false)
  })

  test('deleteSubscription removes the delivery log', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())
    for (const n of [1, 2]) await Webhooks.recordDelivery(db, row(created.id, n))
    expect(await Webhooks.listDeliveries(db, created.id)).toHaveLength(2)

    await Webhooks.deleteSubscription(db, apiKeyOwner, created.id)
    expect(await Webhooks.listDeliveries(db, created.id)).toMatchInlineSnapshot(`[]`)
  })

  test('getDelivery loads one row by id and returns null when absent', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())
    await Webhooks.recordDelivery(db, row(created.id, 1))
    const id = row(created.id, 1).id

    const found = await Webhooks.getDelivery(db, created.id, id)
    expect(found?.eventId).toMatchInlineSnapshot(`"evt_1"`)
    // The persisted row carries the full envelope for replay.
    expect(found?.envelope.id).toMatchInlineSnapshot(`"evt_1"`)

    expect(await Webhooks.getDelivery(db, created.id, 'whd_missing')).toMatchInlineSnapshot(`null`)
  })
})

async function stagedObligation(db: ReturnType<typeof TestApp.database>, logIndex = 4) {
  const subscription = await Webhooks.createSubscription(db, createInput())
  const envelope = Webhooks.buildEnvelope({
    blockNumber: 123,
    createdAt: new Date(),
    data: { amount: '1' },
    logIndex,
    subscription,
  })
  const { created, references } = await Webhooks.ensureQueueEvents(db, [{ envelope, subscription }])
  return { created, envelope, reference: references[0]!, subscription }
}

describe('ensureQueueEvents', () => {
  test('keeps oversized envelopes behind compact idempotent Queue references', async () => {
    const db = TestApp.database()
    const subscription = await Webhooks.createSubscription(
      db,
      createInput({ eventType: 'transaction:included' }),
    )
    const input = `0x${'ab'.repeat(70_000)}`
    const envelope = Webhooks.buildEnvelope({
      blockNumber: 123,
      createdAt: fixedNow(),
      data: { input, meta: { rpc: { input } } },
      logIndex: 4,
      subscription,
    })
    const dispatchable = { envelope, subscription }

    const { references: [reference] } = await Webhooks.ensureQueueEvents(db, [dispatchable]) // prettier-ignore
    await Webhooks.ensureQueueEvents(db, [dispatchable])

    expect(
      new TextEncoder().encode(JSON.stringify({ body: dispatchable })).byteLength,
    ).toBeGreaterThan(128_000)
    expect(new TextEncoder().encode(JSON.stringify({ body: reference })).byteLength).toBeLessThan(
      1_000,
    )
    expect((await Webhooks.getQueueEvent(db, reference!))?.envelope).toEqual(envelope)
  })

  test('delivers a staged oversized envelope from its compact reference', async () => {
    const db = TestApp.database()
    const subscription = await Webhooks.createSubscription(
      db,
      createInput({ eventType: 'transaction:included' }),
    )
    const input = `0x${'ab'.repeat(70_000)}`
    const envelope = Webhooks.buildEnvelope({
      blockNumber: 123,
      data: { input, meta: { rpc: { input } } },
      logIndex: 4,
      subscription,
    })
    const { references: [reference] } = await Webhooks.ensureQueueEvents(db, [{ envelope, subscription }]) // prettier-ignore
    const histograms = new Map<string, number>()
    const metrics = Metrics.from({
      count() {},
      flush() {},
      gauge() {},
      histogram(name, value) {
        histograms.set(name, value)
      },
    })
    let deliveredBytes = 0
    const fetch = (async (_url, init) => {
      deliveredBytes = new TextEncoder().encode(
        typeof init?.body === 'string' ? init.body : '',
      ).byteLength
      return new Response(null, { status: 200 })
    }) satisfies typeof globalThis.fetch

    const result = await Webhooks.deliverQueueEventAndRecord(db, reference!, {
      fetch,
      metrics,
      queuedAt: Date.now(),
    })

    expect(result.status).toBe('delivered')
    expect(deliveredBytes).toBeGreaterThan(128_000)
    expect(histograms.has('webhook_delivery_processing_ms')).toBe(true)
    expect(histograms.has('webhook_delivery_queue_wait_ms')).toBe(true)
    expect(histograms.has('webhook_delivery_subscription_read_ms')).toBe(true)
  })

  test('measures observation to attempt from the staged head stamp', async () => {
    const db = TestApp.database()
    const subscription = await Webhooks.createSubscription(
      db,
      createInput({ eventType: 'transaction:included' }),
    )
    const envelope = Webhooks.buildEnvelope({
      blockNumber: 123,
      data: {},
      logIndex: 4,
      subscription,
    })
    const observedAt = Date.now() - 5_000
    const { references: [reference] } = await Webhooks.ensureQueueEvents(db, [{ envelope, subscription }], { observedAt }) // prettier-ignore
    const histograms = new Map<string, number>()
    const metrics = Metrics.from({
      count() {},
      flush() {},
      gauge() {},
      histogram(name, value) {
        histograms.set(name, value)
      },
    })
    const fetch = (async () => new Response(null, { status: 200 })) satisfies typeof globalThis.fetch // prettier-ignore

    const result = await Webhooks.deliverQueueEventAndRecord(db, reference!, { fetch, metrics })

    expect(result.status).toBe('delivered')
    // Measured from the head stamp, so it covers the whole controllable span
    // rather than only the time since this worker dequeued the message.
    expect(histograms.get('webhook_observed_to_attempt_ms')).toBeGreaterThanOrEqual(5_000)
    expect(histograms.has('webhook_delivery_preflight_ms')).toBe(true)
    expect(histograms.has('webhook_delivery_settle_ms')).toBe(true)
  })

  test('omits observation timing for a replay with no staged head stamp', async () => {
    const db = TestApp.database()
    const subscription = await Webhooks.createSubscription(
      db,
      createInput({ eventType: 'transaction:included' }),
    )
    const envelope = Webhooks.buildEnvelope({
      blockNumber: 124,
      data: {},
      logIndex: 4,
      subscription,
    })
    const { references: [reference] } = await Webhooks.ensureQueueEvents(db, [{ envelope, subscription }]) // prettier-ignore
    const histograms = new Map<string, number>()
    const metrics = Metrics.from({
      count() {},
      flush() {},
      gauge() {},
      histogram(name, value) {
        histograms.set(name, value)
      },
    })
    const fetch = (async () => new Response(null, { status: 200 })) satisfies typeof globalThis.fetch // prettier-ignore

    await Webhooks.deliverQueueEventAndRecord(db, reference!, { fetch, metrics })

    // Absent rather than zero: a replay is unattributable, not instant.
    expect(histograms.has('webhook_observed_to_attempt_ms')).toBe(false)
    expect(histograms.has('webhook_event_to_attempt_ms')).toBe(true)
  })

  test('skips a staged event after its subscription expires', async () => {
    vi.useFakeTimers({ now: fixedNow(), toFake: ['Date'] })
    try {
      const db = TestApp.database()
      const subscription = await Webhooks.createSubscription(
        db,
        createInput({ eventType: 'transaction:included', owner: mppOwner, ttl: 60_000 }),
        { now: fixedNow },
      )
      const envelope = Webhooks.buildEnvelope({
        blockNumber: 123,
        data: {},
        logIndex: 4,
        subscription,
      })
      const { references: [reference] } = await Webhooks.ensureQueueEvents(db, [{ envelope, subscription }]) // prettier-ignore

      vi.setSystemTime(new Date('2026-01-01T00:01:00.000Z'))

      await expect(Webhooks.deliverQueueEventAndRecord(db, reference!)).resolves
        .toMatchInlineSnapshot(`
        {
          "status": "skipped",
        }
      `)
    } finally {
      vi.useRealTimers()
    }
  })

  test('stages once and keeps the first envelope', async () => {
    const db = TestApp.database()
    const { created, envelope, reference, subscription } = await stagedObligation(db)
    const replacement = { ...envelope, data: { amount: '2' } }

    const second = await Webhooks.ensureQueueEvents(db, [{ envelope: replacement, subscription }])

    expect(created).toBe(1)
    expect(second.created).toBe(0)
    expect((await Webhooks.getQueueEvent(db, reference))?.envelope).toEqual(envelope)
  })

  test('never resurrects a terminal obligation', async () => {
    const db = TestApp.database()
    const { envelope, reference, subscription } = await stagedObligation(db)
    const claim = await Webhooks.claimQueueEvent(db, reference)
    expect(claim.type).toBe('claimed')
    if (claim.type !== 'claimed') return
    expect(
      await Webhooks.completeQueueEvent(db, reference, {
        claimedAt: claim.record.attemptingAt!,
        status: 'succeeded',
      }),
    ).toBe(true)

    const again = await Webhooks.ensureQueueEvents(db, [{ envelope, subscription }])
    expect(again.created).toBe(0)
    await expect(Webhooks.claimQueueEvent(db, reference)).resolves.toEqual({
      status: 'succeeded',
      type: 'terminal',
    })
    // Terminal rows read as absent, so no consumer path can redeliver them.
    expect(await Webhooks.getQueueEvent(db, reference)).toBeNull()
  })

  test('stages a replay once its completion has expired', async () => {
    vi.useFakeTimers({ toFake: ['Date'] })
    vi.setSystemTime(fixedNow())
    try {
      const db = TestApp.database()
      const { envelope, reference, subscription } = await stagedObligation(db)
      const claim = await Webhooks.claimQueueEvent(db, reference)
      expect(claim.type).toBe('claimed')
      if (claim.type !== 'claimed') return
      expect(
        await Webhooks.completeQueueEvent(db, reference, {
          claimedAt: claim.record.attemptingAt!,
          status: 'succeeded',
        }),
      ).toBe(true)

      // Past the dedupe window an unpruned completion must not suppress a
      // replay.
      vi.setSystemTime(new Date(fixedNow().getTime() + Webhooks.dedupeRetentionMs + 1_000))
      const replay = await Webhooks.ensureQueueEvents(db, [{ envelope, subscription }])
      expect(replay.created).toBe(1)
      await expect(Webhooks.claimQueueEvent(db, reference)).resolves.toMatchObject({
        type: 'claimed',
      })
    } finally {
      vi.useRealTimers()
    }
  })

  test('stamps a fresh observation on a post-expiry replay', async () => {
    vi.useFakeTimers({ toFake: ['Date'] })
    vi.setSystemTime(fixedNow())
    try {
      const db = TestApp.database()
      const { envelope, reference, subscription } = await stagedObligation(db)
      const claim = await Webhooks.claimQueueEvent(db, reference)
      if (claim.type !== 'claimed') throw new Error(`Unexpected claim: ${claim.type}`)
      await Webhooks.completeQueueEvent(db, reference, {
        claimedAt: claim.record.attemptingAt!,
        status: 'succeeded',
      })

      vi.setSystemTime(new Date(fixedNow().getTime() + Webhooks.dedupeRetentionMs + 1_000))
      const observedAt = Date.now() - 250
      await Webhooks.ensureQueueEvents(db, [{ envelope, subscription }], { observedAt })

      // Carrying the replaced row's stamp would measure this delivery against
      // an observation from the retention window before it.
      const replayed = await Webhooks.claimQueueEvent(db, reference)
      if (replayed.type !== 'claimed') throw new Error(`Unexpected claim: ${replayed.type}`)
      expect(replayed.record.observedAt).toBe(new Date(observedAt).toISOString())
    } finally {
      vi.useRealTimers()
    }
  })

  test('suppresses only the completed obligation when staging a mixed batch', async () => {
    const db = TestApp.database()
    const done = await stagedObligation(db, 4)
    const claim = await Webhooks.claimQueueEvent(db, done.reference)
    if (claim.type !== 'claimed') throw new Error(`Unexpected claim: ${claim.type}`)
    await Webhooks.completeQueueEvent(db, done.reference, {
      claimedAt: claim.record.attemptingAt!,
      status: 'failed',
    })

    const fresh = Webhooks.buildEnvelope({
      blockNumber: 124,
      createdAt: new Date(),
      data: { amount: '1' },
      logIndex: 5,
      subscription: done.subscription,
    })
    const batch = await Webhooks.ensureQueueEvents(db, [
      { envelope: done.envelope, subscription: done.subscription },
      { envelope: fresh, subscription: done.subscription },
    ])

    expect(batch.created).toBe(1)
    await expect(Webhooks.claimQueueEvent(db, done.reference)).resolves.toEqual({
      status: 'failed',
      type: 'terminal',
    })
    await expect(Webhooks.claimQueueEvent(db, batch.references[1]!)).resolves.toMatchObject({
      type: 'claimed',
    })
  })
})

describe('claimQueueEvent', () => {
  beforeEach(() => {
    vi.useFakeTimers({ toFake: ['Date'] })
    vi.setSystemTime(fixedNow())
  })

  afterEach(() => {
    vi.useRealTimers()
  })

  test('reads an expired completion as missing', async () => {
    const db = TestApp.database()
    const { reference } = await stagedObligation(db)
    const claim = await Webhooks.claimQueueEvent(db, reference)
    if (claim.type !== 'claimed') throw new Error(`Unexpected claim: ${claim.type}`)
    await Webhooks.completeQueueEvent(db, reference, {
      claimedAt: claim.record.attemptingAt!,
      status: 'succeeded',
    })

    // Matches a pruned terminal row: past the dedupe window the completion is
    // invisible everywhere, not only to staging.
    vi.setSystemTime(new Date(fixedNow().getTime() + Webhooks.dedupeRetentionMs + 1_000))
    await expect(Webhooks.claimQueueEvent(db, reference)).resolves.toEqual({ type: 'missing' })
  })

  test('claims a due obligation once and reclaims after staleness', async () => {
    const db = TestApp.database()
    const { reference } = await stagedObligation(db)

    const first = await Webhooks.claimQueueEvent(db, reference)
    expect(first.type).toBe('claimed')
    if (first.type !== 'claimed') return
    expect(first.record.attemptCount).toBe(1)

    await expect(Webhooks.claimQueueEvent(db, reference)).resolves.toEqual({
      type: 'attempting',
    })

    vi.setSystemTime(new Date(fixedNow().getTime() + Webhooks.deliveryAttemptStaleMs + 1))
    const reclaimed = await Webhooks.claimQueueEvent(db, reference)
    expect(reclaimed.type).toBe('claimed')
    if (reclaimed.type !== 'claimed') return
    expect(reclaimed.record.attemptCount).toBe(2)
  })

  test('classifies missing, scheduled, and terminal obligations', async () => {
    const db = TestApp.database()
    const { reference } = await stagedObligation(db)

    await expect(
      Webhooks.claimQueueEvent(db, { eventId: 'evt_absent', subscriptionId: 'wh_absent' }),
    ).resolves.toEqual({ type: 'missing' })

    const claim = await Webhooks.claimQueueEvent(db, reference)
    expect(claim.type).toBe('claimed')
    if (claim.type !== 'claimed') return
    expect(
      await Webhooks.scheduleQueueEventRetry(db, reference, {
        attemptCount: 1,
        claimedAt: claim.record.attemptingAt!,
      }),
    ).toBe(true)
    await expect(Webhooks.claimQueueEvent(db, reference)).resolves.toEqual({
      nextAttemptAt: new Date(fixedNow().getTime() + Webhooks.retryDelayMs(1)).toISOString(),
      type: 'scheduled',
    })

    vi.setSystemTime(new Date(fixedNow().getTime() + Webhooks.retryDelayMs(1) + 1))
    const retried = await Webhooks.claimQueueEvent(db, reference)
    expect(retried.type).toBe('claimed')
    if (retried.type !== 'claimed') return
    expect(
      await Webhooks.completeQueueEvent(db, reference, {
        claimedAt: retried.record.attemptingAt!,
        status: 'failed',
      }),
    ).toBe(true)
    await expect(Webhooks.claimQueueEvent(db, reference)).resolves.toEqual({
      status: 'failed',
      type: 'terminal',
    })
  })

  test('keeps owed work claimable indefinitely', async () => {
    const db = TestApp.database()
    const { reference } = await stagedObligation(db)

    // Pending obligations never expire; only terminal transitions stamp a
    // dedupe deadline.
    vi.setSystemTime(new Date(fixedNow().getTime() + Webhooks.dedupeRetentionMs * 3))
    await expect(Webhooks.claimQueueEvent(db, reference)).resolves.toMatchObject({
      type: 'claimed',
    })
  })

  test('leaves pending obligations claimable across a prune', async () => {
    const db = TestApp.database()
    const { reference } = await stagedObligation(db)

    vi.setSystemTime(new Date(fixedNow().getTime() + Webhooks.dedupeRetentionMs + 1_000))
    await Webhooks.pruneExpired(db)
    const claim = await Webhooks.claimQueueEvent(db, reference)
    expect(claim.type).toBe('claimed')
    if (claim.type !== 'claimed') return
    expect(
      await Webhooks.completeQueueEvent(db, reference, {
        claimedAt: claim.record.attemptingAt!,
        status: 'succeeded',
      }),
    ).toBe(true)
  })

  test('fences a stale claimant after a reclaim', async () => {
    const db = TestApp.database()
    const { reference } = await stagedObligation(db)
    const stale = await Webhooks.claimQueueEvent(db, reference)
    expect(stale.type).toBe('claimed')
    if (stale.type !== 'claimed') return

    vi.setSystemTime(new Date(fixedNow().getTime() + Webhooks.deliveryAttemptStaleMs + 1))
    const current = await Webhooks.claimQueueEvent(db, reference)
    expect(current.type).toBe('claimed')
    if (current.type !== 'claimed') return

    const staleClaimedAt = stale.record.attemptingAt!
    expect(
      await Webhooks.completeQueueEvent(db, reference, {
        claimedAt: staleClaimedAt,
        status: 'failed',
      }),
    ).toBe(false)
    expect(
      await Webhooks.scheduleQueueEventRetry(db, reference, {
        attemptCount: 1,
        claimedAt: staleClaimedAt,
      }),
    ).toBe(false)
    expect(
      await Webhooks.completeQueueEvent(db, reference, {
        claimedAt: current.record.attemptingAt!,
        status: 'succeeded',
      }),
    ).toBe(true)
  })
})

describe('retryDelayMs', () => {
  test('backs off exponentially and caps hourly', () => {
    expect([1, 2, 3, 4, 5, 6, 10].map(Webhooks.retryDelayMs)).toMatchInlineSnapshot(`
      [
        5000,
        25000,
        125000,
        625000,
        3125000,
        3600000,
        3600000,
      ]
    `)
  })
})

describe('cursorBlock', () => {
  test('decodes one- and two-part cursors and rejects the rest', () => {
    expect(Webhooks.cursorBlock(null)).toBeUndefined()
    expect(Webhooks.cursorBlock('not-a-cursor')).toBeUndefined()
    expect(Webhooks.cursorBlock(Cursor.encode([123]))).toBe(123)
    expect(Webhooks.cursorBlock(Cursor.encode([456, 7]))).toBe(456)
  })
})

describe('dueQueueEvents', () => {
  beforeEach(() => {
    vi.useFakeTimers({ toFake: ['Date'] })
    vi.setSystemTime(fixedNow())
  })

  afterEach(() => {
    vi.useRealTimers()
  })

  test('lists due pending work and skips in-flight, scheduled, and terminal rows', async () => {
    const db = TestApp.database()
    // Distinct staging times so the due order (next_attempt_at) is deterministic.
    const due = await stagedObligation(db, 1)
    vi.setSystemTime(new Date(fixedNow().getTime() + 1))
    const inFlight = await stagedObligation(db, 2)
    vi.setSystemTime(new Date(fixedNow().getTime() + 2))
    const scheduled = await stagedObligation(db, 3)
    vi.setSystemTime(new Date(fixedNow().getTime() + 3))
    const terminal = await stagedObligation(db, 4)

    expect((await Webhooks.claimQueueEvent(db, inFlight.reference)).type).toBe('claimed')
    const scheduledClaim = await Webhooks.claimQueueEvent(db, scheduled.reference)
    expect(scheduledClaim.type).toBe('claimed')
    if (scheduledClaim.type !== 'claimed') return
    expect(
      await Webhooks.scheduleQueueEventRetry(db, scheduled.reference, {
        attemptCount: 3,
        claimedAt: scheduledClaim.record.attemptingAt!,
      }),
    ).toBe(true)
    const terminalClaim = await Webhooks.claimQueueEvent(db, terminal.reference)
    expect(terminalClaim.type).toBe('claimed')
    if (terminalClaim.type !== 'claimed') return
    expect(
      await Webhooks.completeQueueEvent(db, terminal.reference, {
        claimedAt: terminalClaim.record.attemptingAt!,
        status: 'succeeded',
      }),
    ).toBe(true)

    await expect(Webhooks.dueQueueEvents(db, 10)).resolves.toEqual([due.reference])

    vi.setSystemTime(new Date(fixedNow().getTime() + Webhooks.deliveryAttemptStaleMs + 10))
    await expect(Webhooks.dueQueueEvents(db, 10)).resolves.toEqual([
      due.reference,
      inFlight.reference,
    ])
  })
})

describe('pendingQueueEvents', () => {
  beforeEach(() => {
    vi.useFakeTimers({ toFake: ['Date'] })
    vi.setSystemTime(fixedNow())
  })

  afterEach(() => {
    vi.useRealTimers()
  })

  test('counts pending obligations and reports the oldest staging time', async () => {
    const db = TestApp.database()
    await stagedObligation(db, 1)
    vi.setSystemTime(new Date(fixedNow().getTime() + 1_000))
    const done = await stagedObligation(db, 2)
    const doneClaim = await Webhooks.claimQueueEvent(db, done.reference)
    expect(doneClaim.type).toBe('claimed')
    if (doneClaim.type !== 'claimed') return
    expect(
      await Webhooks.completeQueueEvent(db, done.reference, {
        claimedAt: doneClaim.record.attemptingAt!,
        status: 'succeeded',
      }),
    ).toBe(true)

    await expect(Webhooks.pendingQueueEvents(db)).resolves.toEqual({
      count: 1,
      dueCount: 1,
      oldestAt: fixedNow().toISOString(),
      oldestDueAt: fixedNow().toISOString(),
    })
  })

  test('excludes in-flight claims and scheduled retries from due work', async () => {
    const db = TestApp.database()
    const inFlight = await stagedObligation(db, 1)
    const claim = await Webhooks.claimQueueEvent(db, inFlight.reference)
    expect(claim.type).toBe('claimed')

    await expect(Webhooks.pendingQueueEvents(db)).resolves.toMatchObject({
      count: 1,
      dueCount: 0,
      oldestDueAt: null,
    })
  })
})

describe('deliverClaimedAndRecord', () => {
  test('delivers a claimed envelope and skips an inactive subscription', async () => {
    const db = TestApp.database()
    const { envelope, reference, subscription } = await stagedObligation(db)
    const claim = await Webhooks.claimQueueEvent(db, reference)
    expect(claim.type).toBe('claimed')
    if (claim.type !== 'claimed') return

    let delivered: unknown
    const fetch = (async (_url, init) => {
      delivered = JSON.parse(typeof init?.body === 'string' ? init.body : '{}')
      return new Response(null, { status: 200 })
    }) satisfies typeof globalThis.fetch

    const delivery = await Webhooks.deliverClaimedAndRecord(
      db,
      { envelope: claim.record.envelope, subscriptionId: reference.subscriptionId },
      { fetch },
    )
    expect(delivery.status).toBe('delivered')
    if (delivery.status !== 'delivered') return
    expect(delivery.result.ok).toBe(true)
    expect(delivered).toEqual(envelope)

    await Webhooks.deleteSubscription(db, apiKeyOwner, subscription.id)
    await expect(
      Webhooks.deliverClaimedAndRecord(db, {
        envelope: claim.record.envelope,
        subscriptionId: reference.subscriptionId,
      }),
    ).resolves.toEqual({ status: 'skipped' })
  })
})

describe('buildPingEnvelope', () => {
  test('builds a synthetic ping envelope with a deterministic id from the nonce', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())
    const envelope = Webhooks.buildPingEnvelope(created, {
      createdAt: fixedNow(),
      nonce: 'test-nonce',
    })
    // The id hashes the (random) subscription id, so assert its shape and redact.
    expect(envelope.id).toMatch(/^evt_[0-9a-f]{64}$/)
    expect({ ...envelope, id: '<id>', subscriptionId: '<subscriptionId>' }).toMatchInlineSnapshot(`
      {
        "chainId": 4217,
        "createdAt": "2026-01-01T00:00:00.000Z",
        "data": {
          "ping": true,
        },
        "id": "<id>",
        "subscriptionId": "<subscriptionId>",
        "type": "ping",
      }
    `)
  })

  test('randomizes the id per call so repeated pings are not deduped', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())
    const a = Webhooks.buildPingEnvelope(created)
    const b = Webhooks.buildPingEnvelope(created)
    expect(a.id).not.toBe(b.id)
  })
})

describe('ping', () => {
  test('delivers, records a log row, and leaves the subscription state untouched', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())

    const captured: { body: string; signature: string | null } = { body: '', signature: null }
    const fetch = (async (_url, init) => {
      captured.body = typeof init?.body === 'string' ? init.body : ''
      captured.signature = new Headers(init?.headers).get(Webhooks.signatureHeader)
      return new Response(null, { status: 200 })
    }) satisfies typeof globalThis.fetch

    const { envelope, result } = await Webhooks.ping(db, created, {
      fetch,
      nonce: 'test-nonce',
      now: fixedNow,
    })

    expect(result.ok).toBe(true)
    // The delivered body is signed with the subscription secret.
    expect(
      Webhooks.verify({
        body: captured.body,
        now: 0,
        secret: created.secret,
        signature: captured.signature ?? '',
        tolerance: Number.POSITIVE_INFINITY,
      }),
    ).toBe(true)

    // A ping is observable in the delivery log.
    const [delivery] = await Webhooks.listDeliveries(db, created.id)
    expect(delivery?.eventId).toBe(envelope.id)
    expect(delivery?.status).toMatchInlineSnapshot(`"succeeded"`)

    // A ping never mutates the subscription's failure/lifecycle state.
    const after = await Webhooks.getSubscription(db, apiKeyOwner, created.id)
    expect(after?.failureCount).toMatchInlineSnapshot(`0`)
    expect(after?.status).toMatchInlineSnapshot(`"active"`)
    expect(after?.lastDeliveryAt).toMatchInlineSnapshot(`undefined`)
  })

  test('a failed ping records a failed row without disabling the subscription', async () => {
    const db = TestApp.database()
    const created = await Webhooks.createSubscription(db, createInput())

    const fetch = (async () =>
      new Response(null, { status: 500 })) satisfies typeof globalThis.fetch
    const { result } = await Webhooks.ping(db, created, { fetch, now: fixedNow })

    expect(result.ok).toBe(false)
    const [delivery] = await Webhooks.listDeliveries(db, created.id)
    expect(delivery?.status).toMatchInlineSnapshot(`"failed"`)

    // Crucially, a failed test ping must NOT count toward auto-disable.
    const after = await Webhooks.getSubscription(db, apiKeyOwner, created.id)
    expect(after?.failureCount).toMatchInlineSnapshot(`0`)
    expect(after?.status).toMatchInlineSnapshot(`"active"`)
  })
})