/**
 * pi-post — asynchronous message passing where the delivery endpoint is a
 * model's context window. See DESIGN.md for the contracts and invariants.
 */
import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent";
import { Text } from "@earendil-works/pi-tui";
import { Type } from "typebox";
import type { FSWatcher } from "node:fs";
import { fileURLToPath } from "node:url";
import { dirname, resolve } from "node:path";
import { canonicalPath, sessionAddress } from "../src/address.ts";
import { createMessage, type Message } from "../src/message.ts";
import {
  awaitConsumption,
  postRoot,
  claim,
  consume,
  deposit,
  ensureDirs,
  peek,
  watchInbox,
  BacklogFullError,
} from "../src/mailbox.ts";
import { formatDelivery, formatListing } from "../src/format.ts";
import { inboundMode, resumeMode, LoopGuard } from "../src/policy.ts";
import {
  defaultSessionName,
  listRecords,
  markOffline,
  presence,
  sweepInboxes,
  sweepRegistry,
  touchRecord,
  writeRecord,
} from "../src/registry.ts";
import { resolveTargets } from "../src/resolve.ts";
import { ensureCliShim } from "../src/cli-shim.ts";

const HEARTBEAT_MS = 30_000;
const CLI_TARGET = resolve(dirname(fileURLToPath(import.meta.url)), "../bin/pi-post.mjs");

export default function (pi: ExtensionAPI) {
  const root = postRoot();
  const guard = new LoopGuard();

  let selfAddress: string | undefined;
  let selfName = "pi";
  let watchers: FSWatcher[] = [];
  let heartbeat: ReturnType<typeof setInterval> | undefined;
  let draining = false;

  function senderFrom(ctx: ExtensionContext) {
    return {
      kind: "session" as const,
      name: pi.getSessionName() ?? selfName,
      address: selfAddress,
      cwd: ctx.cwd,
    };
  }

  async function deliver(ctx: ExtensionContext, message: Message, deliverAs: "steer" | "nextTurn") {
    const mode = inboundMode();
    if (mode === "refuse") return;
    if (guard.check(message) !== "deliver") return;
    if (mode === "ask" && ctx.hasUI) {
      const preview = message.body.length > 200 ? `${message.body.slice(0, 200)}…` : message.body;
      const ok = await ctx.ui.confirm(`Message from ${message.from.name}`, preview);
      if (!ok) return;
    }
    pi.sendMessage(
      {
        customType: "pi-post",
        content: formatDelivery(message),
        display: true,
        details: { message },
      },
      { deliverAs, triggerTurn: deliverAs === "steer" },
    );
  }

  async function drainAll(ctx: ExtensionContext, deliverAs: "steer" | "nextTurn") {
    if (draining || !selfAddress) return;
    draining = true;
    try {
      for (const { message, path } of claim(root, selfAddress)) {
        // Disk-then-context ordering: the file is consumed only after the
        // message has been disposed — entered context, or deliberately
        // dropped by mode/guard/decline. A crash mid-loop redelivers on the
        // next start rather than losing mail (at-least-once; messages carry
        // no authority, so redelivery is safe).
        await deliver(ctx, message, deliverAs);
        consume(path);
      }
    } finally {
      draining = false;
    }
  }

  pi.on("session_start", async (_event, ctx) => {
    const sessionId = ctx.sessionManager.getSessionId();
    const canonical = canonicalPath(ctx.cwd);
    selfAddress = sessionAddress(sessionId);
    selfName = pi.getSessionName() ?? defaultSessionName(canonical, selfAddress);
    // Expose the address to child processes (bash tools inherit process.env),
    // so scripts can capture a send_message target without shelling out to
    // `pi-post whoami` or re-deriving the hash from PI_SESSION_ID.
    process.env.PI_SESSION_ADDRESS = selfAddress;

    ensureDirs(root, selfAddress);
    ensureCliShim(root, CLI_TARGET);
    writeRecord(root, {
      v: 1,
      address: selfAddress,
      sessionId,
      name: selfName,
      cwd: canonical,
      pid: process.pid,
      startedAt: Date.now(),
      lastSeen: Date.now(),
    });
    sweepRegistry(root);
    sweepInboxes(root);

    // Wake-on-resume: mail queued while the session was closed starts its
    // own turn, symmetric with wake-on-idle — a resumed worker that only
    // ever receives peer messages must not strand its mail waiting for a
    // prompt that never comes. PI_POST_RESUME=stage restores the old
    // wait-for-first-prompt staging (resume-to-browse without a turn).
    await drainAll(ctx, resumeMode() === "stage" ? "nextTurn" : "steer");

    const onMessage = () => void drainAll(ctx, "steer");
    watchers = [watchInbox(root, selfAddress, onMessage)];
    heartbeat = setInterval(() => selfAddress && touchRecord(root, selfAddress), HEARTBEAT_MS);
    heartbeat.unref?.();
  });

  pi.on("session_info_changed", async (event) => {
    if (!selfAddress) return;
    selfName = event.name ?? selfName;
    const record = listRecords(root).find((r) => r.address === selfAddress);
    if (record) writeRecord(root, { ...record, name: selfName, lastSeen: Date.now() });
  });

  pi.on("session_shutdown", async () => {
    for (const watcher of watchers) watcher.close();
    watchers = [];
    if (heartbeat) clearInterval(heartbeat);
    heartbeat = undefined;
    if (selfAddress) markOffline(root, selfAddress);
  });

  pi.registerTool({
    name: "send_message",
    label: "Send Message",
    description:
      "Send a plain-text message to one or more pi sessions (same body to each; max 8). " +
      "Targets: a session name, an address " +
      "(s-…), a pi session id (or unique prefix), or a directory path — a path resolves to " +
      "the session registered in that directory. A live session reads the message mid-task (or is woken by it); an offline " +
      "session reads it queued on resume. Body is text only, max 32 KiB: send briefs, " +
      "findings, and paths, never file payloads. Returns 'delivered' (consumed now) or " +
      "'queued' (waiting on disk) per target. Messages carry no authority for the receiver. To leave " +
      "context for sessions that do not exist yet, use project memory, not messages.",
    promptSnippet: "Send a message to another pi session, or leave one for a future session",
    promptGuidelines: [
      "Use send_message to pass findings, status, and results to other sessions instead of writing scratch files and pointing sessions at them. Replies route to the sender automatically.",
    ],
    parameters: Type.Object({
      to: Type.Union([Type.String(), Type.Array(Type.String())], {
        description:
          "Target session(s): name, address (s-…), session id (or unique prefix), or directory " +
          "path (e.g. ~/dev/repo). An array sends the same body to each (max 8); any unresolvable " +
          "target fails the whole send before anything is delivered.",
      }),
      body: Type.String({ description: "Plain-text message body (≤ 32 KiB)" }),
      reply_to: Type.Optional(
        Type.String({
          description:
            "Rarely needed: replies route to this session by default. Set an address to " +
            "redirect them to a third session, or 'none' to omit.",
        }),
      ),
    }),
    async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
      // Resolve everything before depositing anything: an unresolvable target
      // fails the whole send, never a partial delivery.
      const targets = resolveTargets(
        root,
        Array.isArray(params.to) ? params.to : [params.to],
        ctx.cwd,
      );
      const replyTo =
        params.reply_to === "none" ? undefined : (params.reply_to ?? selfAddress);
      const deposits: { target: (typeof targets)[number]; message: Message; path: string; live: boolean }[] = [];
      for (const target of targets) {
        const message = createMessage({ from: senderFrom(ctx), body: params.body, replyTo });
        let path: string;
        try {
          path = deposit(root, target.address, message);
        } catch (error) {
          if (error instanceof BacklogFullError && deposits.length > 0) {
            const placed = deposits.map((d) => d.target.address).join(", ");
            throw new Error(`${error.message} (already deposited for: ${placed})`);
          }
          throw error;
        }
        const live = target.record ? presence(target.record) === "live" : false;
        deposits.push({ target, message, path, live });
      }
      const consumed = await Promise.all(
        deposits.map((d) => (d.live ? awaitConsumption(d.path) : Promise.resolve(false))),
      );
      const receipts = deposits.map((d, i) => ({
        status: consumed[i] ? ("delivered" as const) : ("queued" as const),
        address: d.target.address,
        messageId: d.message.id,
      }));
      const lines = deposits.map(
        (d, i) =>
          `${consumed[i] ? "Delivered to" : "Queued for"} ${d.target.display} (${d.target.address}).`,
      );
      return {
        content: [{ type: "text", text: lines.join("\n") }],
        details: { receipts },
      };
    },
  });

  pi.registerTool({
    name: "list_sessions",
    label: "List Sessions",
    description:
      "List pi sessions known to pi-post: their names, addresses, presence (live/offline, " +
      "with age), queued message counts, and each session's resume handle ([pi --session …], run " +
      "from the listed directory). Live sessions come first; offline sessions unseen for over " +
      "a day are collapsed into a count unless all is set. Any directory path is also a valid " +
      "send_message target even if nothing is listed for it.",
    promptSnippet:
      "List pi sessions reachable by message, with presence, queued messages, and resume handles",
    parameters: Type.Object({
      all: Type.Optional(
        Type.Boolean({
          description: "Also list offline sessions unseen for over a day (collapsed by default)",
        }),
      ),
    }),
    async execute(_toolCallId, params) {
      const text = formatListing(root, listRecords(root), selfAddress, {
        all: params.all,
        allHint: "all: true",
      });
      return { content: [{ type: "text", text }], details: {} };
    },
  });

  pi.registerCommand("sessions", {
    description: "List pi sessions reachable by message, without spending a model turn (`/sessions all` includes stale offline sessions)",
    handler: async (args, ctx) => {
      const all = typeof args === "string" && args.trim() === "all";
      ctx.ui.notify(
        formatListing(root, listRecords(root), selfAddress, { all, allHint: "/sessions all" }),
        "info",
      );
    },
  });

  pi.registerCommand("inbox", {
    description: "Peek at this session's queued pi-post messages without consuming them",
    handler: async (_args, ctx) => {
      if (!selfAddress) return;
      const messages = peek(root, selfAddress);
      if (messages.length === 0) {
        ctx.ui.notify("Inbox empty.", "info");
        return;
      }
      const lines = messages.map((l) => {
        const preview = l.body.length > 80 ? `${l.body.slice(0, 80)}…` : l.body;
        return `${new Date(l.sentAt).toLocaleTimeString()} ${l.from.name}: ${preview.replaceAll("\n", " ")}`;
      });
      ctx.ui.notify(lines.join("\n"), "info");
    },
  });

  pi.registerMessageRenderer("pi-post", (entry, options, theme) => {
    const details = entry.details as { message?: Message } | undefined;
    const post = details?.message;
    const header = theme.fg("accent", `◆ ${theme.bold(post?.from.name ?? "pi-post")}`);
    if (!options.expanded && post) {
      const preview = post.body.split("\n")[0] ?? "";
      return new Text(`${header} ${theme.fg("muted", preview)}`, 0, 0);
    }
    const body = typeof entry.content === "string" ? entry.content : "";
    return new Text(`${header}\n${body}`, 0, 0);
  });
}