/**
 * 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, openSync } 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 or HELIOS_SKIP_MEMGRAPH=1: broker not started
    if (process.env.HELIOS_SKIP_INFRA === '1' || process.env.HELIOS_SKIP_MEMGRAPH === '1') { console.warn('[WARN][memgraph-broker-launcher] HELIOS_SKIP_INFRA/MEMGRAPH=1  -  skipping broker phase handler (headless mode)'); return; }
    // P2-11: create run dir here (not at module scope) — deferred to DATA_LAYER
    // so the mkdirSync only runs when the broker will actually be started.
    try {
      require('fs').mkdirSync(_brokerRunDir, { recursive: true, mode: 0o700 });
    } catch { /* dir may already exist — ignore EEXIST */ }
    // Fast path: if broker socket already exists and process alive, emit ready immediately.
    // This prevents the 8s polling wait when the broker is already running from a prior session.
    // (session_start fires AFTER phases  -  too late to trigger emitReady before DATA_LAYER resolves)
    if (!_readyEmitted && isBrokerRunning()) {
      // _pi not available at module scope; use globalThis flag to signal readiness
      _readyEmitted = true;
      (globalThis as any).__helios_broker_ready = true;
      const sm = (() => { try { return require('../lib/safe-memgraph'); } catch { return null; } })();
      if (sm && sm.openWarmGate) sm.openWarmGate();
    }
    await new Promise<void>((resolve) => {
      const timeout = setTimeout(() => { clearInterval(check); resolve(); }, 8000);
      const check = setInterval(() => {
        if (_readyEmitted) { clearInterval(check); clearTimeout(timeout); resolve(); }
        // Secondary fast-path: check broker mid-poll in case it started during our wait
        if (!_readyEmitted && isBrokerRunning()) {
          _readyEmitted = true;
          (globalThis as any).__helios_broker_ready = true;
          const sm = (() => { try { return require('../lib/safe-memgraph'); } catch { return null; } })();
          if (sm && sm.openWarmGate) sm.openWarmGate();
        }
      }, 200);
    });
  });
  _startupPhaseRegistered = true;
} catch { /* startup-lifecycle not available  -  fall back to session_start */ }

const BROKER_SOCK = process.env.HELIOS_BROKER_SOCK || (() => {
  // Use ipc-path.js as the single source of truth for the broker socket path.
  // This ensures the launcher and the broker server always agree on the path,
  // and provides multi-company isolation on Windows via the HELIOS_DATA hash.
  try {
    const { getBrokerIpcPath } = require('../lib/broker/ipc-path.js');
    const { HELIOS_ROOT, HELIOS_DATA } = require('../lib/helios-root.js');
    return getBrokerIpcPath(HELIOS_ROOT, HELIOS_DATA);
  } catch {
    // Fallback: uid-scoped POSIX socket (Windows will use UID=0 but that's fine as fallback)
    const uid = typeof process.getuid === 'function' ? process.getuid() : 0;
    return join(homedir(), '.helios', 'run', `memgraph-broker-${uid}.sock`);
  }
})();

// Lock and log files MUST be real filesystem paths, independent of BROKER_SOCK.
// On Windows, BROKER_SOCK is a named pipe (\\.\pipe\...) — .replace('.sock', ...) is a
// no-op, and writing to a named-pipe path as a file injects data into the broker IPC
// stream. Derive these from heliosDataPath('run') instead.
// P2-11 NX fix: _brokerRunDir path resolution ONLY — no mkdirSync at module scope.
// The directory is created inside the DATA_LAYER phase handler where it belongs,
// eliminating the synchronous filesystem write from the module parse path.
// This runs for every process that imports this file (including workers) — the
// mkdirSync was unnecessary for processes that never start a broker.
const _brokerRunDir = (() => {
  try {
    const { heliosDataPath } = require('../lib/helios-root.js');
    return heliosDataPath('run');
  } catch {
    return join(homedir(), '.helios', 'run');
  }
})();
const BROKER_LOCK          = process.env.HELIOS_BROKER_LOCK || join(_brokerRunDir, 'memgraph-broker.lock');
const BROKER_STARTING_LOCK = join(_brokerRunDir, 'memgraph-broker-starting.lock');
const BROKER_LOG           = join(_brokerRunDir, 'memgraph-broker.log');
const SERVER_PATH = process.env.HELIOS_SERVER_PATH || (() => {
  try {
    const { HELIOS_ROOT } = require('../lib/helios-root.js');
    return join(HELIOS_ROOT, 'lib/broker/server.js');
  } catch {
    return process.env.HELIOS_ROOT
      ? join(process.env.HELIOS_ROOT, 'lib/broker/server.js')
      : 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 {
  // Primary check: lock file (always a real filesystem path — never a named pipe).
  // On Windows, existsSync(BROKER_SOCK) is always false for named pipes, so we
  // cannot use socket existence as the liveness check. The lock file is the
  // authoritative source of truth on all platforms.
  if (!existsSync(BROKER_LOCK)) {
    // No lock file → broker not running. On POSIX also clean up stale socket.
    if (process.platform !== 'win32') { try { unlinkSync(BROKER_SOCK); } catch {} }
    return false;
  }
  try {
    const lock = JSON.parse(readFileSync(BROKER_LOCK, 'utf8'));
    process.kill(lock.pid, 0); // throws ESRCH if process is dead
    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 {}
        if (process.platform !== 'win32') { try { unlinkSync(BROKER_SOCK); } catch {} }
        return false;
      }
    }
    return true;
  } catch {
    // Process dead or lock corrupted — clean up
    try { unlinkSync(BROKER_LOCK); } catch {}
    if (process.platform !== 'win32') { try { unlinkSync(BROKER_SOCK); } catch {} }
    return false;
  }
}

function emitReady(pi: any): void {
  if (_readyEmitted) return;
  _readyEmitted = 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 immediately: queries must not flood an unresponsive broker.
      // Fall-open after 30s so direct-bolt fallback becomes available even if safe-memgraph's
      // own timer was reset by a generation change or missed.
      console.error('[broker-launcher] Warmup ping failed  -  holding warm gate closed:', e?.message);
      setTimeout(() => {
        try {
          const sm2 = require('../lib/safe-memgraph');
          if (sm2.openWarmGate) {
            process.stderr.write('[broker-launcher] Warmup-ping fall-open: opening warm gate after 30s timeout\n');
            sm2.openWarmGate();
          }
        } catch {}
      }, 30_000);
    }
    // Emit AFTER connection confirmed (or after fail-open fallthrough)
    try {
      const _bus = require('../lib/event-bus.js').default;
      if (_bus?.emit) _bus.emit('memgraph_ready', { timestamp: Date.now(), host: 'localhost', port: 7687 });
    } catch (_e) { /* fail-open: event bus optional */ }
    (globalThis as any).__memgraph_broker_ready = true;
    if (pi?.emit) pi.emit('memgraph_ready');
  })();
}

// 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 {
  // BUG-6 fix: ensure run dir exists before writing lock/log files.
  // This guards against the case where startBrokerProcess fires before the
  // DATA_LAYER phase handler has created _brokerRunDir.
  try { require('fs').mkdirSync(_brokerRunDir, { recursive: true, mode: 0o700 }); } catch { /* already exists */ }

  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 explicitly: the parent process may be a Bun binary
  // (helios CLI) or Electron whose execPath isn't a plain `node` binary.
  // In Electron, process.execPath = electron.exe — that is CORRECT to use as
  // execPath for fork(): Electron hosts Node.js and runs forked scripts fully.
  // On POSIX we prefer `which node` to find a standalone node binary when
  // available (avoids dragging in the full Electron runtime for the broker).
  const nodeExec = process.env.HELIOS_NODE_PATH
    || (() => {
      try {
        // Use platform-appropriate binary finder; take first result (where.exe may return multiple lines)
        const cmd = process.platform === 'win32' ? 'where.exe node' : 'which node';
        return require('child_process').execSync(cmd, { encoding: 'utf8' })
          .trim().split(/[\r\n]+/)[0].trim();
      } catch {
        // Fallback to process.execPath — always valid (node.exe on CLI, electron.exe in Electron).
        // Both correctly host the broker script via child_process.fork().
        return process.execPath;
      }
    })();
  // Open a dedicated log file for broker stderr.
  // Using 'inherit' for fd2 would copy the tee pipe write-end into the broker:
  // when node exits but broker survives (detached), the broker keeps the pipe
  // open → tee never gets EOF → asyncio.communicate() hangs for 870s.
  // Fix (Option E per 30-source research): open a real log file for fd2 so
  // broker diagnostics are preserved AND the pipe write-end is not inherited.
  // The parent closes its fd after fork so it doesn't leak a file handle.
  // BROKER_LOG is always a real filesystem path (never a named pipe) — see constant.
  let _brokerLogFd: number | null = null;
  try { _brokerLogFd = openSync(BROKER_LOG, 'a'); } catch { _brokerLogFd = null; }
  _brokerProc = fork(SERVER_PATH, ['--socket', BROKER_SOCK], {
    detached: true,
    windowsHide: true,  // Prevent Windows console popup
    stdio: ['ignore', 'ignore', _brokerLogFd ?? 'ignore', 'ipc'],
    execPath: nodeExec,
    env: { ...process.env },
  } as any);
  // Parent closes its copy of the log fd immediately  -  broker has its own dup.
  if (_brokerLogFd !== null) { try { require('fs').closeSync(_brokerLogFd); } catch {} }

  let _ready = false;
  const _readyTimeout = setTimeout(() => {
    if (!_ready) {
      // Fallback: if broker didn't signal ready within 12s, check lock file.
      // On Windows, named pipes don't appear in the filesystem so existsSync(BROKER_SOCK)
      // is always false — use the lock file as the liveness signal instead.
      if (existsSync(BROKER_LOCK)) {
        console.error('[broker-launcher] Broker lock 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 / HELIOS_SKIP_MEMGRAPH: skip broker startup ──
  if (process.env.HELIOS_SKIP_INFRA === '1' || process.env.HELIOS_SKIP_MEMGRAPH === '1') {
    console.warn('[WARN] [memgraph-broker-launcher] HELIOS_SKIP_INFRA/MEMGRAPH=1  -  skipping broker startup (headless mode)');
    return;
  }

  // Skip local broker when MEMGRAPH_BOLT_URL points to a non-localhost host
  const _boltUrl = process.env.MEMGRAPH_BOLT_URL || 'bolt://127.0.0.1:7687';
  const _boltHostMatch = _boltUrl.match(/bolt:\/\/([^:\/]+)/);
  const _boltHost = _boltHostMatch ? _boltHostMatch[1] : '127.0.0.1';
  if (!['localhost', '127.0.0.1', '::1'].includes(_boltHost)) {
    process.stderr.write('[memgraph-broker-launcher] MEMGRAPH_BOLT_URL points to remote host  -  skipping local broker\n');
    process.env.HELIOS_SKIP_MEMGRAPH = '1';
    return;
  }

  // Start broker immediately at extension load time (not waiting for session_start).
  // This ensures _readyEmitted is set before the DATA_LAYER phase polls for it.
  // The DATA_LAYER handler polls _readyEmitted every 200ms; session_start fires AFTER
  // phases complete  -  too late. Calling here gives the broker time to emit ready
  // before DATA_LAYER begins its 8s countdown.
  if (!_readyEmitted) startBrokerProcess(pi);

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

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