/**
 * Client for the OMP session control socket.
 *
 * The bridge can observe an externally driven thread by tailing its session
 * JSONL, but it cannot write to one. When the OMP side is running the
 * `extension/control-socket.mjs` extension, each live session exposes a Unix
 * socket keyed by session id, which turns "someone else is driving this
 * thread" from read-only into steerable.
 *
 * Every call is best-effort: no socket means the session is not running the
 * extension (or is not running at all), which callers treat as "no control
 * available" rather than as an error.
 */
import net from "node:net";
import os from "node:os";
import path from "node:path";

const DEFAULT_TIMEOUT_MS = 2_000;

export function controlSocketDir() {
  return process.env.OMP_CONTROL_SOCKET_DIR
    || path.join(os.homedir(), ".omp", "codex-bridge", "control");
}

export function controlSocketPath(sessionId) {
  return path.join(controlSocketDir(), `${sessionId}.sock`);
}

/**
 * Send one request and resolve its reply, or null when the session offers no
 * control socket.
 */
export function controlRequest(sessionId, request, { timeoutMs = DEFAULT_TIMEOUT_MS } = {}) {
  if (!sessionId) return Promise.resolve(null);
  const socketPath = controlSocketPath(sessionId);
  return new Promise((resolve) => {
    let settled = false;
    let buffer = "";
    const finish = (value) => {
      if (settled) return;
      settled = true;
      socket.destroy();
      clearTimeout(timer);
      resolve(value);
    };
    const socket = net.connect(socketPath);
    const timer = setTimeout(() => finish(null), timeoutMs);
    timer.unref?.();
    socket.setEncoding("utf8");
    socket.once("connect", () => socket.write(`${JSON.stringify(request)}\n`));
    socket.on("data", (chunk) => {
      buffer += chunk;
      const newline = buffer.indexOf("\n");
      if (newline < 0) return;
      try {
        finish(JSON.parse(buffer.slice(0, newline)));
      } catch {
        finish(null);
      }
    });
    socket.on("error", () => finish(null));
    socket.on("close", () => finish(null));
  });
}

/** Live session snapshot, or null when the session is not controllable. */
export async function controlStatus(sessionId, options = {}) {
  const reply = await controlRequest(sessionId, { op: "status" }, options);
  return reply?.ok ? reply : null;
}

/** True once the running session accepted the message. */
export async function controlSend(sessionId, text, { deliverAs = "steer", ...options } = {}) {
  const reply = await controlRequest(sessionId, { op: deliverAs, text }, options);
  return Boolean(reply?.ok);
}

/** True once the running session accepted model/thinking overrides. */
export async function controlConfigure(sessionId, { model = null, reasoningEffort = null } = {}, options = {}) {
  if (!model && !reasoningEffort) return true;
  const reply = await controlRequest(sessionId, { op: "configure", model, reasoningEffort }, options);
  return Boolean(reply?.ok);
}

export async function controlRespondAsk(sessionId, requestId, answers, options = {}) {
  const reply = await controlRequest(sessionId, { op: "ask_response", requestId, answers }, options);
  return Boolean(reply?.ok);
}

export async function controlCancelAsk(sessionId, requestId, options = {}) {
  const reply = await controlRequest(sessionId, { op: "ask_cancel", requestId }, options);
  return Boolean(reply?.ok);
}

/** True once the running session accepted the interrupt. */
export async function controlInterrupt(sessionId, options = {}) {
  const reply = await controlRequest(sessionId, { op: "interrupt" }, options);
  return Boolean(reply?.ok);
}

/**
 * Ask a live session to compact its own context.
 *
 * Compaction runs far longer than a control round trip, so the session only
 * acknowledges that it started; the compaction itself shows up as a lifecycle
 * event and in the session log.
 */
export async function controlCompact(sessionId, { instructions = null, ...options } = {}) {
  const reply = await controlRequest(sessionId, { op: "compact", instructions }, options);
  return Boolean(reply?.ok);
}

/**
 * Subscribe to live session lifecycle events.
 *
 * Performs handshake, consumes ordered frames, deduplicates sequence numbers,
 * and calls onEvent for each validated frame and onDisconnect when the connection closes.
 *
 * Resolves to a subscription handle { streamId, sequence, status, reset, close }
 * or null if the session offers no control socket or the handshake fails.
 */
export function controlSubscribe(
  sessionId,
  {
    onEvent = null,
    onDisconnect = null,
    streamId = null,
    afterSequence = null,
    timeoutMs = DEFAULT_TIMEOUT_MS,
    signal = null,
  } = {}
) {
  if (!sessionId || signal?.aborted) return Promise.resolve(null);
  const socketPath = controlSocketPath(sessionId);

  return new Promise((resolve) => {
    let settled = false;
    let closed = false;
    let disconnectedCalled = false;
    let buffer = "";
    let handle = null;
    let lastSeenSequence = -1;

    const socket = net.connect(socketPath);

    const timer = setTimeout(() => {
      if (!settled) {
        finish(null);
      }
    }, timeoutMs);
    timer.unref?.();

    const finish = (val) => {
      if (settled) return;
      settled = true;
      clearTimeout(timer);
      signal?.removeEventListener("abort", onAbort);
      if (!val) {
        socket.destroy();
      }
      resolve(val);
    };
    const onAbort = () => finish(null);
    signal?.addEventListener("abort", onAbort, { once: true });

    const triggerDisconnect = () => {
      if (disconnectedCalled) return;
      disconnectedCalled = true;
      clearTimeout(timer);
      signal?.removeEventListener("abort", onAbort);
      if (!settled) {
        settled = true;
        resolve(null);
      }
      onDisconnect?.();
    };

    socket.setEncoding("utf8");

    socket.once("connect", () => {
      const subscribeReq = {
        op: "subscribe",
        streamId: streamId ?? undefined,
        afterSequence: afterSequence ?? undefined,
      };
      socket.write(`${JSON.stringify(subscribeReq)}\n`);
    });

    socket.on("data", (chunk) => {
      buffer += chunk;
      let newline = buffer.indexOf("\n");
      while (newline >= 0) {
        const line = buffer.slice(0, newline).trim();
        buffer = buffer.slice(newline + 1);
        if (line) {
          let msg;
          try {
            msg = JSON.parse(line);
          } catch {
            newline = buffer.indexOf("\n");
            continue;
          }

          if (!handle) {
            // Expecting subscription ACK handshake
            if (msg && msg.ok && msg.op === "subscribe") {
              const ackStreamId = typeof msg.streamId === "string" ? msg.streamId : "";
              const ackSequence = typeof msg.sequence === "number" ? msg.sequence : 0;
              const isReset = Boolean(msg.reset);

              if (!isReset && typeof afterSequence === "number" && Number.isInteger(afterSequence) && afterSequence >= 0) {
                lastSeenSequence = afterSequence;
              } else {
                lastSeenSequence = ackSequence;
              }

              let currentSeq = lastSeenSequence;
              let currentStatus = msg.status ?? null;
              let currentStreamId = ackStreamId;

              handle = {
                get streamId() {
                  return currentStreamId;
                },
                set streamId(val) {
                  currentStreamId = val;
                },
                get sequence() {
                  return currentSeq;
                },
                set sequence(val) {
                  currentSeq = val;
                },
                get status() {
                  return currentStatus;
                },
                set status(val) {
                  currentStatus = val;
                },
                pendingAsks: Array.isArray(msg.pendingAsks) ? msg.pendingAsks : [],
                headSequence: ackSequence,
                reset: isReset,
                close() {
                  if (closed) return;
                  closed = true;
                  socket.destroy();
                },
              };

              finish(handle);
            } else {
              finish(null);
              return;
            }
          } else {
            // Consuming subsequent event frames
            if (msg && msg.type === "event") {
              if (
                msg.streamId === handle.streamId &&
                typeof msg.sequence === "number" &&
                msg.sequence > lastSeenSequence
              ) {
                lastSeenSequence = msg.sequence;
                handle.sequence = msg.sequence;
                if (msg.status !== undefined) {
                  handle.status = msg.status;
                }
                try {
                  onEvent?.(msg);
                } catch {
                  // User callback error should not crash client loop
                }
              }
            }
          }
        }
        newline = buffer.indexOf("\n");
      }
    });

    socket.on("error", () => {
      triggerDisconnect();
    });

    socket.on("close", () => {
      triggerDisconnect();
    });

    socket.on("end", () => {
      triggerDisconnect();
    });
  });
}