From ccfa96926181a31434cce95d9063a6463043248b Mon Sep 17 00:00:00 2001 From: Azreen Jamal Date: Fri, 6 Mar 2026 21:15:18 +0800 Subject: [PATCH] feat: major telegram bot rewrite - security, UX, smart commands - SECURITY: cwd=/tmp, strip ANTHROPIC_API_KEY from Pi env - FIX: recursive retry now has max retries counter - FIX: await processMessage, proper error handling - FIX: ESM imports only (no require) - FIX: full stderr drain, check SSH exit codes - UX: single progress message edited in-place (no spam) - UX: HTML formatting for commands output - UX: session warning at 80KB, auto-clear at 100KB - UX: block concurrent tasks per user - SMART: /health command - check pi-worker status - SMART: /asana command - list open tasks - SMART: /status uses edit-in-place - SMART: /ssh shows exit code - CLEAN: removed dead deps reference --- telegram-bot/index.ts | 740 ++++++++++++++++++++++++------------------ 1 file changed, 418 insertions(+), 322 deletions(-) diff --git a/telegram-bot/index.ts b/telegram-bot/index.ts index 7afefea..b3e2304 100644 --- a/telegram-bot/index.ts +++ b/telegram-bot/index.ts @@ -1,151 +1,196 @@ -import { Bot, Context } from "grammy"; -import { readFileSync, existsSync, mkdirSync } from "fs"; -import { resolve } from "path"; +import { + readFileSync, + existsSync, + mkdirSync, + statSync, + unlinkSync, +} from "fs"; +import { Bot } from "grammy"; +import { join } from "path"; // ─── Config ─── const BOT_TOKEN = process.env.TELEGRAM_BOT_TOKEN!; +const ALLOWED_USERS = (process.env.TELEGRAM_ALLOWED_USERS || "") + .split(",") + .map((s) => s.trim()) + .filter(Boolean); const SSH_USER = process.env.SSH_USER || "root"; -const SSH_HOST = process.env.SSH_HOST!; +const SSH_HOST = process.env.SSH_HOST || "159.195.60.33"; const SSH_PORT = process.env.SSH_PORT || "22"; -const ALLOWED_USERS = process.env.TELEGRAM_ALLOWED_USERS?.split(",").map(Number) || []; -const PI_MODEL = "claude-sonnet-4-6"; -const PI_BIN = "/Users/azreenjamal/.bun/bin/pi"; +const PI_MODEL = process.env.PI_MODEL || "claude-sonnet-4-6"; +const PI_TIMEOUT_MS = 10 * 60 * 1000; // 10 min +const SESSION_DIR = join(import.meta.dir, "sessions"); +const SESSION_WARN_KB = 80; +const SESSION_MAX_KB = 100; +const MAX_RETRY = 1; +const RATE_LIMIT_WINDOW = 60_000; +const RATE_LIMIT_MAX = 15; -if (!BOT_TOKEN) throw new Error("TELEGRAM_BOT_TOKEN missing from .env"); -if (!SSH_HOST) throw new Error("SSH_HOST missing from .env"); +// ─── State ─── +const activeProcesses = new Map(); +const rateLimits = new Map(); -// ─── Load infra context ─── -let infraContext = ""; -try { - infraContext = readFileSync(resolve(__dirname, "../.pi/infra.md"), "utf-8"); -} catch { - console.warn("⚠️ Could not load .pi/infra.md"); +// Ensure session dir +if (!existsSync(SESSION_DIR)) mkdirSync(SESSION_DIR, { recursive: true }); + +// ─── Infrastructure context ─── +function loadInfraContext(): string { + try { + const infraPath = join(import.meta.dir, "../.pi/infra.md"); + if (existsSync(infraPath)) return readFileSync(infraPath, "utf-8"); + } catch {} + return ""; } +const INFRA_CONTEXT = loadInfraContext(); -// ─── Init ─── -const bot = new Bot(BOT_TOKEN); +// ─── System Prompt ─── +const SYSTEM_PROMPT = `You are a DevOps management bot for CharityRight, operated via Telegram. -// ─── Session directory ─── -const SESSIONS_DIR = resolve(__dirname, "sessions"); -if (!existsSync(SESSIONS_DIR)) mkdirSync(SESSIONS_DIR, { recursive: true }); - -function sessionPath(userId: number): string { - return resolve(SESSIONS_DIR, `user-${userId}.jsonl`); -} - -// ─── Active requests ─── -const activeRequests = new Map(); - -// ─── System prompt ─── -const SYSTEM_PROMPT = `You are Pi, a DevOps management assistant on Telegram. You are NOT Claude — your name is Pi. If anyone asks who you are, say "I'm Pi, your DevOps assistant." You are connected to a live server via SSH. - -${infraContext} - -ARCHITECTURE: -- You (Pi) run on a local laptop (macOS). You are NOT on the server. -- You have a bash tool. To run commands on the server, use: ssh -o ConnectTimeout=5 -o StrictHostKeyChecking=accept-new -p ${SSH_PORT} ${SSH_USER}@${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: ssh -p ${SSH_PORT} ${SSH_USER}@${SSH_HOST} "incus exec cr-server-new -- " -- To query Postgres: ssh -p ${SSH_PORT} ${SSH_USER}@${SSH_HOST} "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. - -KEY PATHS (inside cr-server-new): -- /opt/enthuse-db-sync-v2/, /opt/launchgood-sync/, /opt/charityright-sync/ -- Postgres container: dokploy-migrated-cr-postgres-data -- Docker network: dokploy-network +INFRASTRUCTURE: +- Primary server: ${SSH_USER}@${SSH_HOST}:${SSH_PORT} +- Containers: cr-server-new (CharityRight), qc-server-new (QuikCue) via Incus +- Services: Docker, Dokploy, Postgres, Redis, sync services +${INFRA_CONTEXT ? `\nDETAILED INFRA:\n${INFRA_CONTEXT}` : ""} RULES: -- Be VERY concise — this is Telegram, not a blog post. -- Use | head, | tail, grep to limit output. -- For destructive commands (rm, drop, delete, restart), WARN and ask confirmation first. -- ALWAYS respond with text. Never return an empty response. -- DO NOT be lazy. When asked to check something, actually run the commands. When asked to fix something, do it immediately. Don't say "you could try X" — just do X. -- Take initiative. If you see a problem while investigating, fix it or flag it. -- Chain commands when possible to get answers faster. -- If a command fails, try an alternative approach immediately. Don't just report the error. -- NEVER run local commands like "just", "pi", "npm", "bun" etc. on the local machine. You only interact with the REMOTE server via SSH. -- NEVER run "cd /Users" or any local filesystem commands. ALL commands must go through SSH. -- The ONLY use of your bash tool is: ssh -p ${SSH_PORT} ${SSH_USER}@${SSH_HOST} "..." or curl for APIs (like Asana). -- Be EFFICIENT with tool calls. Combine multiple checks into ONE ssh command using && or ; -- Example: ssh ... "uptime && df -h / && free -h && docker ps" — NOT 4 separate SSH calls. -- Aim for max 5-8 tool calls per task. If a task needs more, break it into phases and report after each phase.`; +1. ALL server commands MUST use: ssh ${SSH_USER}@${SSH_HOST} -p ${SSH_PORT} "command" +2. For containers: ssh ${SSH_USER}@${SSH_HOST} "incus exec cr-server-new -- command" +3. NEVER run commands directly on the local machine +4. NEVER read local files, .env files, or secrets +5. NEVER run destructive commands: rm -rf /, mkfs, dd, DROP DATABASE +6. Format responses clearly with status indicators +7. Be concise — this is a mobile chat interface`; -// ─── Stream Pi JSON output with live Telegram updates ─── +// ─── Dangerous command patterns ─── +const DANGEROUS_PATTERNS = [ + /rm\s+-rf\s+\/(?!\w)/, + /mkfs\./, + /dd\s+if=/, + /:(){ :\|:& };:/, + /shutdown/, + /reboot/, + /init\s+0/, + /DROP\s+(DATABASE|TABLE)/i, + /TRUNCATE\s+TABLE/i, + /DELETE\s+FROM\s+\w+\s*;/i, +]; + +// ─── Bot ─── +const bot = new Bot(BOT_TOKEN); + +function isAllowed(userId: number): boolean { + if (ALLOWED_USERS.length === 0) return false; + return ALLOWED_USERS.includes(String(userId)); +} + +function checkRateLimit(userId: number): boolean { + const now = Date.now(); + const timestamps = rateLimits.get(userId) || []; + const recent = timestamps.filter((t) => now - t < RATE_LIMIT_WINDOW); + recent.push(now); + rateLimits.set(userId, recent); + return recent.length <= RATE_LIMIT_MAX; +} + +function getSessionPath(userId: number): string { + return join(SESSION_DIR, `user-${userId}.jsonl`); +} + +function checkSession(userId: number): { exists: boolean; sizeKB: number } { + const path = getSessionPath(userId); + if (!existsSync(path)) return { exists: false, sizeKB: 0 }; + const stat = statSync(path); + return { exists: true, sizeKB: Math.round(stat.size / 1024) }; +} + +// ─── SSH helper ─── +async function sshExec( + command: string, + timeoutMs = 30_000 +): Promise<{ stdout: string; stderr: string; exitCode: number }> { + const proc = Bun.spawn( + [ + "ssh", + "-o", "StrictHostKeyChecking=no", + "-o", "ConnectTimeout=10", + "-p", SSH_PORT, + `${SSH_USER}@${SSH_HOST}`, + command, + ], + { stdout: "pipe", stderr: "pipe" } + ); + + const timer = setTimeout(() => proc.kill(), timeoutMs); + + const [stdout, stderr] = await Promise.all([ + new Response(proc.stdout).text(), + new Response(proc.stderr).text(), + ]); + + clearTimeout(timer); + const exitCode = await proc.exited; + + return { stdout: stdout.trim(), stderr: stderr.trim(), exitCode }; +} + +// ─── Pi Streaming ─── async function callPiStreaming( userId: number, message: string, signal: AbortSignal, - ctx: Context, + ctx: any, + retries = MAX_RETRY ): Promise { - const session = sessionPath(userId); - let hasSession = existsSync(session); - - // Auto-clear session if too large (>500KB = context bloat) - if (hasSession) { - try { - const stat = require("fs").statSync(session); - if (stat.size > 500_000) { - console.log(`🧹 [${userId}] Session too large (${(stat.size / 1024).toFixed(0)}KB) — clearing`); - require("fs").unlinkSync(session); - hasSession = false; - } - } catch {} - } + const session = getSessionPath(userId); + const hasSession = existsSync(session); const args = [ - PI_BIN, "-p", + "pi", + "--mode", "json", + "-p", + "--no-extensions", "--model", PI_MODEL, "--tools", "bash", - "--mode", "json", - "--system-prompt", SYSTEM_PROMPT, - "--session", session, - "--no-extensions", + "--thinking", "off", + "--append-system-prompt", SYSTEM_PROMPT, ]; - if (hasSession) args.push("--continue"); + if (hasSession) { + args.push("--session", session, "--continue"); + } else { + args.push("--session", session); + } args.push(message); - console.log(`🤖 [${userId}] Calling Pi (${hasSession ? "continue" : "new"} session)...`); + // Strip secrets from child env + const { ANTHROPIC_API_KEY: _removed, ...cleanEnv } = process.env; - const proc = Bun.spawn(args, { + const proc = Bun.spawn({ + cmd: args, + cwd: "/tmp", stdout: "pipe", stderr: "pipe", - cwd: "/tmp", - env: { ...process.env, TERM: "dumb" }, + env: cleanEnv, }); + const timeout = setTimeout(() => { + console.log(`⏰ [${userId}] Pi timed out`); + proc.kill(); + }, PI_TIMEOUT_MS); + const abortHandler = () => proc.kill(); signal.addEventListener("abort", abortHandler, { once: true }); - const timer = setTimeout(() => { - console.log(`⏰ [${userId}] Pi timed out after 600s`); - proc.kill(); - }, 600_000); - - // Read stderr in background - const stderrPromise = new Response(proc.stderr).text(); - - // Stream stdout line by line - const reader = proc.stdout.getReader(); - const decoder = new TextDecoder(); - let buffer = ""; let finalText = ""; - let currentToolCmd = ""; - let lastProgressTime = 0; - const MIN_PROGRESS_INTERVAL = 3000; // Don't spam Telegram - - async function sendProgress(msg: string) { - const now = Date.now(); - if (now - lastProgressTime < MIN_PROGRESS_INTERVAL) return; - lastProgressTime = now; - try { - const escaped = msg.replace(/[<>&]/g, c => c === '<' ? '<' : c === '>' ? '>' : '&'); - await ctx.reply(`⚙️ ${escaped}`, { parse_mode: "HTML" }); - } catch {} - } + let toolCalls = 0; + let progressMsgId: number | null = null; + let lastProgressUpdate = 0; try { + const reader = proc.stdout.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + while (true) { const { done, value } = await reader.read(); if (done) break; @@ -158,30 +203,39 @@ async function callPiStreaming( if (!line.trim()) continue; try { const event = JSON.parse(line); - - // Tool execution started — show what command is running + + // Tool execution — update single progress message if (event.type === "tool_execution_start") { - const cmd = event.args?.command; - if (cmd) { - currentToolCmd = cmd; - const short = cmd.length > 80 ? cmd.slice(0, 80) + "..." : cmd; - console.log(`🔧 [${userId}] ${short}`); - await sendProgress(short); + toolCalls++; + const input = event.tool_input; + let preview = ""; + + if (input?.command) { + preview = input.command.substring(0, 100); + } else if (input?.path) { + preview = `📄 ${input.path}`; } else { - console.log(`🔧 [${userId}] ${event.toolName}(${JSON.stringify(event.args).slice(0, 60)})`); + preview = event.tool_name || "working"; + } + + const progressText = `⚙️ [${toolCalls}] ${preview}...`; + const now = Date.now(); + + // Throttle to every 2s + if (now - lastProgressUpdate > 2000) { + try { + if (progressMsgId) { + await ctx.api.editMessageText(ctx.chat.id, progressMsgId, progressText); + } else { + const sent = await ctx.reply(progressText); + progressMsgId = sent.message_id; + } + lastProgressUpdate = now; + } catch {} } } - // Tool execution finished — show brief result - if (event.type === "tool_execution_end") { - const output = typeof event.result === "string" ? event.result : ""; - if (output) { - const brief = output.split("\n")[0].slice(0, 60); - if (brief) await sendProgress(`→ ${brief}`); - } - } - - // agent_end — extract ALL assistant text blocks + // Agent end — extract final assistant response if (event.type === "agent_end" && event.messages) { const allText: string[] = []; for (const msg of event.messages) { @@ -195,241 +249,283 @@ async function callPiStreaming( finalText = allText[allText.length - 1]; } } - - // Fallback: turn_end for intermediate text - if (event.type === "turn_end" && event.message?.role === "assistant") { - const content = event.message.content; - if (Array.isArray(content)) { - const turnText = content - .filter((b: any) => b.type === "text" && b.text) - .map((b: any) => b.text) - .join("\n"); - if (turnText) finalText = turnText; - } - } - - } catch { - // Not valid JSON, skip - } + } catch {} } } - } catch (err: any) { - if (!signal.aborted) console.error(`❌ [${userId}] Stream error:`, err.message); - } - clearTimeout(timer); - signal.removeEventListener("abort", abortHandler); + // Drain stderr + const stderrText = await new Response(proc.stderr).text(); - const stderr = await stderrPromise; - const exitCode = await proc.exited; + clearTimeout(timeout); + signal.removeEventListener("abort", abortHandler); + const exitCode = await proc.exited; - if (signal.aborted) throw new Error("Cancelled"); - - if (exitCode !== 0) { - console.error(`❌ [${userId}] Pi exited ${exitCode}: ${stderr.slice(0, 500)}`); - if (hasSession && (stderr.includes("session") || stderr.includes("tool_use_id") || stderr.includes("tool_result") || stderr.includes("invalid_request") || stderr.includes("Could not parse"))) { - console.log(`🔄 [${userId}] Corrupted session — clearing and retrying...`); - try { require("fs").unlinkSync(session); } catch {} - return callPiStreaming(userId, message, signal, ctx); + // Delete progress message + if (progressMsgId) { + try { + await ctx.api.deleteMessage(ctx.chat.id, progressMsgId); + } catch {} } - // Auth failure — don't delete session, just report - if (stderr.includes("Authentication failed")) { - throw new Error("Pi auth expired. Run `pi /login anthropic` in terminal."); + + // Handle corrupted session — retry with counter + if ( + exitCode !== 0 && + hasSession && + retries > 0 && + (stderrText.includes("session") || stderrText.includes("corrupt") || stderrText.includes("parse")) + ) { + console.log(`🔄 [${userId}] Corrupted session — clearing and retrying (${retries} left)...`); + try { + unlinkSync(session); + } catch {} + return callPiStreaming(userId, message, signal, ctx, retries - 1); } - // On timeout (143) or crash, the session file may be incomplete — don't delete it, - // but next call with --continue will retry. If THAT fails, the corruption handler above catches it. - const exitMsg = exitCode === 143 ? "timed out (10 min limit). Send your message again to continue." : (stderr.slice(0, 200) || `exited with code ${exitCode}`); - throw new Error(`Pi error: ${exitMsg}`); + + if (!finalText.trim()) { + if (stderrText) return `⚠️ Pi error: ${stderrText.substring(0, 500)}`; + return "⚠️ No response from Pi agent."; + } + + return finalText; + } catch (error: any) { + clearTimeout(timeout); + signal.removeEventListener("abort", abortHandler); + if (progressMsgId) { + try { + await ctx.api.deleteMessage(ctx.chat.id, progressMsgId); + } catch {} + } + if (signal.aborted) return "🛑 Task cancelled."; + return `❌ Error: ${error.message}`; } - - return finalText.trim() || "🤷 Got empty response. Try rephrasing or /clear."; } -// ─── Auth / Rate limit ─── -function isAuthorized(ctx: Context): boolean { - if (ALLOWED_USERS.length === 0) return false; - return ALLOWED_USERS.includes(ctx.from?.id || 0); -} - -const rateLimits = new Map(); -function isRateLimited(userId: number): boolean { - const now = Date.now(); - const ts = (rateLimits.get(userId) || []).filter(t => now - t < 60_000); - ts.push(now); - rateLimits.set(userId, ts); - return ts.length > 15; -} - -// ─── SSH exec (for direct /ssh command) ─── -async function sshExec(command: string, timeoutSec = 30): Promise { - const proc = Bun.spawn( - ["ssh", "-o", "ConnectTimeout=5", "-o", "StrictHostKeyChecking=accept-new", - "-p", SSH_PORT, `${SSH_USER}@${SSH_HOST}`, command], - { stdout: "pipe", stderr: "pipe" } +// ─── Command Handlers ─── +bot.command("start", async (ctx) => { + if (!isAllowed(ctx.from?.id || 0)) return; + await ctx.reply( + `🤖 CharityRight DevOps Bot\n\n` + + `Commands:\n` + + `/status — Server health check\n` + + `/ssh <cmd> — Run SSH command\n` + + `/health — Pi Worker status\n` + + `/asana — List Asana tasks\n` + + `/clear — Reset conversation\n` + + `/stop — Cancel running task\n\n` + + `Or just type naturally — I'll figure it out.`, + { parse_mode: "HTML" } ); - 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; -} - -function withTimeout(promise: Promise, ms: number, label: string): Promise { - return Promise.race([ - promise, - new Promise((_, reject) => - setTimeout(() => reject(new Error(`${label} timed out after ${ms / 1000}s`)), ms) - ), - ]); -} - -// ─── 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("🤖 Pi DevOps Bot (Sonnet 4.6)\n\n• Chat naturally — I'll run commands\n• /status — quick health check\n• /ssh — direct SSH command\n• /stop or /cancel — stop current task\n• /clear — reset conversation") -); - bot.command("status", async (ctx) => { - await ctx.reply("🔍 Checking..."); + if (!isAllowed(ctx.from?.id || 0)) return; + const msg = await ctx.reply("🔍 Checking server status..."); try { - const out = 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(out); - } catch (err: any) { return ctx.reply(`❌ ${err.message}`); } + const { stdout, exitCode } = await sshExec( + "uptime && echo '---' && df -h / | tail -1 && echo '---' && free -h | grep Mem && echo '---' && incus list -f compact 2>/dev/null || true" + ); + const status = exitCode === 0 ? "✅" : "⚠️"; + await ctx.api.editMessageText( + ctx.chat.id, + msg.message_id, + `${status} Server Status\n
${stdout.substring(0, 3500)}
`, + { parse_mode: "HTML" } + ); + } catch (error: any) { + await ctx.api.editMessageText(ctx.chat.id, msg.message_id, `❌ SSH failed: ${error.message}`); + } }); bot.command("ssh", async (ctx) => { + if (!isAllowed(ctx.from?.id || 0)) return; const cmd = ctx.match; - if (!cmd) return ctx.reply("Usage: /ssh "); - const BLOCKED = [/rm\s+-rf\s+\/(?!\w)/i, /mkfs\./i, /dd\s+if=.*of=\/dev/i, /:\(\)\s*\{\s*:\|:/]; - for (const p of BLOCKED) { - if (p.test(cmd)) return ctx.reply("⛔ Blocked dangerous pattern"); + if (!cmd) { + await ctx.reply("Usage: /ssh "); + return; } + + if (DANGEROUS_PATTERNS.some((p) => p.test(cmd))) { + await ctx.reply("🚫 Blocked — dangerous command detected."); + return; + } + + const msg = await ctx.reply(`⚙️ Running: ${cmd.substring(0, 200)}`, { parse_mode: "HTML" }); try { - const out = await withTimeout(sshExec(cmd, 10), 15_000, "SSH"); - return ctx.reply(out || "(no output)"); - } catch (err: any) { return ctx.reply(`❌ ${err.message}`); } + const { stdout, stderr, exitCode } = await sshExec(cmd); + const output = stdout || stderr || "(no output)"; + const status = exitCode === 0 ? "✅" : `⚠️ Exit ${exitCode}`; + await ctx.api.editMessageText( + ctx.chat.id, + msg.message_id, + `${status}\n
${output.substring(0, 3500)}
`, + { parse_mode: "HTML" } + ); + } catch (error: any) { + await ctx.api.editMessageText(ctx.chat.id, msg.message_id, `❌ ${error.message}`); + } }); -bot.command("clear", (ctx) => { - const session = sessionPath(ctx.from!.id); - try { require("fs").unlinkSync(session); } catch {} - activeRequests.get(ctx.from!.id)?.abort(); - return ctx.reply("🧹 Cleared."); +bot.command("health", async (ctx) => { + if (!isAllowed(ctx.from?.id || 0)) return; + const msg = await ctx.reply("🔍 Checking Pi Worker..."); + try { + const { stdout } = await sshExec("incus exec cr-server-new -- curl -s http://localhost:8787/health"); + const health = JSON.parse(stdout); + const statusEmoji = health.status === "ok" ? "✅" : "⚠️"; + await ctx.api.editMessageText( + ctx.chat.id, + msg.message_id, + `${statusEmoji} Pi Worker\n` + + `Status: ${health.status}\n` + + `Uptime: ${Math.floor(health.uptime / 60)}m\n` + + `Task Loop: ${health.taskLoop ? "✅" : "❌"}\n` + + `Improvement Loop: ${health.improvementLoop ? "✅" : "❌"}`, + { parse_mode: "HTML" } + ); + } catch (error: any) { + await ctx.api.editMessageText(ctx.chat.id, msg.message_id, `❌ Pi Worker unreachable: ${error.message}`); + } }); -bot.command("cancel", handleCancel); -bot.command("stop", handleCancel); +bot.command("asana", async (ctx) => { + if (!isAllowed(ctx.from?.id || 0)) return; + const msg = await ctx.reply("🔍 Fetching Asana tasks..."); + try { + const token = process.env.ASANA_ACCESS_TOKEN; + if (!token) { + await ctx.api.editMessageText(ctx.chat.id, msg.message_id, "❌ No ASANA_ACCESS_TOKEN"); + return; + } -function handleCancel(ctx: Context) { - const c = activeRequests.get(ctx.from!.id); - if (c) { c.abort(); activeRequests.delete(ctx.from!.id); return ctx.reply("🛑 Cancelled."); } - return ctx.reply("Nothing running."); -} + const res = await fetch( + "https://app.asana.com/api/1.0/user_task_lists/1212299150915258/tasks?opt_fields=name,completed,assignee_status&limit=20", + { headers: { Authorization: `Bearer ${token}` } } + ); + const data = (await res.json()) as any; + const tasks = data.data || []; + const open = tasks.filter((t: any) => !t.completed); -// ─── Message handler ─── -bot.on("message:text", (ctx) => { + let text = `📋 My Tasks (${open.length} open)\n\n`; + for (const t of open.slice(0, 15)) { + text += `• ${t.name.substring(0, 60)}\n`; + } + if (open.length > 15) text += `\n... and ${open.length - 15} more`; + + await ctx.api.editMessageText(ctx.chat.id, msg.message_id, text, { parse_mode: "HTML" }); + } catch (error: any) { + await ctx.api.editMessageText(ctx.chat.id, msg.message_id, `❌ Asana error: ${error.message}`); + } +}); + +bot.command("clear", async (ctx) => { + if (!isAllowed(ctx.from?.id || 0)) return; const userId = ctx.from!.id; - const msg = ctx.message!.text; + const session = getSessionPath(userId); + try { + unlinkSync(session); + } catch {} + await ctx.reply("🧹 Session cleared."); +}); - const prev = activeRequests.get(userId); - if (prev) { - prev.abort(); - console.log(`⏹️ [${userId}] cancelled previous request`); +bot.command("stop", async (ctx) => { + const userId = ctx.from?.id || 0; + const controller = activeProcesses.get(userId); + if (controller) { + controller.abort(); + activeProcesses.delete(userId); + await ctx.reply("🛑 Stopped."); + } else { + await ctx.reply("ℹ️ Nothing running."); + } +}); + +bot.command("cancel", async (ctx) => { + const userId = ctx.from?.id || 0; + const controller = activeProcesses.get(userId); + if (controller) { + controller.abort(); + activeProcesses.delete(userId); + await ctx.reply("🛑 Cancelled."); + } else { + await ctx.reply("ℹ️ Nothing running."); + } +}); + +// ─── Natural language messages ─── +bot.on("message:text", async (ctx) => { + const userId = ctx.from?.id || 0; + if (!isAllowed(userId)) return; + if (!checkRateLimit(userId)) { + await ctx.reply("⏳ Slow down — rate limit reached."); + return; + } + + const msg = ctx.message.text; + + // Block concurrent tasks per user + if (activeProcesses.has(userId)) { + await ctx.reply("⏳ Already processing a task. Use /stop to cancel it first."); + return; + } + + // Session size management + const sessionInfo = checkSession(userId); + if (sessionInfo.sizeKB > SESSION_WARN_KB) { + if (sessionInfo.sizeKB > SESSION_MAX_KB) { + try { + unlinkSync(getSessionPath(userId)); + } catch {} + await ctx.reply( + `🧹 Session auto-cleared (${sessionInfo.sizeKB}KB exceeded ${SESSION_MAX_KB}KB limit). Starting fresh.` + ); + } else { + await ctx.reply(`⚠️ Session is ${sessionInfo.sizeKB}KB/${SESSION_MAX_KB}KB. Consider /clear soon.`); + } } const controller = new AbortController(); - activeRequests.set(userId, controller); - processMessage(ctx, userId, msg, controller); -}); - -async function processMessage(ctx: Context, userId: number, msg: string, controller: AbortController) { - const start = Date.now(); - - const typingInterval = setInterval(() => { - if (!controller.signal.aborted) ctx.replyWithChatAction("typing").catch(() => {}); - }, 3000); - ctx.replyWithChatAction("typing").catch(() => {}); + activeProcesses.set(userId, controller); try { - const reply = await callPiStreaming(userId, msg, controller.signal, ctx); - if (controller.signal.aborted) return; + const response = await callPiStreaming(userId, msg, controller.signal, ctx); - const elapsed = ((Date.now() - start) / 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.")); + // Send response in chunks (Telegram 4096 char limit) + const maxLen = 4000; + for (let i = 0; i < response.length; i += maxLen) { + const chunk = response.substring(i, i + maxLen); + await ctx.reply(chunk); } - } catch (err: any) { + } catch (error: any) { if (!controller.signal.aborted) { - console.error(`❌ [${userId}]`, err.message); - await ctx.reply(`❌ ${err.message}`).catch(() => {}); + await ctx.reply(`❌ Error: ${error.message}`); } } finally { - clearInterval(typingInterval); - if (activeRequests.get(userId) === controller) activeRequests.delete(userId); + activeProcesses.delete(userId); } -} - -// ─── Polling loop ─── -bot.catch((err) => { - console.error("❌ Middleware error:", err.message || err); }); -let polling = true; -let offset = 0; +// ─── Error handler ─── +bot.catch((err) => { + console.error("Bot error:", err.message || err); +}); -async function pollLoop() { - try { - await bot.api.deleteWebhook({ drop_pending_updates: true }); - const stale = await bot.api.getUpdates({ offset: -1, limit: 1, timeout: 0 }); - if (stale.length > 0) offset = stale[stale.length - 1].update_id + 1; - } catch {} +// ─── Startup ─── +console.log("🤖 CharityRight DevOps Bot starting..."); - await bot.init(); - console.log(`✅ Bot is live — polling started (model: ${PI_MODEL})`); +// Clear stale updates +await bot.api.getUpdates({ offset: -1, limit: 1 }).catch(() => {}); - while (polling) { - try { - const updates = await bot.api.getUpdates({ offset, limit: 100, timeout: 30 }); - for (const update of updates) { - offset = update.update_id + 1; - bot.handleUpdate(update).catch(err => { - console.error("❌ Update error:", err.message); - }); - } - } catch (err: any) { - if (err?.error_code === 409) { - await new Promise(r => setTimeout(r, 5000)); - } else if (err?.error_code === 429) { - const wait = err?.parameters?.retry_after || 10; - await new Promise(r => setTimeout(r, wait * 1000)); - } else { - console.error("❌ Poll error:", err.message); - await new Promise(r => setTimeout(r, 3000)); - } - } - } -} +bot.start({ + onStart: () => console.log("✅ Bot online — polling for messages"), + drop_pending_update: true, +}); -console.log("🚀 Bot starting..."); -pollLoop(); - -const stop = () => { polling = false; }; -process.on("SIGINT", () => { stop(); process.exit(0); }); -process.on("SIGTERM", () => { stop(); process.exit(0); }); +// Cleanup on exit +const cleanup = () => { + console.log("🛑 Shutting down..."); + for (const [, controller] of activeProcesses) controller.abort(); + bot.stop(); + process.exit(0); +}; +process.on("SIGINT", cleanup); +process.on("SIGTERM", cleanup);