/**
 * Unix-socket server — PROTOCOL §7.1 transport + §8 socketPath + §9.1 hello-on-connect
 * (saga tb4593.3).
 *
 * Listens on a unix-domain socket (default `~/.shard/daemon.sock`), accepts connections,
 * frames them with the EXISTING `JsonlDecoder`/`encodeJsonl` (LF-only, §2.1 — never
 * `readline`, which splits on U+2028/U+2029), and emits `hello` immediately on connect
 * (`PROTOCOL_VERSION` from lifecycle.ts, `server_version` from package.json — §9.1).
 *
 * Owns connection IO ONLY: accept, frame, hello, track, close. **No routing, no proxy, no
 * message dispatch** — that's tb4593.4 wired on top via {@link SocketServer.onConnect}.
 * The connection hands the caller a `JsonlDecoder`-fed `messages()` async iterable (inbound
 * commands) + a `send` write (outbound events), so .4 owns the multiplexing semantics and
 * this module owns the bytes.
 *
 * ## Socket lifecycle (§8, §13)
 *   - **Boot:** if `socketPath` exists but no listener answers, it's a stale socket from a
 *     crashed daemon — unlink it before bind (defensive; failure to bind then surfaces). The
 *     caller (tb4593.6 daemon wiring) is expected to pre-ensure the parent dir exists.
 *   - **Shutdown:** `stop()` closes the server and `unlink`s the socket file. Best-effort:
 *     a missing file on unlink is swallowed (a concurrent `daemon.shutdown` race is fine).
 *
 * §14.1: the daemon binds a unix socket for local clients; this module is that surface. No
 * network exposure, no token handshake here (auth is implicit filesystem perms per tb4593
 * description; token+origin gate land with the web surface, S15).
 */
import { createConnection, createServer, type Socket, type Server } from 'node:net';
import { lstat, unlink } from 'node:fs/promises';
import { dirname } from 'node:path';
import { mkdir } from 'node:fs/promises';
import { homedir } from 'node:os';
import { JsonlDecoder, encodeJsonl, type JsonlRecord } from '@/protocol/framing/jsonl';
import { PROTOCOL_VERSION } from '@/protocol/lifecycle';
import type { AuthContext } from './auth';
import pkg from '../../package.json' with { type: 'json' };

/** `hello` capabilities advertised by the daemon socket (§9.1). `streaming-deltas` matches
 *  the stdio agent server (src/transports/stdio.ts). */
const DAEMON_CAPABILITIES: readonly string[] = ['streaming-deltas'];

/**
 * A server-side handle to one accepted client connection. The socket IO is delegated to
 * framing helpers; `messages()` yields inbound JSONL records (each a `JsonlRecord` so the
 * caller decides how to handle parse errors per §8), `send()` writes an outbound message.
 */
export interface Connection {
  /** A stable id for this connection (UUID; used by .4 for attach-set bookkeeping). */
  readonly id: string;
  /** Local filesystem-authenticated connection context. */
  readonly auth: AuthContext;
  /** Underlying socket (for close/error introspection). */
  readonly socket: Socket;
  /** Queue a message to the client (JSONL-encoded, §2.1). Returns once the OS has it. */
  send(message: unknown): Promise<void>;
  /** Inbound JSONL records, parsed per-record (parse failures carry `parseError`). */
  messages(): AsyncGenerator<JsonlRecord>;
  /** Whether the connection is still open. */
  readonly closed: boolean;
}

/** Listener invoked for every accepted connection, AFTER `hello` is sent (§9.1). */
export type ConnectListener = (conn: Connection) => void;

/** Internal: a connection's mutable state shared across the `Connection` view + iteration. */
interface ConnState {
  id: string;
  socket: Socket;
  decoder: JsonlDecoder;
  closed: boolean;
  /** Pending inbound records awaiting a `messages()` consumer. */
  queue: JsonlRecord[];
  /** Resolved when either a record is queued or the socket ends (so the iterator unblocks). */
  drain: () => void;
  /** Set when the connection has been torn down (idempotent). */
  teardown: () => void;
}

/** Constructor options for {@link SocketServer}. All optional; sane defaults for §13. */
export interface SocketServerOptions {
  /** Socket path (default `~/.shard/daemon.sock`). */
  socketPath?: string;
  /** Server version string for `hello` (default: `shard <pkg.version>`). */
  serverVersion?: string;
  /** Backlog for `listen()` (default 128). */
  backlog?: number;
}

/**
 * Unix-socket server for the daemon. Call {@link SocketServer.start} to bind, then either
 * subscribe via {@link SocketServer.onConnect} or iterate {@link SocketServer.connections}.
 * Call {@link SocketServer.stop} to close + unlink.
 */
export class SocketServer {
  private readonly socketPath: string;
  private readonly serverVersion: string;
  private readonly backlog: number;
  private server: Server | undefined;
  private boundSocketIdentity: { dev: number; ino: number } | undefined;
  private readonly listeners = new Set<ConnectListener>();
  private readonly conns = new Map<string, ConnState>();
  /** Pending connections not yet consumed by an `onConnect` listener. */
  private pending: Connection[] = [];
  /** Resolvers for an outstanding `connections()` async iterator `next()`. */
  private waiters: Array<(conn: Connection) => void> = [];

  constructor(opts: SocketServerOptions = {}) {
    this.socketPath = opts.socketPath ?? defaultSocketPath();
    this.serverVersion = opts.serverVersion ?? `shard ${pkg.version ?? '0.0.0-unknown'}`;
    this.backlog = opts.backlog ?? 128;
  }

  /** The configured socket path. */
  get path(): string {
    return this.socketPath;
  }

  /**
   * Bind + listen. Removes a stale socket file at `socketPath` defensively (a crashed
   * daemon may leave one; `EADDRINUSE` would block the bind). Ensures the parent dir
   * exists. Resolves once listening; never rejects (bind errors throw — caller handles).
   */
  async start(): Promise<void> {
    // Ensure the parent dir exists (~/.shard/ in production).
    await mkdir(dirname(this.socketPath), { recursive: true });

    // Never unlink a socket owned by a live daemon. Unlinking a pathname does not close
    // the existing listener: a second daemon can otherwise bind the same name while the
    // first remains alive, splitting web traffic and managed-agent callbacks across two
    // brokers. Only remove the path after proving that no listener accepts connections.
    if (await socketAcceptsConnections(this.socketPath)) {
      const error = new Error(`daemon socket is already in use: ${this.socketPath}`) as NodeJS.ErrnoException;
      error.code = 'EADDRINUSE';
      throw error;
    }
    await safeUnlink(this.socketPath);

    return new Promise<void>((resolve, reject) => {
      const server = createServer((socket) => this.handleConnection(socket));
      server.on('error', (err) => {
        // Surface bind-time errors to the caller via the start() promise; runtime errors
        // (a dropped client) are handled per-socket in handleConnection.
        if (this.server === undefined) {
          reject(err);
        } else {
          // Post-start error: log to the daemon's stderr; server stays up unless fatal.
          process.stderr.write(`shardd: socket server error: ${err.message}\n`);
        }
      });
      server.listen({ path: this.socketPath, backlog: this.backlog }, () => {
        this.server = server;
        void lstat(this.socketPath).then((stat) => {
          if (this.server === server) this.boundSocketIdentity = { dev: stat.dev, ino: stat.ino };
          resolve();
        }, (error) => {
          if (this.server === server) this.server = undefined;
          server.close(() => reject(error));
        });
      });
    });
  }

  /**
   * Register a listener fired for every accepted connection (after `hello` is sent).
   * Returns an unsubscribe function. Listeners are the integration point for tb4593.4
   * (proxy subscribe) and tb4593.5 (orchestrator command dispatch).
   */
  onConnect(listener: ConnectListener): () => void {
    this.listeners.add(listener);
    // Replay pending connections a late subscriber missed (mirrors the EventBus pattern).
    for (const conn of this.pending) {
      queueMicrotask(() => listener(conn));
    }
    this.pending = [];
    return () => {
      this.listeners.delete(listener);
    };
  }

  /**
   * Async-iterable of accepted connections (alternative to {@link onConnect}). Consumed
   * by the daemon's main accept loop if it prefers an iterator-driven shape.
   */
  async *connections(): AsyncGenerator<Connection> {
    while (this.server !== undefined) {
      const conn = await new Promise<Connection | undefined>((resolve) => {
        if (this.pending.length > 0) {
          resolve(this.pending.shift());
          return;
        }
        this.waiters.push((c) => resolve(c));
      });
      if (conn === undefined) {
        // Server stopped mid-await (a stop()-flushed waiter). End the iterator.
        break;
      }
      yield conn;
    }
  }

  /**
   * Snapshot of currently-open connections (for .4 fan-out bookkeeping / `agent.attach`
   * resolution). Returns the live `Connection` views; callers must not hold beyond the
   * connection's lifetime.
   */
  liveConnections(): Connection[] {
    return [...this.conns.values()].map((s) => this.toConnection(s));
  }

  /**
   * Graceful shutdown: stop accepting, destroy live sockets, unlink the socket file.
   * Resolves once the server's `close` callback fires. Idempotent (subsequent calls no-op).
   */
  async stop(): Promise<void> {
    const server = this.server;
    if (server === undefined) {
      return;
    }
    this.server = undefined;

    // Flush any outstanding `connections()` waiters so they unblock + end their loop.
    while (this.waiters.length > 0) {
      this.waiters.shift()?.(undefined as unknown as Connection);
    }
    this.pending = [];

    // Destroy every live connection (tears down per-conn decoder + resolves iterators).
    for (const state of [...this.conns.values()]) {
      state.teardown();
    }

    await new Promise<void>((resolve) => {
      server.close(() => resolve());
    });
    const identity = this.boundSocketIdentity;
    this.boundSocketIdentity = undefined;
    if (identity !== undefined) await unlinkIfOwned(this.socketPath, identity);
  }

  // ─── Internal ──────────────────────────────────────────────────────────────────

  /** Wire up one accepted socket: hello-on-connect, decoder, inbound queue, teardown. */
  private handleConnection(socket: Socket): void {
    if (this.server === undefined) {
      // Server stopped between the `listen` callback and this `connection` event. Refuse.
      socket.destroy();
      return;
    }
    const id = crypto.randomUUID();
    const decoder = new JsonlDecoder();
    let waiter: (() => void) | undefined;

    const state: ConnState = {
      id,
      socket,
      decoder,
      closed: false,
      queue: [],
      drain: () => {
        waiter?.();
      },
      teardown: () => {
        if (state.closed) return;
        state.closed = true;
        try {
          socket.destroy();
        } catch {
          // Already destroyed.
        }
        // Drop all server-owned references as soon as the socket closes. Without this,
        // every historical client socket remained reachable for the daemon's lifetime.
        this.conns.delete(id);
        this.pending = this.pending.filter((conn) => conn.id !== id);
        // Unblock a `messages()` consumer waiting for the next record (so it observes EOF).
        state.drain();
      },
    };
    this.conns.set(id, state);

    // §9.1 hello — first on connect, before any inbound command can arrive. Validate
    // against HelloSchema shape (mirrors the stdio server's sendRaw). Caller checks the
    // protocol_version per §7.
    const hello = {
      type: 'hello' as const,
      protocol_version: PROTOCOL_VERSION,
      server_version: this.serverVersion,
      capabilities: [...DAEMON_CAPABILITIES],
    };
    socket.write(encodeJsonl(hello));

    // Feed inbound bytes to the decoder; each completed record is queued + drains the
    // `messages()` iterator. A socket `error` (drop, reset) tears the conn down.
    socket.on('data', (chunk: Buffer) => {
      decoder.write(chunk);
      const records = decoder.parse();
      if (records.length > 0) {
        state.queue.push(...records);
        state.drain();
      }
    });
    socket.on('error', (err) => {
      if (!state.closed) {
        process.stderr.write(`shardd: connection ${id} error: ${err.message}\n`);
      }
      state.teardown();
    });
    socket.on('close', () => {
      state.teardown();
    });

    const conn = this.toConnection(state);

    // Distribute to subscribers + the async iterator queue.
    if (this.listeners.size > 0) {
      for (const listener of this.listeners) {
        queueMicrotask(() => listener(conn));
      }
    } else {
      // No subscribers yet — buffer for a late `onConnect`/`connections()` consumer.
      this.pending.push(conn);
    }
    if (this.waiters.length > 0) {
      const resolve = this.waiters.shift();
      if (resolve !== undefined) {
        // Don't re-buffer — hand straight to the waiter, skipping `pending` entirely.
        const idx = this.pending.indexOf(conn);
        if (idx >= 0) this.pending.splice(idx, 1);
        queueMicrotask(() => resolve(conn));
      }
    }
  }

  /** Build the public `Connection` view from internal `ConnState`. */
  private toConnection(state: ConnState): Connection {
    return {
      id: state.id,
      auth: { kind: 'local' },
      socket: state.socket,
      get closed() {
        return state.closed;
      },
      send: (message: unknown) => writeEnqueued(state.socket, message),
      messages: () => connMessages(state),
    };
  }
}

// ─── Module-private helpers ──────────────────────────────────────────────────────

/** Resolve the default socket path (~/.shard/daemon.sock, §13). */
function defaultSocketPath(): string {
  return `${homedir()}/.shard/daemon.sock`;
}

/**
 * Write a JSONL-encoded message to a socket, resolving once the bytes are queued. Uses
 * `socket.write`'s callback to detect backpressure-friendly completion. Rejects if the
 * socket is closed before the write flushes (caller decides whether to drop or requeue).
 */
function writeEnqueued(socket: Socket, message: unknown): Promise<void> {
  return new Promise<void>((resolve, reject) => {
    if (socket.destroyed) {
      reject(new Error('socket closed'));
      return;
    }
    const ok = socket.write(encodeJsonl(message), (err) => {
      if (err !== undefined && err !== null) {
        reject(err);
      } else {
        resolve();
      }
    });
    void ok; // backpressure handled by node's highWaterMark; v0.1 doesn't cork on `false`.
  });
}

/** The inbound messages async-iterable for one connection. */
async function* connMessages(state: ConnState): AsyncGenerator<JsonlRecord> {
  while (!state.closed) {
    // Drain whatever is buffered synchronously first (no await if the queue has records).
    if (state.queue.length > 0) {
      yield state.queue.shift() as JsonlRecord;
      continue;
    }
    // Wait for either new records or teardown (EOF). One promise per stall.
    await new Promise<void>((resolve) => {
      // `drain` is overwritten per stall so only the current waiter unblocks.
      state.drain = () => resolve();
    });
    if (state.queue.length === 0 && state.closed) {
      // Teardown woke us with nothing left to yield — EOF.
      return;
    }
  }
}

/**
 * Probe an existing Unix socket without mutating its pathname. A successful connect proves
 * that another daemon owns it. Only ENOENT/ECONNREFUSED prove the path is stale; timeouts
 * and unusual errors fail closed so startup never destroys a listener it could not verify.
 */
async function socketAcceptsConnections(path: string): Promise<boolean> {
  return new Promise<boolean>((resolve) => {
    const socket = createConnection({ path });
    let settled = false;
    const finish = (live: boolean) => {
      if (settled) return;
      settled = true;
      clearTimeout(timer);
      socket.destroy();
      resolve(live);
    };
    const timer = setTimeout(() => finish(true), 250);
    timer.unref?.();
    socket.once('connect', () => finish(true));
    socket.once('error', (error: NodeJS.ErrnoException) => {
      finish(error.code !== 'ENOENT' && error.code !== 'ECONNREFUSED');
    });
  });
}

/** Remove the pathname only when it still names the socket this server originally bound. */
async function unlinkIfOwned(path: string, identity: { dev: number; ino: number }): Promise<void> {
  try {
    const stat = await lstat(path);
    if (stat.dev === identity.dev && stat.ino === identity.ino) await unlink(path);
  } catch (error) {
    if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error;
  }
}

/** Unlink, swallowing "not found" (a stale socket is the normal pre-bind state). */
async function safeUnlink(path: string): Promise<void> {
  try {
    await unlink(path);
  } catch (err) {
    const code = (err as NodeJS.ErrnoException).code;
    if (code !== 'ENOENT') {
      // A different error (EACCES, EISDIR) is a real bind-blocker — let it surface.
      throw err;
    }
  }
}