import assert from 'node:assert/strict'
import { execFileSync } from 'node:child_process'
import { randomBytes } from 'node:crypto'
import { lstat, mkdtemp, rm } from 'node:fs/promises'
import { createServer } from 'node:net'
import { tmpdir } from 'node:os'
import { isAbsolute, join, resolve } from 'node:path'
import { fileURLToPath } from 'node:url'
import EmbeddedPostgres from 'embedded-postgres'
import pg from 'pg'
import { applySqlMigrations } from '../../src/db/migrate.ts'
import { MIGRATION_MANIFEST } from '../../src/db/migration-manifest.ts'

assert.deepEqual(process.argv.slice(2), ['--owned-embedded'])
const bin = process.env.MMAPP_PG_CLIENT_BIN
assert(bin && isAbsolute(bin) && resolve(bin) === bin, 'Set MMAPP_PG_CLIENT_BIN to the exact local PostgreSQL client directory')
const psql = join(bin, process.platform === 'win32' ? 'psql.exe' : 'psql')
const executable = await lstat(psql)
assert(executable.isFile() && !executable.isSymbolicLink(), 'PostgreSQL client must be a regular file')
const clientEnv = Object.fromEntries(Object.entries(process.env).filter(([name]) => !/^PG/i.test(name)))
const run = (command, args, options = {}) => execFileSync(command, args, {
  timeout: 10000, maxBuffer: 16384, encoding: 'utf8', windowsHide: true, ...options,
})
assert.match(run(psql, ['--version'], { env: clientEnv }), /^psql \(PostgreSQL\) (?:1[6-9]|[2-9][0-9])\./)
const witness = fileURLToPath(new URL('../../../tmp/mcp-netflow-linux-topology-20261104/collector-drain-witness.py', import.meta.url))
const { IDENTITY_SQL, SQL } = JSON.parse(run('python', ['-I', '-B', '-c',
  'import json,runpy,sys; m=runpy.run_path(sys.argv[1]); print(json.dumps({k:m[k] for k in ("IDENTITY_SQL","SQL")}))', witness]))
assert(typeof SQL === 'string' && typeof IDENTITY_SQL === 'string')

const directory = await mkdtemp(join(tmpdir(), 'mm-netflow-drain-pg-'))
const password = randomBytes(24).toString('hex')
const host = '127.0.0.1', database = 'postgres', user = 'mcp_test'
const listener = createServer()
let embedded, pool, started = false
try {
  await new Promise((done, reject) => { listener.once('error', reject); listener.listen(0, host, done) })
  const { port } = listener.address()
  await new Promise((done, reject) => listener.close(error => error ? reject(error) : done()))
  embedded = new EmbeddedPostgres({
    databaseDir: join(directory, 'pg'), port, user, password, persistent: true, createPostgresUser: false,
    initdbFlags: ['--encoding=UTF8'], postgresFlags: ['-h', host, '-c', 'unix_socket_directories='],
    onLog: () => {}, onError: () => {},
  })
  await embedded.initialise()
  await embedded.start()
  started = true
  pool = new pg.Pool({ host, port, database, user, password, max: 1, connectionTimeoutMillis: 5000 })
  await applySqlMigrations(pool)
  const { rows: [version] } = await pool.query("SELECT current_setting('server_version_num')::int AS number, (SELECT count(*)::int FROM schema_migrations) AS migrations")
  assert(version.number >= 180000 && version.number <= 189999)
  assert.equal(version.migrations, MIGRATION_MANIFEST.length)

  const runId = 'a012bcde', owner = 'netflow-operator-' + runId, collectorPort = 51000
  const read = (sql, probePort = collectorPort) => JSON.parse(run(psql, ['-X', '-w', '-qAt',
    '-h', host, '-p', String(port), '-U', user, '-d', database, '-v', 'ON_ERROR_STOP=1',
    '-v', 'database=' + database, '-v', 'socket_dir=/tmp/mcp-netflow-' + runId + '/pg-socket',
    '-v', 'run_id=' + runId, '-v', 'port=' + probePort],
  { env: { ...clientEnv, PGCONNECT_TIMEOUT: '2', PGPASSWORD: password }, input: sql }).trim())
  const identity = read(IDENTITY_SQL)
  assert.deepEqual(Object.keys(identity).sort(), ['oid', 'source'])
  assert.match(identity.oid, /^[1-9][0-9]{0,9}$/)
  // The production source predicate deliberately rejects this TCP fixture.
  assert.equal(identity.source, false)
  const expect = (planCount, enabled, verified, ordered = true, sqlPort = collectorPort) =>
    assert.deepEqual(read(SQL, sqlPort), { ...identity, connectionCount: 1, planCount, enabled, verified, ordered })

  const uuid = index => 'a012bcde-0000-4000-8000-' + String(index).padStart(12, '0')
  const connection = uuid(1)
  await pool.query(`INSERT INTO mcp_connections(id,name,owner_id,owner_permissions,policy,generation)
    VALUES($1,$2,$3,'[]','{}','inert-witness')`, [connection, 'Owned NetFlow ' + runId, owner])
  await pool.query("UPDATE traffic_flow_settings SET collector_ip='198.18.242.3', flow_listen_port=$1, enabled=false WHERE id=1", [collectorPort])
  expect(0, false, [])

  async function seed(index, enabled) {
    const plan = uuid(index), operation = uuid(index + 1), receipt = uuid(index + 2)
    const base = Date.UTC(2026, 0, 1) + index * 1000
    const at = offset => new Date(base + offset)
    const result = { receiptId: receipt, configurationConfirmed: true, receiptConfirmed: true,
      listener: 'request-acknowledged', automaticReplay: false }
    await pool.query(`INSERT INTO mcp_plans(id,connection_id,operation_id,operation_version,policy_version,generation,payload,evidence,state,expires_at,approved_by,approved_at,created_at)
      VALUES($1,$2,'netflow.settings.update',3,1,'inert-witness','{}','{}','completed',$3,$4,$5,$6)`,
    [plan, connection, at(9000), owner, at(100), at(0)])
    await pool.query(`INSERT INTO mcp_operations(id,plan_id,connection_id,idempotency_key,state,verification,created_at,finished_at)
      VALUES($1,$2,$3,$4,'applied','passed',$5,$6)`, [operation, plan, connection, 'inert-' + index, at(200), at(400)])
    await pool.query(`INSERT INTO mcp_operation_steps(operation_id,step_index,state,intended_state,result,created_at,finished_at)
      VALUES($1,0,'verified',$2,$3,$4,$5)`,
    [operation, { service: 'netflow.settings.update', input: { receiptId: receipt, enabled } }, result, at(300), at(400)])
    await pool.query(`INSERT INTO mcp_audit(actor_id,action,connection_id,payload,created_at)
      VALUES($1,'plan.approve',$2,$3,$4)`, [owner, connection, { planId: plan }, at(100)])
    await pool.query(`INSERT INTO flow_settings_intents(receipt_id,phase,request_proof,state,result,created_at)
      VALUES($1,'configuration','inert-proof','completed',$2,$3),($1,'listener','inert-proof','completed',$2,$3)`, [receipt, result, at(300)])
    await pool.query('UPDATE traffic_flow_settings SET enabled=$1 WHERE id=1', [enabled])
    return [plan, operation, receipt]
  }
  const enabled = await seed(10, true)
  expect(1, true, [enabled])
  const disabled = await seed(20, false)
  expect(2, false, [enabled, disabled])
  expect(2, null, [enabled, disabled], true, collectorPort + 1)

  await pool.query("UPDATE flow_settings_intents SET request_proof='foreign' WHERE receipt_id=$1 AND phase='listener'", [disabled[2]])
  expect(2, false, [enabled])
  await pool.query("UPDATE flow_settings_intents SET request_proof='inert-proof' WHERE receipt_id=$1 AND phase='listener'", [disabled[2]])
  const { rows: [duplicate] } = await pool.query(`INSERT INTO mcp_audit(actor_id,action,connection_id,payload)
    VALUES($1,'plan.approve',$2,$3) RETURNING id`, [owner, connection, { planId: disabled[0] }])
  expect(2, false, [enabled])
  await pool.query('DELETE FROM mcp_audit WHERE id=$1', [duplicate.id])
  const { rows: [original] } = await pool.query(`UPDATE mcp_operations SET finished_at=(SELECT created_at + interval '1 second' FROM mcp_plans WHERE id=$2)
    WHERE id=$1 RETURNING created_at`, [enabled[1], disabled[0]])
  expect(2, false, [enabled, disabled], false)
  await pool.query("UPDATE mcp_operations SET finished_at=$2::timestamptz + interval '200 milliseconds' WHERE id=$1", [enabled[1], original.created_at])
  expect(2, false, [enabled, disabled])
  console.log(JSON.stringify({ migrations: version.migrations, exactPsqlQueries: true, inertEnableDisableAndRejections: true,
    linuxSocketSourceAccepted: false, realMcpExecutionClaimed: false }))
} finally {
  try {
    await pool?.end()
  } finally {
    if (started) await embedded.stop()
    if (listener.listening) await new Promise(done => listener.close(done))
    await rm(directory, { recursive: true, force: true, maxRetries: 5, retryDelay: 200 })
  }
}