/**
 * The remote-ops broker — stage 3's answering half (hook-process-boundary,
 * design D12).
 *
 * A hook holds no SSH credential: the jail does not bind `~/.ssh`, so a
 * hand-built `ssh` cannot authenticate anywhere. `@celilo/capabilities`'
 * remote primitives detect the hook environment and send each operation here
 * as a STRUCTURED request. This side holds the key, checks the target against
 * the policy the caller injected, and runs the real primitive.
 *
 * Structured, never a shell string: a broker that accepted a command line
 * from the jail would run arbitrary code as celilo OUTSIDE the jail, which is
 * the exact thing the boundary exists to prevent. The same reasoning refuses
 * `opts.env` (it never crosses; the asking half says so loudly) and confines
 * the stream primitives' LOCAL paths to the directories this run already owns
 * — without that check `streamBackup` would be a write-anything-as-celilo
 * oracle aimed at `master.key`.
 *
 * **Attribution (D12).** Every operation on this channel belongs to the
 * module whose hook is running — the policy is CONSTRUCTED for that module by
 * the invoking service. A capability provider's internal transport (the
 * `public_web` upload) does not pass through here at all: providers run in
 * celilo's own process (design D10), and the one hand-built-ssh provider call
 * site is being replaced by an Ansible converge under
 * `openspec/changes/capability-owned-tables` (celilo#1014). The same
 * `checkTarget` contract serves a provider-owned policy if provider transport
 * is ever brokered — the owner is a construction argument, never inferred
 * from a caller.
 *
 * Unlike the capability socket (one connection, one hook), this socket
 * answers MANY short-lived connections: the asking half spawns a client per
 * call, one request line in, one response line out.
 *
 * **A remote operation blocks celilo's event loop for its duration** — the
 * primitives are execSync-backed, bounded by each call's own `timeoutMs`.
 * That is not new exposure: before the process boundary every hook ran its
 * ssh on this same loop, and provider transport (design D10) still does.
 * The asking side is blocked in spawnSync anyway, so nothing concurrent is
 * lost; timers (the hook's total/idle kill) fire late by at most one op.
 *
 * Execution function (Rule 10.1) — owns the socket and performs the calls.
 */

import { lstatSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from 'node:fs';
import { type Server, createServer } from 'node:net';
import { tmpdir } from 'node:os';
import { basename, dirname, join, resolve, sep } from 'node:path';
import {
  type RemoteTarget,
  type RunResult,
  type Runner,
  execRunner,
  installAuthorizedKey,
  remoteExec,
  streamBackup,
  streamRestore,
} from '@celilo/capabilities';
import { z } from 'zod';
import { createLineReader } from './hook-protocol';
import type { HookLogger } from './types';

/** The socket's filename beside the capability socket ('s') in the run dir. */
const REMOTE_SOCKET_NAME = 'r';

/** Refuse a request line larger than this rather than buffering it. */
const MAX_REQUEST_BYTES = 64 * 1024 * 1024;

const TargetSchema = z.object({
  ipv4_address: z.string().min(1),
  user: z.string().optional(),
  port: z.number().int().optional(),
  /**
   * The module's OWN key, as content (task 5.4). The asking half reads the
   * file inside the jail, so a hook can only send bytes it may already read
   * — never a path for celilo to open on its behalf.
   */
  identityContent: z.string().optional(),
});

const OptsSchema = z
  .object({
    timeoutMs: z.number().optional(),
    input: z.string().optional(),
  })
  .optional();

const RemoteRequestSchema = z.discriminatedUnion('op', [
  z.object({
    op: z.literal('remoteExec'),
    target: TargetSchema,
    command: z.string(),
    opts: OptsSchema,
  }),
  z.object({
    op: z.literal('streamBackup'),
    target: TargetSchema,
    producerCommand: z.string(),
    localPath: z.string(),
    opts: OptsSchema,
  }),
  z.object({
    op: z.literal('streamRestore'),
    target: TargetSchema,
    localPath: z.string(),
    consumerCommand: z.string(),
    opts: OptsSchema,
  }),
  z.object({
    op: z.literal('installAuthorizedKey'),
    target: TargetSchema,
    password: z.string(),
    opts: OptsSchema,
  }),
  z.object({ op: z.literal('checkTarget'), target: TargetSchema }),
]);

type RemoteRequest = z.infer<typeof RemoteRequestSchema>;
type BridgeTarget = z.infer<typeof TargetSchema>;

export type RemoteCheckResult = { allowed: true } | { allowed: false; message: string };

/**
 * D12's reachability policy, built by the invoking service for the module the
 * run belongs to (`services/remote-access.ts`). The broker only enforces it.
 */
export interface RemoteAccessPolicy {
  /** The module every operation on this channel is attributed to. */
  moduleId: string;
  /**
   * May this module reach `target`? Called per operation, so it reads fresh
   * state. `hasOwnCredential` is true when the request carries the module's
   * own key or password rather than riding celilo's fleet key.
   */
  checkTarget(
    target: { ipv4_address: string; user?: string },
    hasOwnCredential: boolean,
  ): RemoteCheckResult;
}

export interface RemoteBrokerOptions {
  /** Directory holding the capability socket; the remote socket sits beside it. */
  socketDir: string;
  /**
   * Absent means DENY: a run started without a policy refuses every
   * operation, naming the gap. Fail closed, never open (Rule 6.4).
   */
  policy?: RemoteAccessPolicy;
  /** Local roots `streamRestore` may read from. Realpath'd here once. */
  readableRoots: readonly string[];
  /** Local roots `streamBackup` may write into. Realpath'd here once. */
  writableRoots: readonly string[];
  logger: HookLogger;
  /** Every operation feeds the idle tracker — remote work is real work. */
  onActivity: () => void;
  /** Injectable transport for tests (default: the real execRunner). */
  runner?: Runner;
}

export interface RemoteBroker {
  /** Unix socket path, handed to the child as CELILO_HOOK_REMOTE_SOCKET. */
  socketPath: string;
  /** Refuse further operations. Idempotent. Called on kill and on cleanup. */
  stop(): void;
  /** Close the listener. The socket file goes with the run directory. */
  close(): void;
}

/** Serve remote operations for one hook run. */
export async function startRemoteBroker(options: RemoteBrokerOptions): Promise<RemoteBroker> {
  const socketPath = join(options.socketDir, REMOTE_SOCKET_NAME);
  const runner = options.runner ?? execRunner;
  const readableRoots = options.readableRoots.map(realpathOrSelf);
  const writableRoots = options.writableRoots.map(realpathOrSelf);
  let stopped = false;

  const server: Server = createServer((socket) => {
    socket.setEncoding('utf-8');
    let answered = false;
    let received = 0;

    const answer = (response: object) => {
      if (answered) return;
      answered = true;
      socket.end(`${JSON.stringify(response)}\n`);
    };

    const feed = createLineReader((line) => {
      if (answered) return;
      options.onActivity();

      let parsed: RemoteRequest;
      try {
        parsed = RemoteRequestSchema.parse(JSON.parse(line));
      } catch (error) {
        answer(
          failure(
            `remote-ops broker: malformed request (${error instanceof Error ? error.message : String(error)})`,
          ),
        );
        return;
      }

      if (stopped) {
        // Same rule as the capability broker: the kill is only real if the
        // channel refuses afterwards (celilo#1003).
        answer(refusal(parsed.op, 'Hook run has ended; refusing remote operation.'));
        return;
      }

      answer(dispatch(parsed));
    });

    socket.on('data', (chunk: string) => {
      received += chunk.length;
      if (received > MAX_REQUEST_BYTES) {
        answer(failure('remote-ops broker: request too large'));
        socket.destroy();
        return;
      }
      feed(chunk);
    });
    socket.on('error', () => {
      // A client that died mid-request has nobody to answer.
    });
  });

  function dispatch(request: RemoteRequest): object {
    const hasOwnCredential =
      request.op === 'installAuthorizedKey' || request.target.identityContent !== undefined;

    const verdict: RemoteCheckResult = options.policy
      ? options.policy.checkTarget(
          { ipv4_address: request.target.ipv4_address, user: request.target.user },
          hasOwnCredential,
        )
      : {
          allowed: false,
          message:
            'This hook invocation was started without a remote-access policy, so remote operations are unavailable. This is a celilo defect — the invoking service must pass one.',
        };

    if (!verdict.allowed) {
      options.logger.warn(`remote-ops refused: ${verdict.message}`);
      return refusal(request.op, verdict.message);
    }

    try {
      return { ...perform(request) };
    } catch (error) {
      return failure(
        `remote-ops broker: ${error instanceof Error ? error.message : String(error)}`,
      );
    }
  }

  function perform(request: RemoteRequest): RunResult | RemoteCheckResult {
    switch (request.op) {
      case 'checkTarget':
        return { allowed: true };

      case 'remoteExec':
        return withIdentity(request.target, (target) =>
          remoteExec(target, request.command, request.opts ?? {}, runner),
        );

      case 'streamBackup': {
        const local = containedPath(request.localPath, writableRoots, 'write');
        if ('error' in local) return { ok: false, stdout: '', stderr: local.error };
        return withIdentity(request.target, (target) =>
          streamBackup(target, request.producerCommand, local.path, runner, request.opts ?? {}),
        );
      }

      case 'streamRestore': {
        const local = containedPath(request.localPath, readableRoots, 'read');
        if ('error' in local) return { ok: false, stdout: '', stderr: local.error };
        return withIdentity(request.target, (target) =>
          streamRestore(target, local.path, request.consumerCommand, runner, request.opts ?? {}),
        );
      }

      case 'installAuthorizedKey':
        return withIdentity(request.target, (target) =>
          installAuthorizedKey(target, request.password, runner, request.opts ?? {}),
        );
    }
  }

  await new Promise<void>((resolvePromise, reject) => {
    server.once('error', reject);
    server.listen(socketPath, resolvePromise);
  });

  return {
    socketPath,
    stop: () => {
      stopped = true;
    },
    close: () => {
      stopped = true;
      server.close();
    },
  };
}

/** A refusal in the shape the refused op's caller reads. */
function refusal(op: RemoteRequest['op'], message: string): object {
  return op === 'checkTarget' ? { allowed: false, message } : failure(message);
}

function failure(message: string): { ok: false; stdout: ''; stderr: string } {
  return { ok: false, stdout: '', stderr: message };
}

/**
 * Materialise the module's own key for the one call (task 5.4), then remove
 * it. Content in, path out: the asking half read the bytes inside the jail,
 * and the file below exists only for ssh's `-i`, which takes no stdin.
 */
function withIdentity(target: BridgeTarget, run: (target: RemoteTarget) => RunResult): RunResult {
  const base: RemoteTarget = {
    ipv4_address: target.ipv4_address,
    ...(target.user !== undefined ? { user: target.user } : {}),
    ...(target.port !== undefined ? { port: target.port } : {}),
  };
  if (target.identityContent === undefined) return run(base);

  const dir = mkdtempSync(join(tmpdir(), 'celilo-hook-identity-'));
  const keyPath = join(dir, 'key');
  try {
    writeFileSync(keyPath, target.identityContent, { mode: 0o600 });
    return run({ ...base, identityFile: keyPath });
  } finally {
    rmSync(dir, { recursive: true, force: true });
  }
}

/**
 * Confine a stream primitive's LOCAL path to this run's granted roots.
 *
 * Every component is resolved through the filesystem before the containment
 * test. For a read, `realpath` of the whole path does that. For a write the
 * leaf may not exist yet, so the PARENT is realpath'd — and a leaf that does
 * exist as a symlink is refused outright: the shell's `>` follows it, so a
 * link planted inside a writable directory (`state/evil` →
 * `/var/celilo/master.key`, dangling inside the jail, resolving outside it)
 * would otherwise aim celilo's write anywhere. Refusing the link beats
 * resolving it, because the hook's own subprocesses could re-point it
 * between a check and the write.
 *
 * The mount set binds the same path inside and outside the jail, which is
 * what makes a contained path name the same file for both sides — and why a
 * path OUTSIDE the roots (a bare `/tmp/x` under the run's private tmpfs)
 * must be refused: celilo would write a file the hook then cannot even see.
 */
function containedPath(
  localPath: string,
  roots: readonly string[],
  access: 'read' | 'write',
): { path: string } | { error: string } {
  let real: string;
  try {
    if (access === 'read') {
      real = realpathSync(resolve(localPath));
    } else {
      const resolved = resolve(localPath);
      real = join(realpathSync(dirname(resolved)), basename(resolved));
      // lstat, not existsSync: a dangling link (its target absent inside the
      // jail) still IS a link, and that is the case that aims celilo's write
      // out of the jail. existsSync follows the link and misses it.
      if (isSymlink(real)) {
        return {
          error: `remote-ops broker: local path '${localPath}' is a symlink; a stream write follows links, so it must name the file itself.`,
        };
      }
    }
  } catch (error) {
    return {
      error: `remote-ops broker: local path '${localPath}' is not usable (${error instanceof Error ? error.message : String(error)})`,
    };
  }

  if (roots.some((root) => real === root || real.startsWith(root + sep))) return { path: real };
  return {
    error: `remote-ops broker: local path '${localPath}' is outside this run's ${access === 'write' ? 'writable' : 'readable'} directories. Use ctx.stateDir or a declared path input (backup_dir, restore_dir).`,
  };
}

function realpathOrSelf(path: string): string {
  try {
    return realpathSync(resolve(path));
  } catch {
    return resolve(path);
  }
}

/** True if `path` itself is a symlink, dangling target included. */
function isSymlink(path: string): boolean {
  try {
    return lstatSync(path).isSymbolicLink();
  } catch {
    return false;
  }
}