import assert from 'node:assert/strict'
import { randomUUID } from 'node:crypto'
import { readFileSync, writeFileSync } from 'node:fs'
import { resolve, basename } from 'node:path'
import { spawn } from 'node:child_process'
import { parseArgs } from 'node:util'
import dotenv from 'dotenv'
import { createServiceRoleClient } from '../../src/lib/supabase/service-client'
import { invoiceFixture, providerPage, type LoadProvider } from './fixtures'

const PROJECT = 'metjnjrhvujscngnpzdv'
const { values } = parseArgs({ options: {
  env: { type: 'string', default: '.env.provider-load.local' },
  provider: { type: 'string', default: 'both' }, count: { type: 'string', default: '25000' },
  budget: { type: 'string', default: '60000' }, 'detail-every': { type: 'string', default: '0' },
  'delay-ms': { type: 'string', default: '0' }, report: { type: 'string', default: '.env.provider-load-report.json' },
  child: { type: 'boolean', default: false }, job: { type: 'string' }, fault: { type: 'string' },
  'rate-limit': { type: 'boolean', default: false },
} })
const count = Number(values.count), budget = Number(values.budget)
const detailEvery = Number(values['detail-every']), delay = Number(values['delay-ms'])
assert.ok(Number.isInteger(count) && count >= 30 && count <= 100000)
assert.ok(Number.isInteger(budget) && budget >= 15000 && budget <= 210000)
assert.ok(Number.isInteger(detailEvery) && detailEvery >= 0)
assert.ok(Number.isInteger(delay) && delay >= 0 && delay <= 10000)
assert.ok(['visma', 'bokio', 'both'].includes(values.provider!))
assert.notEqual(basename(values.env!), '.env.local', 'Never load the production env file')
const config = dotenv.parse(readFileSync(resolve(values.env!)))
const url = config.NEXT_PUBLIC_SUPABASE_URL ?? config.SUPABASE_URL
const key = config.SUPABASE_SERVICE_ROLE_KEY ?? config.SERVICE_ROLE_KEY
assert.equal(url, `https://${PROJECT}.supabase.co`, 'This harness only writes to erp-base staging')
assert.ok(key, 'A staging service-role key is required; never paste it into logs')
// Deliberately copy only these settings, never mail, analytics, or Redis credentials.
process.env.NEXT_PUBLIC_SUPABASE_URL = url
process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY = config.NEXT_PUBLIC_SUPABASE_ANON_KEY ?? config.SUPABASE_ANON_KEY ?? 'synthetic-unused-anon-key'
process.env.SUPABASE_SERVICE_ROLE_KEY = key
process.env.PERSONNUMMER_ENCRYPTION_KEY = 'synthetic-provider-load-test-only'
for (const name of ['UPSTASH_REDIS_REST_URL', 'UPSTASH_REDIS_REST_TOKEN', 'KV_REST_API_URL', 'KV_REST_API_TOKEN',
  'UPSTASH_STORAGE_KV_REST_API_URL', 'UPSTASH_STORAGE_KV_REST_API_TOKEN']) delete process.env[name]
Object.assign(process.env, { NODE_ENV: 'test' })
const originalFetch = globalThis.fetch
const client = () => createServiceRoleClient(url, key, { auth: { persistSession: false, autoRefreshToken: false },
  global: { fetch: (input, init) => originalFetch(input, { ...init, signal: AbortSignal.timeout(30000) }) } })
const db = client()
async function checked<T>(query: PromiseLike<{ data: T; error: { message: string } | null }>): Promise<T> {
  const result = await query
  if (result.error) throw new Error(result.error.message)
  return result.data
}
const sleep = (ms: number) => new Promise<void>(done => setTimeout(done, ms))

async function child() {
  assert.ok(values.job)
  const provider = values.provider as LoadProvider
  assert.notEqual(provider, 'both')
  let providerRequests = 0, detailRequests = 0, maxDatabaseMs = 0, largestWriteBytes = 0
  let injected = false, rateLimited = false, leaseChecked = false
  globalThis.fetch = async (input, init) => {
    const address = new URL(typeof input === 'string' ? input : input instanceof URL ? input.href : input.url)
    if (address.hostname === `${PROJECT}.supabase.co`) {
      if (init?.body && typeof init.body === 'string') largestWriteBytes = Math.max(largestWriteBytes, Buffer.byteLength(init.body))
      const start = performance.now()
      const response = await originalFetch(input, init)
      maxDatabaseMs = Math.max(maxDatabaseMs, performance.now() - start)
      if (address.pathname.endsWith('/claim_provider_migration_job') && response.ok && !leaseChecked) {
        const claimed = await response.clone().json()
        if (claimed?.id) {
          const other = await checked(db.rpc('claim_provider_migration_job', { p_job_id: values.job, p_worker_id: randomUUID() }))
          assert.ok(!other?.id, 'Two workers claimed the same active lease')
          leaseChecked = true
        }
      }
      if (!injected && response.ok && address.pathname.endsWith('/commit_provider_migration_records')) {
        injected = true
        if (values.fault === 'crash') {
          process.stdout.write('LOAD_CRASH_AFTER_COMMIT ' + JSON.stringify({ rateLimited }) + '\n', () => process.kill(process.pid, 'SIGKILL'))
          return new Promise<Response>(() => {})
        }
        if (values.fault === 'lost-ack') return Response.json({ message: 'Synthetic lost commit acknowledgement' }, { status: 503 })
      }
      return response
    }
    const host = provider === 'visma' ? 'eaccountingapi.vismaonline.com' : 'api.bokio.se'
    assert.equal(address.hostname, host, 'Unexpected outbound request blocked')
    assert.equal(new Headers(init?.headers).get('Authorization'), 'Bearer synthetic-load-token')
    providerRequests++
    if (delay) await sleep(delay)
    if (values['rate-limit'] && !rateLimited) {
      rateLimited = true
      return Response.json({ message: 'Synthetic rate limit' }, { status: 429, headers: { 'Retry-After': '1' } })
    }
    const detail = address.pathname.match(/\/load-(\d+)$/)
    if (detail) { detailRequests++; return Response.json(invoiceFixture(provider, Number(detail[1]))) }
    assert.ok(address.pathname.endsWith(provider === 'visma' ? '/customerinvoices' : '/invoices'))
    const page = Number(address.searchParams.get(provider === 'visma' ? '$page' : 'page'))
    const size = Number(address.searchParams.get(provider === 'visma' ? '$pagesize' : 'pageSize'))
    assert.ok(page >= 1 && size > 0)
    return Response.json(providerPage(provider, page, size, count, detailEvery))
  }
  const { runProviderMigrationWorker } = await import('../../src/extensions/general/arcim-migration/lib/migration-job-worker')
  const start = performance.now()
  const result = await runProviderMigrationWorker({ jobId: values.job, budgetMs: budget })
  const elapsedMs = Math.round(performance.now() - start)
  assert.ok(elapsedMs <= budget + 1500, `Worker exceeded ${budget}ms budget: ${elapsedMs}ms`)
  assert.ok(largestWriteBytes <= 760000, `Oversized write: ${largestWriteBytes} bytes`)
  console.log('LOAD_RESULT ' + JSON.stringify({ ...result, elapsedMs, providerRequests, detailRequests,
    maxDatabaseMs: Math.round(maxDatabaseMs), largestWriteBytes, rssBytes: process.memoryUsage().rss,
    leaseChecked, injected, rateLimited }))
}

async function runChild(provider: LoadProvider, job: string, fault?: string, rateLimit = false) {
  const args = ['--import', 'tsx', '--conditions', 'react-server', resolve('scripts/provider-migration/load-staging.ts'),
    '--child', '--env', resolve(values.env!), '--provider', provider, '--count', String(count), '--budget', String(budget),
    '--detail-every', String(detailEvery), '--delay-ms', String(delay), '--job', job]
  if (fault) args.push('--fault', fault)
  if (rateLimit) args.push('--rate-limit')
  return new Promise<Record<string, number | boolean>>((done, fail) => {
    const worker = spawn(process.execPath, args, { stdio: ['ignore', 'pipe', 'pipe'] })
    let output = '', errors = ''
    worker.stdout.on('data', data => { output = (output + data.toString()).slice(-100000) })
    worker.stderr.on('data', data => { errors = (errors + data.toString()).slice(-10000) })
    const watchdog = setTimeout(() => { worker.kill('SIGKILL'); fail(new Error('Worker watchdog exceeded budget')) }, budget + 15000)
    worker.on('error', error => { clearTimeout(watchdog); fail(error) })
    worker.on('exit', (code, signal) => {
      clearTimeout(watchdog)
      if (signal === 'SIGKILL' && fault === 'crash' && output.includes('LOAD_CRASH_AFTER_COMMIT')) return done({ crashed: true, rateLimited: output.includes('LOAD_CRASH_AFTER_COMMIT {"rateLimited":true}') })
      const report = output.split('\n').find(line => line.startsWith('LOAD_RESULT '))
      if (code !== 0 || !report) return fail(new Error(`Worker failed (${code}/${signal}): ${errors || output}`))
      done(JSON.parse(report.slice('LOAD_RESULT '.length)))
    })
  })
}

async function main() {
  if (values.child) return child()
  const reports: Record<string, unknown>[] = []
  const providers: LoadProvider[] = values.provider === 'both' ? ['visma', 'bokio'] : [values.provider as LoadProvider]
  try {
    for (const provider of providers) {
      const start = performance.now()
      const company = randomUUID(), consent = randomUUID()
      const auth = await db.auth.admin.createUser({ email: `provider-load-${company}@test.invalid`, email_confirm: true })
      if (auth.error) throw new Error(auth.error.message)
      const user = auth.data.user.id
      await checked(db.from('companies').insert({ id: company, name: `SYNTHETIC provider load ${provider} ${count}`, entity_type: 'aktiebolag', created_by: user }))
      await checked(db.from('company_members').insert({ company_id: company, user_id: user, role: 'owner' }))
      await checked(db.from('provider_consents').insert({ id: consent, company_id: company, name: 'Synthetic load test', provider, org_number: '556000-0000' }))
      await checked(db.from('provider_consent_tokens').insert({ consent_id: consent, provider, access_token: 'synthetic-load-token', provider_company_id: 'synthetic-company' }))
      // Admission fixture only. This benchmark imports invoice registers, not a ledger.
      await checked(db.from('sie_imports').insert({ company_id: company, user_id: user, filename: 'synthetic-load-precondition.se', file_hash: company, sie_type: 4, status: 'completed' }))
      const job = await checked(db.rpc('create_provider_migration_job', { p_company_id: company, p_user_id: user,
        p_consent_id: consent, p_resources: ['salesInvoices'], p_scope: null }))
      const report = { provider, company, job: job.id, user, count, detailEvery, delay, budget, invocations: [] as Record<string, number | boolean>[], state: 'running', elapsedMs: 0 }
      reports.push(report)
      console.log(`${provider}: ${count} invoices; synthetic company ${company}; job ${job.id}`)
      let crashDone = false, lostAckDone = false, rateLimitDone = false
      for (let attempt = 0; attempt < 2000; attempt++) {
        const row = await checked(db.from('migration_jobs').select('*').eq('id', job.id).eq('company_id', company).single())
        if (row.state === 'completed') { report.state = row.state; break }
        assert.notEqual(row.state, 'needs_attention', `Unexpected review state: ${row.error_code}`)
        // Accelerate only this synthetic job's lease/backoff after the forced crash.
        if (row.state === 'running' || row.state === 'retry_wait') await checked(db.from('migration_jobs')
          .update({ lease_until: new Date(0).toISOString(), next_attempt_at: new Date(0).toISOString() })
          .eq('id', job.id).eq('company_id', company))
        const fault: string | undefined = !crashDone ? 'crash' : !lostAckDone ? 'lost-ack' : undefined
        const result = await runChild(provider, job.id, fault, !rateLimitDone)
        rateLimitDone ||= result.rateLimited === true
        crashDone ||= result.crashed === true
        lostAckDone ||= fault === 'lost-ack' && result.injected === true
        report.invocations.push(result)
        const counts = await checked(db.rpc('provider_migration_counts', { p_job_id: job.id }))
        console.log(`${provider} run ${attempt + 1}: ${JSON.stringify({ ...result, counts })}`)
        writeFileSync(values.report!, JSON.stringify({ project: PROJECT, syntheticProviderResponses: true, reports }, null, 2))
      }
      assert.equal(report.state, 'completed')
      assert.ok(crashDone && lostAckDone && rateLimitDone, 'All recovery faults must be exercised; use more invoices or a shorter budget')
      for (const [table, expected] of [['invoices', count], ['customers', Math.min(count, 250)]] as const) {
        const result = await db.from(table).select('id', { head: true, count: 'exact' }).eq('company_id', company)
        assert.equal(result.error, null); assert.equal(result.count, expected, `${table} count`)
      }
      const lineCount = await db.from('invoice_items').select('id,invoices!inner(company_id)', { head: true, count: 'exact' }).eq('invoices.company_id', company)
      assert.equal(lineCount.error, null); assert.equal(lineCount.count, count * 3)
      const completions = await db.from('processing_history').select('event_id', { head: true, count: 'exact' })
        .eq('company_id', company).eq('event_type', 'InvoiceRowsCompleted')
      assert.equal(completions.error, null); assert.equal(completions.count, count, 'Duplicate row-completion events')
      const receipts = (await checked(db.rpc('provider_migration_counts', { p_job_id: job.id })))[0]
      assert.equal(Number(receipts.total), count)
      assert.equal(Number(receipts.completed), count)
      assert.equal(Number(receipts.needs_attention), 0)
      let totalOre = 0, seen = 0
      for (let offset = 0; offset < count; offset += 1000) {
        const rows = await checked(db.from('invoices').select('total,subtotal,vat_amount').eq('company_id', company).order('id').range(offset, offset + 999))
        for (const row of rows ?? []) { assert.equal(Number(row.subtotal), 300); assert.equal(Number(row.vat_amount), 75); totalOre += Math.round(Number(row.total) * 100); seen++ }
      }
      assert.equal(seen, count); assert.equal(totalOre, count * 37500)
      report.elapsedMs = Math.round(performance.now() - start)
      console.log(`${provider}: PASS, ${count} invoices, ${count * 3} lines, no duplicates, exact totals, ${report.elapsedMs}ms overall`)
    }
  } finally {
    writeFileSync(values.report!, JSON.stringify({ project: PROJECT, syntheticProviderResponses: true, reports }, null, 2))
  }
}
void main().catch(error => { console.error(error.message); process.exitCode = 1 })