import { sql, eq } from 'drizzle-orm'
import { it, expect, vi } from 'vitest'
import { createPgliteClient } from '../../db/src/postgres/pglite.js'
import { createJobRegistry } from './index.js'
import {
  dispatchOutboxRow,
  enqueueOutbox,
  insertOutboxRow,
  outboxSchema,
  outboxTable,
} from './outbox.js'

type Env = Record<string, never>

const CREATE_OUTBOX = sql`
  CREATE TABLE outbox (
    id uuid PRIMARY KEY,
    aggregate_type text NOT NULL,
    aggregate_id text NOT NULL,
    event_type text NOT NULL,
    payload jsonb NOT NULL,
    processed_at timestamptz,
    failed_at timestamptz,
    retry_count integer NOT NULL DEFAULT 0,
    last_error text,
    created_at timestamptz NOT NULL DEFAULT NOW()
  )
`

function outboxId(suffix: string): string {
  return `10000000-0000-4000-8000-${suffix.padStart(12, '0')}`
}

function okSchema() {
  return {
    '~standard': {
      version: 1 as const,
      vendor: 'test',
      validate: (value: unknown) => ({ value: value as { x: number } }),
    },
  }
}

async function setupDb() {
  const db = createPgliteClient({ schema: outboxSchema })
  await db.execute(CREATE_OUTBOX)
  return db
}

it('insertOutboxRow inside a rolled-back tx leaves no row', async () => {
  const db = await setupDb()
  await expect(
    db.transaction(async (tx) => {
      await insertOutboxRow(tx, {
        id: outboxId('1'),
        aggregateType: 'deal',
        aggregateId: 'd1',
        eventType: 'deal.created',
        payload: { x: 1 },
      })
      throw new Error('rollback')
    }),
  ).rejects.toThrow('rollback')

  const rows = await db.select().from(outboxTable)
  expect(rows).toHaveLength(0)
})

it('dispatchOutboxRow sets processed_at and runs the handler', async () => {
  const db = await setupDb()
  const handler = vi.fn()
  const registry = createJobRegistry<Env>()
  registry.register('evt', okSchema(), handler)

  const row = await insertOutboxRow(db, {
    id: outboxId('2'),
    aggregateType: 'deal',
    aggregateId: 'd2',
    eventType: 'evt',
    payload: { x: 9 },
  })

  const result = await dispatchOutboxRow(registry, db, row, {})
  expect(result).toBe('dispatched')
  expect(handler).toHaveBeenCalledWith({}, { x: 9 }, { type: 'evt', payload: { x: 9 } })

  const [updated] = await db.select().from(outboxTable).where(eq(outboxTable.id, row.id))
  expect(updated?.processedAt).toBeInstanceOf(Date)
})

it('re-dispatch of an already-processed row is skipped', async () => {
  const db = await setupDb()
  const handler = vi.fn()
  const registry = createJobRegistry<Env>()
  registry.register('evt', okSchema(), handler)

  const row = await insertOutboxRow(db, {
    id: outboxId('3'),
    aggregateType: 'deal',
    aggregateId: 'd3',
    eventType: 'evt',
    payload: { x: 1 },
    processedAt: new Date(),
  })

  const result = await dispatchOutboxRow(registry, db, row, {})
  expect(result).toBe('skipped')
  expect(handler).not.toHaveBeenCalled()
})

it('enqueueOutbox sends outboxId and never throws', async () => {
  const send = vi.fn(async () => {})
  enqueueOutbox({ send }, outboxId('4'))
  await new Promise((r) => setTimeout(r, 0))
  expect(send).toHaveBeenCalledWith({ outboxId: outboxId('4') })

  const failSend = vi.fn(async () => {
    throw new Error('queue down')
  })
  expect(() => enqueueOutbox({ send: failSend }, outboxId('5'))).not.toThrow()
  await new Promise((r) => setTimeout(r, 0))
})