/** Cross-process Telegram update inbox helpers. */

import { existsSync } from "node:fs";
import { appendFile, mkdir, readFile, stat, writeFile } from "node:fs/promises";
import { dirname } from "node:path";

export interface TelegramInboxItem<TUpdate = any> {
  update: TUpdate;
  receivedAt: number;
}

export interface TelegramInboxDelta {
  text: string;
  cursor: number;
}

export interface TelegramInboxParseResult<TUpdate = any> {
  valid: TelegramInboxItem<TUpdate>[];
  invalid: Array<{ line: string; error: string }>;
}

export async function ensureTelegramInbox(path: string): Promise<void> {
  await mkdir(dirname(path), { recursive: true });
  if (!existsSync(path)) await writeFile(path, "", { mode: 0o600 });
}

export async function appendTelegramInboxItem<TUpdate>(
  path: string,
  item: TelegramInboxItem<TUpdate>,
): Promise<void> {
  await ensureTelegramInbox(path);
  await appendFile(path, JSON.stringify(item) + "\n", "utf8");
}

export async function getTelegramInboxSize(path: string): Promise<number> {
  await ensureTelegramInbox(path);
  return (await stat(path)).size;
}

export async function readTelegramInboxDelta(path: string, cursor: number): Promise<TelegramInboxDelta> {
  await ensureTelegramInbox(path);
  const content = await readFile(path);
  const safeCursor = Math.max(0, Math.min(cursor, content.length));
  return {
    text: content.subarray(safeCursor).toString("utf8"),
    cursor: content.length,
  };
}

export function parseTelegramInboxLines<TUpdate = any>(text: string): TelegramInboxParseResult<TUpdate> {
  const valid: TelegramInboxItem<TUpdate>[] = [];
  const invalid: Array<{ line: string; error: string }> = [];
  for (const line of text.split("\n")) {
    if (!line.trim()) continue;
    try {
      valid.push(JSON.parse(line) as TelegramInboxItem<TUpdate>);
    } catch (error) {
      invalid.push({ line, error: error instanceof Error ? error.message : String(error) });
    }
  }
  return { valid, invalid };
}

export interface TelegramInboxConsumerOptions<TUpdate> {
  inboxPath: string;
  initialCursor?: number;
  onUpdate: (update: TUpdate) => Promise<void> | void;
  onInvalidLine: (invalid: { line: string; error: string }) => void;
  onCursor?: (cursor: number) => Promise<void> | void;
}

export interface TelegramInboxConsumer {
  pollOnce: () => Promise<void>;
  start: () => void;
  stop: () => void;
}

export function createTelegramInboxConsumer<TUpdate>(
  options: TelegramInboxConsumerOptions<TUpdate>,
): TelegramInboxConsumer {
  let cursor = options.initialCursor ?? 0;
  let timer: ReturnType<typeof setInterval> | undefined;
  const pollOnce = async () => {
    const delta = await readTelegramInboxDelta(options.inboxPath, cursor);
    const parsed = parseTelegramInboxLines<TUpdate>(delta.text);
    for (const invalid of parsed.invalid) options.onInvalidLine(invalid);
    for (const item of parsed.valid) await options.onUpdate(item.update);
    cursor = delta.cursor;
    await options.onCursor?.(cursor);
  };
  return {
    pollOnce,
    start: () => {
      if (timer) return;
      timer = setInterval(() => {
        void pollOnce();
      }, 1000);
      timer.unref?.();
    },
    stop: () => {
      if (timer) clearInterval(timer);
      timer = undefined;
    },
  };
}