import { CompiledQuery, sql } from 'kysely'
import { nanoid } from 'nanoid'
import * as pg from 'pg'

import * as Runtime from '../../test/runtime.js'
import * as TestRoutes from '../../test/Routes.js'
import * as Transfer from '../internal/routes/Transfer.js'
import * as RoutesTransferTransactions from './tables/routesTransferTransactions.js'
import * as RoutesTransfers from './tables/routesTransfers.js'
import * as RoutesTransferSubsidies from './tables/routesTransferSubsidies.js'
import * as Db from './Db.js'
import * as Webhooks from '../internal/Webhooks.js'
import * as Invitations from './tables/invitations.js'
import * as ApiKeyOwnerTombstones from './tables/apiKeyOwnerTombstones.js'
import * as Organizations from './tables/organizations.js'
import * as Projects from './tables/projects.js'
import * as WebhookSubscriptions from './tables/webhookSubscriptions.js'

const create = () => Db.postgres({ connectionString: Runtime.postgresUrl, schema: `t_${nanoid()}` })

describe('postgres', () => {
  test('behavior: migrate is idempotent', async () => {
    const db = create()
    await db.migrate()
    await db.migrate()
    expect(await Organizations.get(db, 'missing')).toBeUndefined()
    await db.close()
  })

  test('behavior: raw owner deletes fence API keys for older application code', async () => {
    const db = create()
    await db.migrate()
    await db.kysely
      .updateTable('api_key_owner_tombstones')
      .set({ createdAt: new Date(Date.now() - 25 * 60 * 60 * 1_000).toISOString() })
      .where('id', '=', 'routes-legacy-attempt-drain')
      .execute()
    const organization = await Organizations.create(db, { name: 'Old worker owner' })
    const project = await Projects.create(db, {
      name: 'Old worker project',
      orgId: organization.id,
    })

    await db.kysely.deleteFrom('projects').where('id', '=', project.id).execute()
    await db.kysely.deleteFrom('organizations').where('id', '=', organization.id).execute()

    await expect(
      ApiKeyOwnerTombstones.isDeleted(db, {
        orgId: organization.id,
        projectId: project.id,
      }),
    ).resolves.toBe(true)
    await expect(ApiKeyOwnerTombstones.isDeleted(db, { orgId: organization.id })).resolves.toBe(
      true,
    )
    await db.close()
  })

  test('behavior: the first Routes migration drains and fences rolling ownerless writers', async () => {
    const schema = `t_${nanoid()}`
    const db = Db.postgres({ connectionString: Runtime.postgresUrl })
    try {
      await db.kysely.executeQuery(
        CompiledQuery.raw(`CREATE SCHEMA ${pg.escapeIdentifier(schema)}`),
      )
      await db.transaction(async (tx) => {
        await sql`SELECT set_config('search_path', quote_ident(${schema}), true)`.execute(tx.kysely)
        for (const migration of Db.migrations) {
          if (migration.name === '0255_routes_organization_delete_fence') break
          await migration.up(tx.kysely).execute()
        }
        await Organizations.create(tx, { id: 'org_rollout', name: 'Rollout owner' })
        const migration = Db.migrations.find(
          (candidate) => candidate.name === '0255_routes_organization_delete_fence',
        )
        if (!migration) throw new Error('Expected the first Routes migration.')
        await migration.up(tx.kysely).execute()

        await sql`SAVEPOINT expected_rollout_delete_failure`.execute(tx.kysely)
        await expect(
          tx.kysely.deleteFrom('organizations').where('id', '=', 'org_rollout').execute(),
        ).rejects.toMatchObject({
          code: '23503',
          message: 'route lifecycle rollout drain is active',
        })
        await sql`ROLLBACK TO SAVEPOINT expected_rollout_delete_failure`.execute(tx.kysely)
        await tx.kysely
          .updateTable('api_key_owner_tombstones')
          .set({ createdAt: new Date(Date.now() - 25 * 60 * 60 * 1_000).toISOString() })
          .where('id', '=', 'routes-legacy-attempt-drain')
          .execute()
        await tx.kysely.deleteFrom('organizations').where('id', '=', 'org_rollout').execute()

        await sql`SAVEPOINT expected_ownerless_writer_failure`.execute(tx.kysely)
        await expect(
          sql`INSERT INTO routes_idempotency_requests (api_key_id, key_hash, request_hash, status, response, transfer_id, created_at, expires_at) VALUES ('key_rollout', 'hash_rollout', 'request_rollout', 'pending', NULL, NULL, '2026-09-04T00:00:00.000Z', '2099-09-04T00:00:00.000Z')`.execute(
            tx.kysely,
          ),
        ).rejects.toMatchObject({ code: '23503' })
        await sql`ROLLBACK TO SAVEPOINT expected_ownerless_writer_failure`.execute(tx.kysely)
      })
    } finally {
      try {
        await db.kysely.executeQuery(
          CompiledQuery.raw(`DROP SCHEMA IF EXISTS ${pg.escapeIdentifier(schema)} CASCADE`),
        )
      } finally {
        await db.close()
      }
    }
  })

  test('behavior: raw project deletes hold the application owner locks', async () => {
    const schema = `t_${nanoid()}`
    const first = Db.postgres({ connectionString: Runtime.postgresUrl, schema })
    const second = Db.postgres({ connectionString: Runtime.postgresUrl, schema })
    await first.migrate()
    const organization = await Organizations.create(first, { name: 'Concurrent owner' })
    const project = await Projects.create(first, {
      name: 'Concurrent project',
      orgId: organization.id,
    })
    const deleted = Promise.withResolvers<void>()
    const release = Promise.withResolvers<void>()
    const deletion = first.transaction(async (tx) => {
      await tx.kysely.deleteFrom('projects').where('id', '=', project.id).execute()
      deleted.resolve()
      await release.promise
    })
    await deleted.promise

    const active = second.transaction((tx) =>
      ApiKeyOwnerTombstones.lockActive(tx, {
        orgId: organization.id,
        projectId: project.id,
      }),
    )
    const state = await Promise.race([
      active.then(() => 'resolved' as const),
      new Promise<'blocked'>((resolve) => setTimeout(() => resolve('blocked'), 100)),
    ])
    release.resolve()
    expect(state).toBe('blocked')
    await expect(active).resolves.toBe(false)
    await deletion
    await Promise.all([first.close(), second.close()])
  })

  test('behavior: upgrades legacy source claims and captures old subsidy writers', async () => {
    const db = create()
    await db.migrate()
    try {
      await sql`DROP TRIGGER routes_transfer_subsidy_capture ON routes_transfers`.execute(db.kysely)
      await sql`DROP TABLE routes_transfer_subsidies`.execute(db.kysely)
      await sql`DROP TRIGGER routes_transfer_transaction_scope ON routes_transfer_transactions`.execute(
        db.kysely,
      )
      await sql`ALTER TABLE routes_transfer_transactions DROP COLUMN org_id CASCADE, DROP COLUMN environment, DROP COLUMN project_id`.execute(
        db.kysely,
      )
      await sql`CREATE UNIQUE INDEX routes_transfer_transactions_source_unique ON routes_transfer_transactions (chain_id, transaction_ref) WHERE role = 'source'`.execute(
        db.kysely,
      )
      const snapshot = TestRoutes.transferSnapshot({
        sourceTransactionHashes: [`0x${'ab'.repeat(32)}`],
      })
      const transaction = {
        account: `0x${'aa'.repeat(20)}`,
        amount: '1',
        chainId: 'eip155:4217',
        hash: `0x${'cd'.repeat(32)}`,
        nonce: 0,
        transaction: '0x01',
      } as const
      const amount = {
        baseUnits: '1',
        currency: 'USD',
        decimals: 6,
        formatted: '0.000001',
      } as const
      const owner = await Transfer.create(db, {
        apiKeyId: 'key_legacy',
        environment: 'production',
        orgId: 'org_legacy',
        snapshot,
      })
      await RoutesTransferTransactions.insert(db, {
        chainId: snapshot.sourceChain.id,
        createdAt: owner.createdAt,
        role: 'source',
        transactionRef: snapshot.sourceTransactionHashes![0]!,
        transferId: owner.id,
      })
      const prepared = await Transfer.transition(db, {
        expectedVersion: 1,
        id: owner.id,
        providerState: { transferSubsidy: transaction },
        status: 'processing',
        subsidyAccount: transaction.account,
        subsidyAmount: amount,
        subsidyChainId: transaction.chainId,
        subsidyTokenAddress: snapshot.destinationToken.address.toLowerCase(),
        subsidyTransactionHash: transaction.hash,
      })
      expect(prepared.type).toBe('applied')
      const migrations = Db.migrations.filter(({ name }) => /^023[678]_/.test(name))
      for (const migration of migrations) await migration.up(db.kysely).execute()
      for (const migration of migrations) await migration.up(db.kysely).execute()
      await sql`SET search_path TO public`.execute(db.kysely)
      const evidence = await RoutesTransferTransactions.getSource(db, {
        chainId: snapshot.sourceChain.id,
        transactionRef: snapshot.sourceTransactionHashes![0]!,
        transferId: owner.id,
      })
      expect(evidence).toMatchObject({
        environment: 'production',
        orgId: 'org_legacy',
        projectId: null,
        transferId: owner.id,
      })
      const settlement = await db.transaction((tx) =>
        RoutesTransferSubsidies.getForUpdateIn(tx, {
          sourceChainId: snapshot.sourceChain.id,
          transactionRef: snapshot.sourceTransactionHashes![0]!,
        }),
      )
      expect(settlement).toMatchObject({ transaction, transferId: owner.id })
      const foreign = await Transfer.create(db, {
        apiKeyId: 'key_foreign',
        environment: 'production',
        orgId: 'org_foreign',
        snapshot,
      })
      expect(
        (
          await RoutesTransferTransactions.claimSource(db, {
            chainId: snapshot.sourceChain.id,
            createdAt: foreign.createdAt,
            role: 'source',
            transactionRef: snapshot.sourceTransactionHashes![0]!,
            transferId: foreign.id,
          })
        ).type,
      ).toBe('claimed')
      // Simulate the previous Worker persisting another payout without the new settlement lookup.
      await expect(
        Transfer.transition(db, {
          expectedVersion: 1,
          id: foreign.id,
          providerState: { transferSubsidy: { ...transaction, hash: `0x${'ef'.repeat(32)}` } },
          status: 'processing',
          subsidyAccount: transaction.account,
          subsidyAmount: amount,
          subsidyChainId: transaction.chainId,
          subsidyTokenAddress: snapshot.destinationToken.address.toLowerCase(),
          subsidyTransactionHash: `0x${'ef'.repeat(32)}`,
        }),
      ).rejects.toThrow('Source payment already has a different subsidy settlement')
      expect((await RoutesTransfers.get(db, foreign.id))?.status).toBe('awaiting-source')
    } finally {
      await db.close()
    }
  })

  test('behavior: legacy invitation inserts do not grant early access', async () => {
    const db = create()
    await db.migrate()
    await Organizations.create(db, { id: 'org_legacy', name: 'Legacy' })
    await sql`
      INSERT INTO invitations (
        id, org_id, email, role, invited_by, expires_at, accepted_at, revoked_at, created_at
      ) VALUES (
        'inv_legacy', 'org_legacy', 'legacy@example.org', 'member', 'usr_legacy',
        '2027-01-01T00:00:00.000Z', NULL, NULL, '2026-07-20T00:00:00.000Z'
      )
    `.execute(db.kysely)

    expect((await Invitations.get(db, 'inv_legacy'))?.grantsEarlyAccess).toBe(false)
    await db.close()
  })

  test('behavior: raw webhook staging respects schema scoping', async () => {
    // Exercises the one raw statement against a schema-scoped db whose only
    // qualification is WithSchemaPlugin, matching preview Workers: migrate on
    // one pool (whose session gains a search_path), then query from a fresh
    // pool that never ran it.
    const schema = `t_${nanoid()}`
    const migrator = Db.postgres({ connectionString: Runtime.postgresUrl, schema })
    await migrator.migrate()
    await migrator.close()
    const db = Db.postgres({ connectionString: Runtime.postgresUrl, schema })
    const subscription = await Webhooks.createSubscription(db, {
      chainId: 4217,
      destination: { type: 'url', url: 'https://hooks.example.com/endpoint' },
      eventType: 'token:transfer',
      owner: { orgId: 'org_1', type: 'api_key' },
    })
    const envelope = Webhooks.buildEnvelope({
      blockNumber: 123,
      createdAt: new Date(),
      data: { amount: '1' },
      logIndex: 4,
      subscription,
    })

    const staged = await Webhooks.ensureQueueEvents(db, [{ envelope, subscription }])
    expect(staged.created).toBe(1)
    const claim = await Webhooks.claimQueueEvent(db, staged.references[0]!)
    if (claim.type !== 'claimed') throw new Error(`Unexpected claim: ${claim.type}`)
    await Webhooks.completeQueueEvent(db, staged.references[0]!, {
      claimedAt: claim.record.attemptingAt!,
      status: 'succeeded',
    })

    // The completion read inside the same raw statement must also resolve
    // inside the schema for the replay to be suppressed.
    const replay = await Webhooks.ensureQueueEvents(db, [{ envelope, subscription }])
    expect(replay.created).toBe(0)
    await db.close()
  })

  test('behavior: creates a Postgres db without schema scoping', async () => {
    const db = Db.postgres({ connectionString: Runtime.postgresUrl })
    await db.close()
  })

  test('behavior: defers schema creation until migration', async () => {
    const schema = `t_${nanoid()}`
    const db = Db.postgres({ connectionString: Runtime.postgresUrl, schema })
    const exists = async () =>
      (
        await sql<{
          exists: boolean
        }>`SELECT EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = ${schema}) AS exists`.execute(
          db.kysely,
        )
      ).rows[0]!.exists

    await sql`SELECT 1`.execute(db.kysely)
    expect(await exists()).toBe(false)
    await db.migrate()
    expect(await exists()).toBe(true)
    await db.close()
  })

  test('behavior: qualifies repository queries with the configured schema', async () => {
    const db = Db.postgres({
      connectionString: Runtime.postgresUrl,
      schema: `t_MixedCase_${nanoid()}`,
    })
    await sql`SET search_path TO public`.execute(db.kysely)
    await db.migrate()
    await sql`SET search_path TO public`.execute(db.kysely)
    await db.transaction(async (tx) => {
      await Organizations.create(tx, { id: 'org_qualified', name: 'Qualified' })
    })
    expect((await Organizations.get(db, 'org_qualified'))?.name).toBe('Qualified')
    expect(await WebhookSubscriptions.getCursor(db, 'wh_missing')).toBeNull()
    await db.close()
  })

  test('behavior: concurrent migrate tolerates the ledger race', async () => {
    const schema = `t_${nanoid()}`
    // Several runners migrate the same fresh schema at once (mirrors parallel
    // deploys). The session advisory lock serializes them; none throw.
    const dbs = Array.from({ length: 5 }, () =>
      Db.postgres({ connectionString: Runtime.postgresUrl, schema }),
    )
    await Promise.all(dbs.map((db) => db.migrate()))
    const created = await Organizations.create(dbs[0]!, { id: 'org_race', name: 'Race' })
    expect(created.id).toBe('org_race')
    await Promise.all(dbs.map((db) => db.close()))
  })

  test('error: rejects a modified applied migration', async () => {
    const db = create()
    await db.migrate()
    await db.kysely
      .updateTable('migrations')
      .set({ checksum: '00000000' })
      .where('name', '=', Db.migrations[0]!.name)
      .execute()
    await expect(db.migrate()).rejects.toThrow(/checksum mismatch/)
    await db.close()
  })

  test('behavior: transaction rolls back on throw', async () => {
    const db = create()
    await db.migrate()
    await expect(
      db.transaction(async (tx) => {
        await Organizations.create(tx, { id: 'org_2', name: 'Roll' })
        throw new Error('boom')
      }),
    ).rejects.toThrow('boom')
    expect(await Organizations.get(db, 'org_2')).toBeUndefined()
    await db.close()
  })

  test('error: transaction rejects nesting', async () => {
    const db = create()
    await db.migrate()
    await expect(db.transaction((tx) => tx.transaction(async () => 1))).rejects.toThrow(/nesting/)
    await db.close()
  })

  test('behavior: transaction commits and returns the value', async () => {
    const db = create()
    await db.migrate()
    const id = await db.transaction(
      async (tx) => (await Organizations.create(tx, { id: 'org_3', name: 'Commit' })).id,
    )
    expect(id).toBe('org_3')
    expect((await Organizations.get(db, 'org_3'))?.name).toBe('Commit')
    await db.close()
  })
})