/**
 * memgraph-broker-launcher.ts — Auto-starts the Memgraph connection broker.
 *
 * Uses Node.js fork() with IPC channel for readiness signaling.
 * The broker sends { type: 'ready' } after its socket is listening.
 * This eliminates filesystem polling and timing races entirely.
 *
 * Lifecycle:
 *   session_start → fork broker → wait for IPC 'ready' → emit memgraph_ready → disconnect IPC
 *   session_shutdown → release reference (broker survives for other sessions)
 */
import type { ExtensionAPI } from '@helios-agent/pi-coding-agent';
import { fork, type ChildProcess } from 'node:child_process';
import { existsSync, unlinkSync, readFileSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { homedir } from 'node:os';

let _startupPhaseRegistered = false;
try {
  const { registerStartupPhase, StartupPhase } = require('../lib/startup-lifecycle.ts');
  registerStartupPhase('memgraph-broker', StartupPhase.DATA_LAYER, async () => {
    // HELIOS_SKIP_INFRA=1: headless/judge mode — broker not started, resolve immediately
    if (process.env.HELIOS_SKIP_INFRA === '1') return;
    await new Promise<void>((resolve) => {
      const timeout = setTimeout(() => { clearInterval(check); resolve(); }, 8000);
      const check = setInterval(() => {
        if (_readyEmitted) { clearInterval(check); clearTimeout(timeout); resolve(); }
      }, 200);
    });
  });
  _startupPhaseRegistered = true;
} catch { /* startup-lifecycle not available — fall back to session_start */ }

const BROKER_SOCK = process.env.HELIOS_BROKER_SOCK || join(homedir(), 'helios-agent', 'run', 'memgraph-broker.sock');
const BROKER_LOCK = BROKER_SOCK.replace('.sock', '.lock');
const BROKER_STARTING_LOCK = BROKER_SOCK.replace('.sock', '-starting.lock');
const SERVER_PATH = join(homedir(), 'helios-agent/lib/broker/server.js');

// Set at module load time (before factory runs) so safe-memgraph sees it immediately
(globalThis as any).__helios_broker_managed = true;

let _brokerProc: ChildProcess | null = null;
let _restartAttempts = 0;
let _readyEmitted = false;
// AR-H17: Module-level counter for mutex-wait path to prevent unbounded retry loop.
// The mutex setTimeout path calls verifySocketThenEmit(pi, BROKER_SOCK) without an attempt
// argument, meaning it always starts from attempt=0. This independent counter caps it.
let _verifyAttemptFromMutex = 0;
const MAX_MUTEX_VERIFY_ATTEMPTS = 25;

function isBrokerRunning(): boolean {
  if (!existsSync(BROKER_SOCK)) return false;
  if (existsSync(BROKER_LOCK)) {
    try {
      const lock = JSON.parse(readFileSync(BROKER_LOCK, 'utf8'));
      process.kill(lock.pid, 0);
      if (lock.codeHash) {
        const { createHash } = require('crypto');
        const { readdirSync } = require('fs');
        const { dirname } = require('path');
        const brokerDir = dirname(SERVER_PATH);
        const h = createHash('md5');
        readdirSync(brokerDir).filter((f: string) => f.endsWith('.js')).sort().forEach((f: string) => {
          h.update(readFileSync(join(brokerDir, f)));
        });
        try { h.update(readFileSync(join(brokerDir, '..', 'safe-memgraph.js'))); } catch {}
        const currentHash = h.digest('hex');
        if (lock.codeHash !== currentHash) {
          process.stderr.write(`[broker-launcher] Broker code changed (${lock.codeHash.slice(0,8)}→${currentHash.slice(0,8)}), restarting...\n`);
          try { process.kill(lock.pid, 'SIGTERM'); } catch {}
          try { unlinkSync(BROKER_LOCK); } catch {}
          try { unlinkSync(BROKER_SOCK); } catch {}
          return false;
        }
      }
      return true;
    } catch {
      try { unlinkSync(BROKER_LOCK); } catch {}
      try { unlinkSync(BROKER_SOCK); } catch {}
      return false;
    }
  }
  try { unlinkSync(BROKER_SOCK); } catch {}
  return false;
}

function emitReady(pi: any): void {
  if (_readyEmitted) return;
  _readyEmitted = true;
  if (pi.emit) pi.emit('memgraph_ready');
  (globalThis as any).__memgraph_broker_ready = true;
  (async () => {
    try {
      const sm = require('../lib/safe-memgraph');
      const client = sm.getBrokerClient ? sm.getBrokerClient() : null;
      if (client && client.connectToBroker) {
        await client.connectToBroker();
        // Verify full round-trip works before opening gate
        if (client.sendRequest) {
          await client.sendRequest({ id: 'warmup-ping', method: 'stats' });
        }
      }
      if (sm.openWarmGate) sm.openWarmGate();
    } catch (e: any) {
      // Warmup ping failed — broker is alive but unresponsive (frozen transaction, OOM, etc).
      // Do NOT open the warm gate: queries must not flood an unresponsive broker.
      // safe-memgraph's 30s fail-open timer will eventually open the gate for direct-bolt fallback.
      console.error('[broker-launcher] Warmup ping failed — holding warm gate closed:', e?.message);
    }
  })();
}

// LK-06: Add attempt counter to cap infinite retry loop
function verifySocketThenEmit(pi: any, socketPath: string, attempt = 0): void {
  if (attempt > 50) {
    process.stderr.write('[memgraph-broker] ERROR: broker socket never appeared after 50 attempts\n');
    return;
  }
  const net = require('net');
  const sock = net.createConnection(socketPath);
  // LK-07: Store the 3s ready-timer handle so it can be cleared on socket events
  const readyTimer = setTimeout(() => { sock.destroy(); emitReady(pi); }, 3000);
  sock.on('connect', () => {
    clearTimeout(readyTimer);
    sock.destroy();
    emitReady(pi);
  });
  sock.on('error', () => {
    clearTimeout(readyTimer);
    sock.destroy();
    setTimeout(() => verifySocketThenEmit(pi, socketPath, attempt + 1), 200);
  });
}

function startBrokerProcess(pi: any): void {
  if (isBrokerRunning()) {
    console.error('[broker-launcher] Broker already running, skipping start');
    emitReady(pi);
    return;
  }
  // Mutex: prevent two sessions from starting broker simultaneously
  if (existsSync(BROKER_STARTING_LOCK)) {
    try {
      const startLock = JSON.parse(readFileSync(BROKER_STARTING_LOCK, 'utf8'));
      if (Date.now() - startLock.ts < 15000) {
        // Another process is starting the broker — wait for it
        console.error('[broker-launcher] Another process is starting broker, waiting...');
        setTimeout(() => {
          if (_readyEmitted) return; // already done
          _verifyAttemptFromMutex++;
          if (_verifyAttemptFromMutex > MAX_MUTEX_VERIFY_ATTEMPTS) {
            process.stderr.write('[memgraph-broker] ERROR: mutex wait path exhausted 25 attempts — giving up\n');
            return;
          }
          verifySocketThenEmit(pi, BROKER_SOCK, 0);
        }, 5000);
        return;
      }
    } catch {}
    try { unlinkSync(BROKER_STARTING_LOCK); } catch {}
  }
  try { writeFileSync(BROKER_STARTING_LOCK, JSON.stringify({ pid: process.pid, ts: Date.now() })); } catch {}
  if (!existsSync(SERVER_PATH)) {
    console.error('[broker-launcher] server.js not found at ' + SERVER_PATH);
    return;
  }
  console.error('[broker-launcher] Starting Memgraph broker...');

  // fork() creates an IPC channel — broker can send us a 'ready' message
  // MUST specify execPath=node explicitly: the parent process may be a Bun binary
  // (helios CLI) whose execPath doesn't understand --socket as a script arg.
  const nodeExec = process.env.HELIOS_NODE_PATH
    || (() => { try { const cmd = process.platform === 'win32' ? 'where.exe node' : 'which node'; return require('child_process').execSync(cmd, { encoding: 'utf8', stdio: 'pipe' }).trim().split('\n')[0].trim(); } catch { return process.execPath || 'node'; } })();
  _brokerProc = fork(SERVER_PATH, ['--socket', BROKER_SOCK], {
    detached: true,
    stdio: ['ignore', 'ignore', 'inherit', 'ipc'],
    execPath: nodeExec,
    env: { ...process.env },
  });

  let _ready = false;
  const _readyTimeout = setTimeout(() => {
    if (!_ready) {
      // Fallback: if broker didn't signal ready within 12s, check socket directly
      if (existsSync(BROKER_SOCK)) {
        console.error('[broker-launcher] Broker socket exists (IPC signal missed) — treating as ready');
        _ready = true;
        emitReady(pi);
      } else {
        console.error('[broker-launcher] Broker did not become ready within 12s');
        if (_restartAttempts < 3) {
          _restartAttempts++;
          setTimeout(() => startBrokerProcess(pi), 2000);
        }
      }
    }
    // Disconnect IPC regardless (don't hold parent open)
    if (_brokerProc && _brokerProc.connected) {
      _brokerProc.disconnect();
    }
  }, 12000);

  _brokerProc.on('message', (msg: any) => {
    if (msg && msg.type === 'ready') {
      _ready = true;
      _restartAttempts = 0;
      clearTimeout(_readyTimeout);
      try { unlinkSync(BROKER_STARTING_LOCK); } catch {}
      console.error('[broker-launcher] Broker ready (PID ' + _brokerProc?.pid + ', socket: ' + (msg.socketPath || BROKER_SOCK) + ')');
      verifySocketThenEmit(pi, msg.socketPath || BROKER_SOCK);
      if (_brokerProc && _brokerProc.connected) {
        _brokerProc.disconnect();
      }
      _brokerProc?.unref();
    }
  });

  _brokerProc.on('error', (err: Error) => {
    console.error('[broker-launcher] fork error: ' + err.message);
    _brokerProc = null;
    clearTimeout(_readyTimeout);
  });

  _brokerProc.on('exit', (code: number | null) => {
    clearTimeout(_readyTimeout); // R3-H48: always clear, regardless of exit reason
    _brokerProc = null;
    if (!_ready && code !== null && code !== 0) {
      console.error('[broker-launcher] Broker exited with code ' + code);
      if (_restartAttempts < 3) {
        _restartAttempts++;
        setTimeout(() => startBrokerProcess(pi), 2000);
      } else {
        // All retries exhausted — open warm gate for direct-bolt fallback
        console.error('[broker-launcher] All retries exhausted — opening warm gate for direct fallback');
        try { const sm = require('../lib/safe-memgraph'); if (sm.openWarmGate) sm.openWarmGate(); } catch {}
      }
    }
  });

  // Unref immediately so parent doesn't wait for detached child exit
  _brokerProc.unref();
}

function stopBrokerProcess(): void {
  if (_brokerProc) {
    try { process.kill(_brokerProc.pid!, 'SIGTERM'); } catch {}
    _brokerProc = null;
  }
}

export default function (pi: ExtensionAPI) {
  // ── HELIOS_SKIP_INFRA: headless/judge mode — skip broker startup ──
  // Set by helios-rpc.js --headless (e.g. rpc_judge_v2.py). Only infra extensions skip.
  if (process.env.HELIOS_SKIP_INFRA === '1') {
    process.stderr.write('[memgraph-broker-launcher] HELIOS_SKIP_INFRA=1 — skipping broker startup (headless mode)\n');
    return;
  }

  pi.on('session_start', () => {
    if (!_readyEmitted) startBrokerProcess(pi);
  });

  pi.on('session_shutdown', () => {
    _brokerProc = null;
  });
}