feat(telegram-remote): ship lib/ code, add voice transcription, chat pagination, stream photos
- Add lib/ (plain-ESM plugin code, no build step) so the published tarball actually contains the plugin; un-ignore lib/ for this package - Sync README: voice/audio transcription via home-lab Whishper (default large-v2, device cuda, language auto), /chats pagination and subagent hiding, whisper* config keys, updated security note
This commit is contained in:
327
packages/telegram-remote/lib/state.js
Normal file
327
packages/telegram-remote/lib/state.js
Normal file
@@ -0,0 +1,327 @@
|
||||
/**
|
||||
* 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);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user