/**
 * PluginState for dsh-telegram-remote: per-chat state, logging, push queue,
 * prompt-echo suppression.
 */

import { appendFile, rename, stat } from "node:fs/promises";
import { readFileSync, writeFileSync, renameSync } from "node:fs";
import { join } from "node:path";
import { dshHome, nowIso, sleep, shortId } from "./util.js";

const LOG_MAX_BYTES = 2 * 1024 * 1024;
const LOG_ROTATE_EVERY = 64; // check size only every N writes
const STATE_DEBOUNCE_MS = 250;

export class PluginState {
  ctx;
  config;
  bot;
  logPath;
  statePath;
  chats = new Map();
  recentPrompts = new Map();
  jobStatuses = new Map();
  pendingSends = new Map();
  pendingOpts = new Map();
  sendingChats = new Set();
  restartScriptPath;

  constructor(ctx, config, bot) {
    this.ctx = ctx;
    // Apply code-side defaults (the plugin deliberately ships no Config
    // schema, so loader entries may omit any key).
    this.config = {
      botToken: "",
      tokenEnv: "TELEGRAM_BOT_TOKEN",
      tokenFile: "",
      allowedUserIds: [],
      ownerChatId: undefined,
      workspaceRoot: process.cwd(),
      defaultSessionId: "",
      allowEval: true,
      notifyOnStartup: true,
      stateFile: "",
      logFile: "",
      pollTimeoutSec: 50,
      maxOutputBytes: 120000,
      whisperBaseUrl: "http://192.168.31.159:8082",
      whisperModel: "large-v2",
      whisperDevice: "cuda",
      whisperLanguage: "auto",
      whisperTimeoutMs: 120000,
      whisperMaxBytes: 20971520,
      ...(config ?? {}),
    };
    this.bot = bot;
    this.logPath = config.logFile || join(dshHome(), "telegram-remote.log");
    this.statePath = config.stateFile || join(dshHome(), "telegram-remote-state.json");
    this.restartScriptPath = join(dshHome(), "telegram-remote-restart.ps1");
    // Batched async log writer: log() is fire-and-forget and never blocks
    // the event loop (the old appendFileSync froze polling on every line).
    // Lines accumulate and flush in one appendFile per microtask drain.
    this._logBuffer = [];
    this._logFlushing = false;
    this._logWrites = 0;
    this._stateTimer = null;
    this._stateDirty = false;
    this.loadState();
  }

  log(message) {
    this._logBuffer.push("[" + nowIso() + "] " + message);
    try { this.ctx.logger.info("[telegram-remote] " + message); } catch {}
    if (!this._logFlushing) {
      this._logFlushing = true;
      queueMicrotask(() => {
        const lines = this._logBuffer.splice(0);
        this._logFlushing = false;
        this._writeLog(lines);
      });
    }
  }

  /** Append a batch of lines; rotate the file lazily every N batches. */
  async _writeLog(lines) {
    if (lines.length === 0) return;
    const path = this.logPath;
    try {
      await appendFile(path, lines.join("\n") + "\n", "utf8");
      if (++this._logWrites % LOG_ROTATE_EVERY === 0) {
        try {
          const size = (await stat(path)).size;
          if (size > LOG_MAX_BYTES) await rename(path, path + ".1");
        } catch {}
      }
    } catch {}
  }

  loadState() {
    try {
      const raw = JSON.parse(readFileSync(this.statePath, "utf8"));
      for (const [chatId, value] of Object.entries(raw.chats ?? {})) {
        this.chats.set(Number(chatId), {
          sessionId: value.sessionId ?? "",
          notify: value.notify ?? "off",
        });
      }
      if (typeof raw.botOffset === "number" && raw.botOffset > this.bot.offset) {
        this.bot.offset = raw.botOffset;
      }
    } catch {}
  }

  /** Schedule a state write; coalesces bursts (one write per 250ms window). */
  saveState() {
    this._stateDirty = true;
    if (this._stateTimer) return;
    this._stateTimer = setTimeout(() => {
      this._stateTimer = null;
      if (this._stateDirty) {
        this._stateDirty = false;
        this.writeState();
      }
    }, STATE_DEBOUNCE_MS);
  }

  /** Write the state file immediately (reboot/shutdown/dispose paths). */
  flushState() {
    if (this._stateTimer) {
      clearTimeout(this._stateTimer);
      this._stateTimer = null;
    }
    this._stateDirty = false;
    this.writeState();
  }

  writeState() {
    const payload = {
      chats: Object.fromEntries([...this.chats.entries()].map(([id, v]) => [String(id), v])),
      botOffset: this.bot?.offset ?? 0,
    };
    try {
      // Atomic: write a sibling temp file then rename, so a crash mid-write
      // never leaves a truncated state file behind.
      const tmp = this.statePath + ".tmp";
      writeFileSync(tmp, JSON.stringify(payload, null, 2), "utf8");
      try { renameSync(tmp, this.statePath); } catch { writeFileSync(this.statePath, JSON.stringify(payload, null, 2), "utf8"); }
    } catch (error) {
      this.log("state save failed: " + error.message);
    }
  }

  chatState(chatId) {
    let state = this.chats.get(chatId);
    if (!state) {
      state = { sessionId: this.config.defaultSessionId || "", notify: "session", sessionIds: [], lastListAt: 0 };
      this.chats.set(chatId, state);
    }
    return state;
  }

  isAuthorized(userId) {
    if (this.config.ownerChatId != null && userId === this.config.ownerChatId) return true;
    return Array.isArray(this.config.allowedUserIds) && this.config.allowedUserIds.includes(userId);
  }

  async listSessions() {
    const { callApi } = await import("./util.js");
    try {
      const result = await callApi(this.ctx, "sessions", "list", {});
      const items = Array.isArray(result) ? result : (result?.items ?? []);
      if (items.length > 0) return items;
    } catch {}
    // Fallback for profiles without the web api-proxy: read the live/corpus
    // registries directly.
    const items = [];
    const sessions = this.ctx.get("sessions");
    const sessionQuery = this.ctx.get("sessionQuery");
    const seen = new Set();
    if (sessions) {
      for (const session of sessions.list()) {
        seen.add(session.id);
        items.push({
          sessionId: session.id,
          updatedAt: session.header?.updatedAt ?? session.header?.createdAt ?? 0,
          running: this.ctx.get("agents")?.get(session.id)?.status === "running",
          blank: !session.events.some((event) => event.type === "turn/start"),
          cwd: session.header?.cwd,
          origin: session.header?.origin,
        });
      }
    }
    if (sessionQuery) {
      try {
        for (const record of await sessionQuery.listSessions()) {
          if (seen.has(record.header.id)) continue;
          items.push({
            sessionId: record.header.id,
            updatedAt: record.header.createdAt ?? 0,
            running: false,
            blank: true,
            cwd: record.header.cwd,
            origin: record.header.origin,
          });
        }
      } catch {}
    }
    items.sort((a, b) => (b.updatedAt ?? 0) - (a.updatedAt ?? 0));
    return items;
  }

  /** Resolve a session id argument: full id, prefix, "last", or the chat's active session. */
  async resolveSessionId(chatId, arg) {
    const wanted = String(arg ?? "").trim();
    let sessionId = wanted;
    if (!sessionId) {
      sessionId = this.chatState(chatId).sessionId;
      if (!sessionId) return null;
      return sessionId;
    }
    if (sessionId === "last") {
      const summaries = await this.listSessions();
      if (!summaries || summaries.length === 0) return null;
      return summaries[0].sessionId;
    }
    const summaries = await this.listSessions();
    if (summaries) {
      const match = summaries.find((item) => item.sessionId === sessionId);
      const prefix = summaries.find((item) => item.sessionId.startsWith(sessionId));
      return (match ?? prefix ?? null)?.sessionId ?? sessionId;
    }
    return sessionId;
  }

  notePrompt(sessionId, text) {
    const list = this.recentPrompts.get(sessionId) ?? [];
    list.push({ text: String(text).trim().slice(0, 200), at: Date.now() });
    while (list.length > 20) list.shift();
    this.recentPrompts.set(sessionId, list);
    // Bounded memory: drop sessions whose prompts all expired long ago.
    if (this.recentPrompts.size > 64) {
      const now = Date.now();
      for (const [sid, entries] of this.recentPrompts) {
        const fresh = entries.filter((entry) => now - entry.at < 600_000);
        if (fresh.length === 0) this.recentPrompts.delete(sid);
        else this.recentPrompts.set(sid, fresh);
      }
    }
  }

  isRecentPrompt(sessionId, text) {
    const list = this.recentPrompts.get(sessionId) ?? [];
    const needle = String(text).trim().slice(0, 200);
    const now = Date.now();
    return list.some((entry) => now - entry.at < 90_000 && entry.text === needle);
  }

  /** Record a job status transition; keeps only the most recent entries. */
  noteJobStatus(jobId, status) {
    this.jobStatuses.set(jobId, status);
    if (this.jobStatuses.size > 256) {
      const oldest = this.jobStatuses.keys().next().value;
      this.jobStatuses.delete(oldest);
    }
  }

  async #drain(chatId) {
    if (this.sendingChats.has(chatId)) return;
    this.sendingChats.add(chatId);
    try {
      for (;;) {
        const text = this.pendingSends.get(chatId);
        if (text === undefined) break;
        this.pendingSends.delete(chatId);
        const opts = this.pendingOpts.get(chatId) ?? {};
        this.pendingOpts.delete(chatId);
        try {
          await this.bot.send(chatId, text, opts.replyTo ? { replyToMessageId: opts.replyTo } : {});
        } catch (error) {
          this.log("push to chat " + chatId + " failed: " + error.message);
          const again = this.pendingSends.get(chatId);
          this.pendingSends.set(chatId, (again ? again + "\n\n" : "") + text);
          this.pendingOpts.set(chatId, opts);
          await sleep(5000);
          break;
        }
      }
    } finally {
      this.sendingChats.delete(chatId);
    }
  }

  /**
   * Push a session event to interested chats.
   * @param category - "reply" (assistant replies, turn ends: delivered to
   *   every chat with notify session/all) or "detail" (tool calls, web-side
   *   user messages, errors: only notify=all chats — too noisy otherwise).
   */
  notifySession(sessionId, text, category = "reply") {
    for (const [chatId, state] of this.chats) {
      if (state.notify === "off") continue;
      if (state.notify === "session" && sessionId !== state.sessionId) continue;
      if (category === "detail" && state.notify !== "all") continue;
      // The active session reads as a plain conversation — no session id
      // prefix. Other sessions (notify=all) get a small label to distinguish.
      const body = sessionId === state.sessionId
        ? text
        : "<b>" + shortId(sessionId) + "</b>\n" + text;
      this.push(chatId, body, { replyTo: state.lastUserMessageId });
    }
  }

  /** Coalescing push sender: one in-flight send per chat, pending text merged. */
  push(chatId, text, opts = {}) {
    const existing = this.pendingSends.get(chatId);
    if (existing !== undefined) {
      this.pendingSends.set(chatId, existing + "\n\n" + text);
      return;
    }
    this.pendingSends.set(chatId, text);
    this.pendingOpts.set(chatId, opts);
    void this.#drain(chatId);
  }

  notifyAll(text) {
    for (const chatId of this.chats.keys()) this.push(chatId, text);
  }
}