clinera-site/telegram-bot/index.ts

import { Bot, Context } from "grammy";
import { run } from "@grammyjs/runner";
import Anthropic from "@anthropic-ai/sdk";
import { readFileSync } from "fs";
import { resolve } from "path";

// ─── Config ───
const BOT_TOKEN = process.env.TELEGRAM_BOT_TOKEN!;
const ANTHROPIC_KEY = process.env.ANTHROPIC_API_KEY!;
const SSH_USER = process.env.SSH_USER || "root";
const SSH_HOST = process.env.SSH_HOST!;
const SSH_PORT = process.env.SSH_PORT || "22";
const ALLOWED_USERS = process.env.TELEGRAM_ALLOWED_USERS?.split(",").map(Number) || [];

if (!BOT_TOKEN) throw new Error("TELEGRAM_BOT_TOKEN missing from .env");
if (!ANTHROPIC_KEY) throw new Error("ANTHROPIC_API_KEY missing from .env");
if (!SSH_HOST) throw new Error("SSH_HOST missing from .env");

// ─── Load infra context ───
let infraContext = "";
try {
  infraContext = readFileSync(resolve(__dirname, "../.pi/infra.md"), "utf-8");
} catch {
  console.warn("⚠️  Could not load .pi/infra.md");
}

// ─── Init ───
const bot = new Bot(BOT_TOKEN);
const anthropic = new Anthropic({ apiKey: ANTHROPIC_KEY });

// Per-user conversation history
const conversations = new Map<number, Anthropic.MessageParam[]>();
const MAX_HISTORY = 30; // raw entries before compression triggers
const MAX_TOKENS_ESTIMATE = 12000; // compress when estimated tokens exceed this

// Per-user abort controllers — cancel stale requests
const activeRequests = new Map<number, AbortController>();

// ─── Context compression ───
function estimateTokens(messages: Anthropic.MessageParam[]): number {
  let chars = 0;
  for (const msg of messages) {
    if (typeof msg.content === "string") {
      chars += msg.content.length;
    } else if (Array.isArray(msg.content)) {
      for (const block of msg.content) {
        if ("text" in block) chars += (block as any).text.length;
        else if ("content" in block) chars += String((block as any).content).length;
        else chars += 100; // tool_use blocks etc
      }
    }
  }
  return Math.ceil(chars / 4); // rough chars-to-tokens
}

async function compressHistory(userId: number, history: Anthropic.MessageParam[]): Promise<void> {
  if (history.length < 10 || estimateTokens(history) < MAX_TOKENS_ESTIMATE) return;

  // Keep last 6 messages intact (current conversation), compress the rest
  const toCompress = history.slice(0, -6);
  const toKeep = history.slice(-6);

  // Build a text summary of old messages
  const summaryParts: string[] = [];
  for (const msg of toCompress) {
    if (typeof msg.content === "string") {
      const role = msg.role === "user" ? "User" : "Assistant";
      summaryParts.push(`${role}: ${msg.content.slice(0, 200)}`);
    } else if (Array.isArray(msg.content)) {
      for (const block of msg.content) {
        if ("text" in block && (block as any).text) {
          const role = msg.role === "user" ? "User" : "Assistant";
          summaryParts.push(`${role}: ${(block as any).text.slice(0, 200)}`);
        } else if ("content" in block && (block as any).type === "tool_result") {
          summaryParts.push(`[cmd output: ${String((block as any).content).slice(0, 100)}]`);
        } else if ((block as any).type === "tool_use") {
          summaryParts.push(`[ran: ${JSON.stringify((block as any).input).slice(0, 100)}]`);
        }
      }
    }
  }

  const compressedSummary = summaryParts.join("\n").slice(0, 2000);

  // Replace history with compressed summary + recent messages
  history.length = 0;
  history.push({
    role: "user",
    content: `[Previous conversation summary]\n${compressedSummary}\n[End summary — continue from here]`,
  });
  history.push({
    role: "assistant",
    content: "Understood, I have the context from our previous conversation. How can I help?",
  });
  history.push(...toKeep);

  console.log(`📦 [${userId}] Compressed history: ${toCompress.length + toKeep.length} → ${history.length} entries (~${estimateTokens(history)} tokens)`);
}

// ─── Auth ───
function isAuthorized(ctx: Context): boolean {
  if (ALLOWED_USERS.length === 0) return false;
  return ALLOWED_USERS.includes(ctx.from?.id || 0);
}

// ─── Rate limiting ───
const rateLimits = new Map<number, number[]>();
const RATE_LIMIT_WINDOW = 60_000;
const RATE_LIMIT_MAX = 15;

function isRateLimited(userId: number): boolean {
  const now = Date.now();
  const timestamps = (rateLimits.get(userId) || []).filter(t => now - t < RATE_LIMIT_WINDOW);
  timestamps.push(now);
  rateLimits.set(userId, timestamps);
  return timestamps.length > RATE_LIMIT_MAX;
}

// ─── Blocked commands ───
const BLOCKED_PATTERNS = [
  /rm\s+-rf\s+\/(?!\w)/i,
  /mkfs\./i,
  /dd\s+if=.*of=\/dev/i,
  /:\(\)\s*\{\s*:\|:/,
];

function isDangerous(cmd: string): string | null {
  for (const pattern of BLOCKED_PATTERNS) {
    if (pattern.test(cmd)) return `⛔ Blocked dangerous pattern`;
  }
  return null;
}

// ─── SSH exec ───
async function sshExec(command: string, timeoutSec = 30): Promise<string> {
  const proc = Bun.spawn(
    ["ssh", "-o", "ConnectTimeout=5", "-o", "StrictHostKeyChecking=accept-new",
     "-p", SSH_PORT, `${SSH_USER}@${SSH_HOST}`, command],
    { stdout: "pipe", stderr: "pipe" }
  );

  const timer = setTimeout(() => proc.kill(), timeoutSec * 1000);
  const [stdout, stderr] = await Promise.all([
    new Response(proc.stdout).text(),
    new Response(proc.stderr).text(),
  ]);
  clearTimeout(timer);
  await proc.exited;

  const output = (stdout + stderr).trim();
  return output.length > 2000 ? output.slice(0, 2000) + "\n...(truncated)" : output;
}

// ─── Promise with timeout helper ───
function withTimeout<T>(promise: Promise<T>, ms: number, label: string): Promise<T> {
  return Promise.race([
    promise,
    new Promise<never>((_, reject) =>
      setTimeout(() => reject(new Error(`${label} timed out after ${ms / 1000}s`)), ms)
    ),
  ]);
}

// ─── System prompt ───
const SYSTEM_PROMPT = `You are a DevOps assistant on Telegram. Connected to a live server via SSH.

${infraContext}

ARCHITECTURE:
- The bot runs on a LOCAL LAPTOP (macOS). It is NOT on the server.
- run_command executes via SSH on the PRIMARY HOST (${SSH_HOST}).
- The primary host runs HAProxy and Incus (LXC containers).
- Docker runs INSIDE the Incus container "cr-server-new" (IP 10.213.16.224), NOT on the primary host.
- To run Docker/sync commands, you MUST wrap with: incus exec cr-server-new -- <command>
- Example: incus exec cr-server-new -- docker ps
- Example: incus exec cr-server-new -- bash /opt/enthuse-db-sync-v2/run-sync.sh
- To query Postgres: incus exec cr-server-new -- docker exec dokploy-migrated-cr-postgres-data psql -U postgres -d donation_warehouse -c "SELECT ..."
- NEVER run docker commands directly on the primary host — they won't work.

KEY PATHS (inside cr-server-new):
- /opt/enthuse-db-sync-v2/ — Enthuse sync
- /opt/launchgood-sync/ — LaunchGood sync  
- /opt/charityright-sync/ — CharityRight sync
- Postgres container: dokploy-migrated-cr-postgres-data
- Docker network: dokploy-network

RULES:
- Be VERY concise — this is Telegram, not a terminal.
- Keep command output short. Use | head, | tail, grep to limit output.
- For destructive commands, WARN and ask for confirmation first.
- ALWAYS end your response with a text message to the user. Never end on a tool call with no text summary.`;

// ─── Claude tools ───
const tools: Anthropic.Tool[] = [
  {
    name: "run_command",
    description: "Run a shell command on the server via SSH. Keep commands short and focused. Always pipe through head/tail/grep to limit output.",
    input_schema: {
      type: "object" as const,
      properties: {
        command: { type: "string", description: "Shell command" },
      },
      required: ["command"],
    },
  },
];

// ─── Chat with Claude — with cancellation + timeout ───
// Global typing context so SSH exec can trigger typing too
let activeCtx: Context | null = null;

async function chat(userId: number, message: string, signal: AbortSignal): Promise<string> {
  if (!conversations.has(userId)) conversations.set(userId, []);
  const history = conversations.get(userId)!;

  // Snapshot history length before we start — rollback to here on error
  const historySnapshot = history.length;

  history.push({ role: "user", content: message });

  // Compress old history if getting too large
  await compressHistory(userId, history);
  while (history.length > MAX_HISTORY) history.shift();

  try {
    return await _chatInner(userId, history, signal);
  } catch (err) {
    // Rollback history to prevent corruption (dangling tool_use without tool_result)
    history.splice(historySnapshot);
    throw err;
  }
}

async function _chatInner(userId: number, history: Anthropic.MessageParam[], signal: AbortSignal): Promise<string> {
  let rounds = 0;
  const MAX_ROUNDS = 10;

  let response = await withTimeout(
    anthropic.messages.create({
      model: "claude-sonnet-4-20250514",
      max_tokens: 1024,
      system: SYSTEM_PROMPT,
      tools,
      messages: history,
    }),
    90_000,
    "Claude API"
  );

  while (response.stop_reason === "tool_use" && rounds < MAX_ROUNDS) {
    if (signal.aborted) throw new Error("Cancelled — new message received");
    rounds++;

    const assistantContent = response.content;
    history.push({ role: "assistant", content: assistantContent });

    const toolResults: Anthropic.ToolResultBlockParam[] = [];

    for (const block of assistantContent) {
      if (block.type === "tool_use") {
        if (signal.aborted) throw new Error("Cancelled");
        const input = block.input as { command: string };
        console.log(`🔧 [${userId}] ${input.command}`);

        const danger = isDangerous(input.command);
        if (danger) {
          toolResults.push({ type: "tool_result", tool_use_id: block.id, content: danger, is_error: true });
          continue;
        }
        try {
          // Show what's running
          if (activeCtx) {
            const shortCmd = input.command.length > 80 ? input.command.slice(0, 80) + "..." : input.command;
            await activeCtx.reply(`⚙️ \`${shortCmd}\``, { parse_mode: "Markdown" }).catch(() => {});
          }
          const output = await sshExec(input.command);
          toolResults.push({ type: "tool_result", tool_use_id: block.id, content: output || "(no output)" });
        } catch (err: any) {
          toolResults.push({ type: "tool_result", tool_use_id: block.id, content: `ERROR: ${err.message}`, is_error: true });
        }
      }
    }

    // Always push tool_results immediately after assistant tool_use — keeps history valid
    history.push({ role: "user", content: toolResults });

    response = await withTimeout(
      anthropic.messages.create({
        model: "claude-sonnet-4-20250514",
        max_tokens: 1024,
        system: SYSTEM_PROMPT,
        tools,
        messages: history,
      }),
      90_000,
      "Claude API"
    );
  }

  if (response.stop_reason === "tool_use") {
    // Max rounds — force summary
    const assistantContent = response.content;
    history.push({ role: "assistant", content: assistantContent });

    const toolResults: Anthropic.ToolResultBlockParam[] = [];
    for (const block of assistantContent) {
      if (block.type === "tool_use") {
        toolResults.push({ type: "tool_result", tool_use_id: block.id, content: "Skipped — max rounds reached. Summarize what you have so far.", is_error: true });
      }
    }
    history.push({ role: "user", content: toolResults });

    const finalResponse = await withTimeout(
      anthropic.messages.create({
        model: "claude-sonnet-4-20250514",
        max_tokens: 1024,
        system: SYSTEM_PROMPT,
        messages: history,
      }),
      60_000,
      "Claude summary"
    );

    const text = finalResponse.content
      .filter((b): b is Anthropic.TextBlock => b.type === "text")
      .map((b) => b.text)
      .join("\n")
      .trim();
    history.push({ role: "assistant", content: text || "Done." });
    return text || "Done (max rounds reached).";
  }

  const text = response.content
    .filter((b): b is Anthropic.TextBlock => b.type === "text")
    .map((b) => b.text)
    .join("\n")
    .trim();

  if (!text) {
    // Claude gave no text — ask it to summarize
    history.push({ role: "assistant", content: response.content });
    history.push({ role: "user", content: "Now give me a brief summary of what you found. Text only, no tool calls." });
    const followUp = await withTimeout(
      anthropic.messages.create({
        model: "claude-sonnet-4-20250514",
        max_tokens: 1024,
        system: SYSTEM_PROMPT,
        messages: history,
      }),
      60_000,
      "Claude follow-up"
    );
    const followUpText = followUp.content
      .filter((b): b is Anthropic.TextBlock => b.type === "text")
      .map((b) => b.text)
      .join("\n")
      .trim();
    history.push({ role: "assistant", content: followUpText || "Done." });
    return followUpText || "Done.";
  }

  history.push({ role: "assistant", content: text });
  return text;
}

// ─── Security middleware ───
bot.use(async (ctx, next) => {
  if (ctx.chat?.type !== "private") return;
  if (!isAuthorized(ctx)) {
    console.log(`🚫 Blocked: ${ctx.from?.id} (@${ctx.from?.username})`);
    return;
  }
  if (ctx.from && isRateLimited(ctx.from.id)) {
    return ctx.reply("⏳ Slow down.");
  }
  await next();
});

// ─── Commands ───
bot.command("start", (ctx) =>
  ctx.reply(
    "🤖 CR Management Bot\n\n" +
    "• Chat naturally — I understand your infra\n" +
    "• /status — server health\n" +
    "• /ssh <cmd> — run command directly\n" +
    "• /cancel — stop current task\n" +
    "• /clear — reset conversation"
  )
);

bot.command("status", async (ctx) => {
  await ctx.reply("🔍 Checking...");
  try {
    const output = await withTimeout(
      sshExec("uptime && echo --- && df -h / | tail -1 && echo --- && free -h | grep Mem && echo --- && incus list -f csv -c ns 2>/dev/null", 10),
      15_000, "SSH"
    );
    return ctx.reply(output);
  } catch (err: any) {
    return ctx.reply(`❌ ${err.message}`);
  }
});

bot.command("ssh", async (ctx) => {
  const cmd = ctx.match;
  if (!cmd) return ctx.reply("Usage: /ssh <command>");
  const danger = isDangerous(cmd);
  if (danger) return ctx.reply(danger);
  try {
    const output = await withTimeout(sshExec(cmd, 10), 15_000, "SSH");
    return ctx.reply(output || "(no output)");
  } catch (err: any) {
    return ctx.reply(`❌ ${err.message}`);
  }
});

bot.command("clear", (ctx) => {
  conversations.delete(ctx.from!.id);
  activeRequests.get(ctx.from!.id)?.abort();
  return ctx.reply("🧹 Cleared.");
});

bot.command("cancel", (ctx) => {
  const controller = activeRequests.get(ctx.from!.id);
  if (controller) {
    controller.abort();
    activeRequests.delete(ctx.from!.id);
    return ctx.reply("🛑 Cancelled.");
  }
  return ctx.reply("Nothing running.");
});

// ─── Free text → Claude (fire-and-forget, never blocks bot) ───
bot.on("message:text", (ctx) => {
  const userId = ctx.from!.id;
  const msg = ctx.message!.text;

  // Cancel any in-flight request for this user
  const prev = activeRequests.get(userId);
  if (prev) {
    prev.abort();
    // Clear corrupted history to avoid tool_result mismatch
    conversations.delete(userId);
    console.log(`⏹️ [${userId}] cancelled previous + cleared history`);
  }

  const controller = new AbortController();
  activeRequests.set(userId, controller);

  // Fire and forget — bot handler returns immediately
  processMessage(ctx, userId, msg, controller);
});

async function processMessage(ctx: Context, userId: number, msg: string, controller: AbortController) {
  const startTime = Date.now();
  activeCtx = ctx;

  // Keep typing every 3s — Telegram typing indicator expires after ~5s
  const typingInterval = setInterval(async () => {
    if (controller.signal.aborted) return;
    ctx.replyWithChatAction("typing").catch(() => {});
  }, 3000);
  ctx.replyWithChatAction("typing").catch(() => {});

  try {
    const reply = await chat(userId, msg, controller.signal);
    if (controller.signal.aborted) return;

    const elapsed = ((Date.now() - startTime) / 1000).toFixed(1);
    console.log(`📤 [${userId}] ${reply.length} chars in ${elapsed}s`);

    const chunks = reply.match(/[\s\S]{1,4000}/g) || ["(empty)"];
    for (const chunk of chunks) {
      await ctx.reply(chunk).catch(() => ctx.reply("⚠️ Send failed."));
    }
  } catch (err: any) {
    if (!controller.signal.aborted) {
      console.error(`❌ [${userId}]`, err.message);
      await ctx.reply(`❌ ${err.message}`).catch(() => {});
    }
  } finally {
    clearInterval(typingInterval);
    activeCtx = null;
    if (activeRequests.get(userId) === controller) activeRequests.delete(userId);
  }
}

// ─── Launch with concurrent runner (never blocks on slow handlers) ───
console.log("🚀 Bot starting...");
const runner = run(bot);
console.log(`✅ Bot is live (concurrent mode)`);

// Graceful shutdown
const stop = () => { runner.isRunning() && runner.stop(); };
process.on("SIGINT", stop);
process.on("SIGTERM", stop);