/** * Telegram bridge config and pairing helpers * Zones: telegram config, pairing, filesystem * Owns persisted bot/session pairing state, local config storage, live config controls, authorization policy, and first-user pairing side effects */ import { randomUUID } from "node:crypto"; import { chmodSync, existsSync, mkdirSync, readFileSync, renameSync, statSync, writeFileSync, } from "node:fs"; import { chmod, mkdir, rename, writeFile } from "node:fs/promises"; import { resolveAgentDir, resolveTelegramConfigPath } from "./paths.ts"; import type { TelegramInboundHandlerConfig } from "./inbound.ts"; import type { CommandTemplateObjectConfig } from "./command-templates.ts"; import { withTelegramFileTransaction } from "./locks.ts"; const CONFIG_RUNTIME_KEY = "__piTelegramConfigRuntime__"; function getConfigPath(): string { return resolveTelegramConfigPath(); } export type TelegramOutboundCommandTemplateConfig = string | CommandTemplateObjectConfig; export interface TelegramOutboundHandlerConfig extends CommandTemplateObjectConfig { type?: string; match?: string | string[]; output?: string; timeout?: number | string; } export type TelegramTimeMode = "hidden" | "always" | "interval"; export interface TelegramTimeConfig { injectionMode?: TelegramTimeMode; interval?: number; } export interface ResolvedTelegramTimeConfig { injectionMode: TelegramTimeMode; interval: number; timezone: string; } export type TelegramAssistantRenderingMode = "rich" | "html"; export interface TelegramConfig { botToken?: string; botUsername?: string; botId?: number; allowedUserId?: number; lastUpdateId?: number; inboundHandlers?: TelegramInboundHandlerConfig[]; attachmentHandlers?: TelegramInboundHandlerConfig[]; outboundHandlers?: TelegramOutboundHandlerConfig[]; proactivePush?: boolean; assistant?: { draftPreviews?: boolean; rendering?: TelegramAssistantRenderingMode; }; /** @deprecated use assistant.draftPreviews */ draftPreviews?: boolean; /** @deprecated use assistant.draftPreviews */ richDraftPreviews?: boolean; /** @deprecated use assistant.rendering */ assistantRendering?: TelegramAssistantRenderingMode; voice?: { replyMode?: "manual" | "mirror" | "always"; /** Whether to attach the provider's transcriptText as caption on voice messages */ sendTranscript?: boolean; }; time?: TelegramTimeConfig; /** Named bot/session profiles (e.g. "work", "omp"). */ profiles?: Record; } /** * Per-profile bot/session identity fields. * Stored under `profiles.` in telegram.json. * Shared bridge settings (inboundHandlers, outboundHandlers, voice, time, * assistant, proactivePush) stay at the top level. */ export interface TelegramBotProfile { botToken: string; botUsername?: string; botId?: number; allowedUserId?: number; lastUpdateId?: number; } /** Profile names must contain only lowercase ASCII letters and digits; max 32 chars. */ const TELEGRAM_PROFILE_NAME_PATTERN = /^[a-z0-9]{1,32}$/; const TELEGRAM_RESERVED_PROFILE_NAMES: ReadonlySet = new Set([ "default", "main", "active", ]); export function isValidTelegramProfileName(name: string): boolean { return ( TELEGRAM_PROFILE_NAME_PATTERN.test(name) && !TELEGRAM_RESERVED_PROFILE_NAMES.has(name) ); } /** * Resolve the effective config for a named (or default) profile. * Returns bot/session fields from the named profile, falling back to * top-level fields for the default profile. Shared bridge settings * always come from the top level. */ export function resolveTelegramActiveProfile( config: TelegramConfig, profileName?: string, ): { botToken?: string; botUsername?: string; botId?: number; allowedUserId?: number; lastUpdateId?: number; } { if (!profileName || !config.profiles?.[profileName]) { return { botToken: config.botToken, botUsername: config.botUsername, botId: config.botId, allowedUserId: config.allowedUserId, lastUpdateId: config.lastUpdateId, }; } const profile = config.profiles[profileName]; return { botToken: profile.botToken, botUsername: profile.botUsername, botId: profile.botId, allowedUserId: profile.allowedUserId, lastUpdateId: profile.lastUpdateId, }; } /** List defined profile names. */ export function getTelegramProfileNames(config: TelegramConfig): string[] { return Object.keys(config.profiles ?? {}).sort(); } export interface TelegramConfigStore { get: () => TelegramConfig; getStoredConfig: () => TelegramConfig; set: (config: TelegramConfig) => void; update: (mutate: (config: TelegramConfig) => void) => void; activateProfile: (profileName: string | undefined) => boolean; getActiveProfileName: () => string | undefined; getBotToken: () => string | undefined; hasBotToken: () => boolean; getAllowedUserId: () => number | undefined; getInboundHandlers: () => TelegramInboundHandlerConfig[] | undefined; getAttachmentHandlers: () => TelegramInboundHandlerConfig[] | undefined; getOutboundHandlers: () => TelegramOutboundHandlerConfig[] | undefined; setAllowedUserId: (userId: number) => void; load: () => Promise; persist: (config?: TelegramConfig) => Promise; } export interface TelegramConfigStoreOptions { initialConfig?: TelegramConfig; agentDir?: string; configPath?: string; recordRuntimeEvent?: ( category: string, error: unknown, details?: Record, ) => void; } export interface TelegramInvalidConfigRecovery { configPath: string; recoveryPath: string; error: unknown; } export interface TelegramConfigRuntime { updateVoiceConfig: (voice: NonNullable) => void; } export function setGlobalTelegramConfigRuntime( runtime: TelegramConfigRuntime | undefined, ): void { const globals = globalThis as Record; if (runtime) globals[CONFIG_RUNTIME_KEY] = runtime; else delete globals[CONFIG_RUNTIME_KEY]; } export function updateTelegramVoiceConfig( voice: NonNullable, ): boolean { const runtime = (globalThis as Record)[ CONFIG_RUNTIME_KEY ] as TelegramConfigRuntime | undefined; if (!runtime || typeof runtime.updateVoiceConfig !== "function") return false; runtime.updateVoiceConfig(voice); return true; } type TelegramMutableConfigStore = Pick< TelegramConfigStore, "get" | "set" | "persist" > & { load?: () => Promise; }; function isEmptyTelegramConfig(config: TelegramConfig): boolean { return Object.keys(config).length === 0; } async function loadLatestTelegramConfig( configStore: TelegramMutableConfigStore, ): Promise { if (!configStore.load) return; const before = configStore.get(); await configStore.load(); if ( !isEmptyTelegramConfig(before) && isEmptyTelegramConfig(configStore.get()) ) { configStore.set(before); } } export function bindGlobalTelegramConfigRuntime( configStore: TelegramMutableConfigStore, ): void { setGlobalTelegramConfigRuntime({ updateVoiceConfig(voice) { const current = configStore.get(); const next = { ...current, voice: { ...(current.voice ?? {}), ...voice }, }; configStore.set(next); void configStore.persist(next); }, }); } function getInvalidTelegramConfigRecoveryPath(configPath: string): string { return `${configPath}.invalid-${process.pid}-${Date.now()}`; } export async function readTelegramConfig( configPath: string, options: { onInvalidConfig?: (recovery: TelegramInvalidConfigRecovery) => void; } = {}, ): Promise { return withTelegramFileTransaction(`${configPath}.transaction`, () => { if (!existsSync(configPath)) return {}; const identity = statSync(configPath); const content = readFileSync(configPath, "utf8"); try { return JSON.parse(content) as TelegramConfig; } catch (error) { const currentIdentity = statSync(configPath); if ( currentIdentity.dev !== identity.dev || currentIdentity.ino !== identity.ino || currentIdentity.size !== identity.size || currentIdentity.mtimeMs !== identity.mtimeMs ) { throw new Error( `Telegram config changed while validating invalid content: ${configPath}`, { cause: error }, ); } const recoveryPath = getInvalidTelegramConfigRecoveryPath(configPath); renameSync(configPath, recoveryPath); options.onInvalidConfig?.({ configPath, recoveryPath, error }); return {}; } }); } export async function writeTelegramConfig( agentDir: string, configPath: string, config: TelegramConfig, ): Promise { await mkdir(agentDir, { recursive: true }); const tempConfigPath = `${configPath}.tmp-${process.pid}-${Date.now()}`; await writeFile(tempConfigPath, JSON.stringify(config, null, "\t") + "\n", { encoding: "utf8", mode: 0o600, }); await chmod(tempConfigPath, 0o600); await rename(tempConfigPath, configPath); await chmod(configPath, 0o600); } function isPlainConfigRecord(value: unknown): value is Record { return !!value && typeof value === "object" && !Array.isArray(value); } function cloneTelegramConfig(value: T): T { return structuredClone(value); } function configValuesEqual(left: unknown, right: unknown): boolean { return JSON.stringify(left) === JSON.stringify(right); } function mergeTelegramConfigDelta( base: Record, desired: Record, latest: Record, ): Record { const merged = cloneTelegramConfig(latest); for (const key of new Set([...Object.keys(base), ...Object.keys(desired)])) { const baseHas = Object.hasOwn(base, key); const desiredHas = Object.hasOwn(desired, key); const baseValue = base[key]; const desiredValue = desired[key]; if (baseHas === desiredHas && configValuesEqual(baseValue, desiredValue)) { continue; } if (!desiredHas) { if (key !== "lastUpdateId") delete merged[key]; continue; } if ( key === "lastUpdateId" && typeof desiredValue === "number" && typeof merged[key] === "number" ) { merged[key] = Math.max(desiredValue, merged[key] as number); continue; } if ( isPlainConfigRecord(desiredValue) && (!baseHas || isPlainConfigRecord(baseValue)) ) { merged[key] = mergeTelegramConfigDelta( isPlainConfigRecord(baseValue) ? baseValue : {}, desiredValue, isPlainConfigRecord(merged[key]) ? merged[key] : {}, ); continue; } merged[key] = cloneTelegramConfig(desiredValue); } return merged; } function readTelegramConfigForTransaction(configPath: string): TelegramConfig { if (!existsSync(configPath)) return {}; const parsed: unknown = JSON.parse(readFileSync(configPath, "utf8")); if (!isPlainConfigRecord(parsed)) { throw new Error(`Invalid Telegram config object: ${configPath}`); } return parsed as TelegramConfig; } function writeTelegramConfigInTransaction( agentDir: string, configPath: string, config: TelegramConfig, ): void { mkdirSync(agentDir, { recursive: true, mode: 0o700 }); const tempConfigPath = `${configPath}.tmp-${process.pid}-${randomUUID()}`; writeFileSync(tempConfigPath, `${JSON.stringify(config, null, "\t")}\n`, { encoding: "utf8", mode: 0o600, }); chmodSync(tempConfigPath, 0o600); renameSync(tempConfigPath, configPath); chmodSync(configPath, 0o600); } export function getTelegramProfileFields( config: TelegramConfig, ): TelegramBotProfile | undefined { const token = config.botToken?.trim(); if (!token) return undefined; return { botToken: token, botUsername: config.botUsername, botId: config.botId, allowedUserId: config.allowedUserId, lastUpdateId: config.lastUpdateId, }; } function applyTelegramProfile( config: TelegramConfig, profileName: string | undefined, ): TelegramConfig { if (!profileName) return config; const profile = config.profiles?.[profileName]; if (!profile) return config; return { ...config, botToken: profile.botToken, botUsername: profile.botUsername, botId: profile.botId, allowedUserId: profile.allowedUserId, lastUpdateId: profile.lastUpdateId, }; } function storeTelegramEffectiveConfig( baseConfig: TelegramConfig, nextConfig: TelegramConfig, profileName: string | undefined, ): TelegramConfig { if (!profileName) return nextConfig; const profile = getTelegramProfileFields(nextConfig); const profiles = { ...(baseConfig.profiles ?? {}) }; if (profile) profiles[profileName] = profile; else delete profiles[profileName]; return { ...nextConfig, botToken: baseConfig.botToken, botUsername: baseConfig.botUsername, botId: baseConfig.botId, allowedUserId: baseConfig.allowedUserId, lastUpdateId: baseConfig.lastUpdateId, profiles: Object.keys(profiles).length > 0 ? profiles : undefined, }; } export function createTelegramConfigStore( options: TelegramConfigStoreOptions = {}, ): TelegramConfigStore { let config: TelegramConfig = cloneTelegramConfig(options.initialConfig ?? {}); let persistedConfig: TelegramConfig = {}; let mutationVersion = 0; let persistQueue: Promise = Promise.resolve(); let activeProfileName: string | undefined; const agentDir = options.agentDir ?? resolveAgentDir(); const configPath = options.configPath ?? getConfigPath(); const getEffectiveConfig = () => applyTelegramProfile(config, activeProfileName); const setEffectiveConfig = (nextConfig: TelegramConfig) => { config = storeTelegramEffectiveConfig( config, nextConfig, activeProfileName, ); mutationVersion += 1; }; return { get: getEffectiveConfig, getStoredConfig: () => config, set: setEffectiveConfig, update: (mutate) => { const nextConfig = getEffectiveConfig(); mutate(nextConfig); setEffectiveConfig(nextConfig); }, activateProfile: (profileName) => { if (profileName && !config.profiles?.[profileName]) return false; activeProfileName = profileName; return true; }, getActiveProfileName: () => activeProfileName, getBotToken: () => getEffectiveConfig().botToken, hasBotToken: () => !!getEffectiveConfig().botToken, getAllowedUserId: () => getEffectiveConfig().allowedUserId, getInboundHandlers: () => [ ...(config.inboundHandlers ?? []), ...(config.attachmentHandlers ?? []), ], getAttachmentHandlers: () => config.attachmentHandlers, getOutboundHandlers: () => config.outboundHandlers, setAllowedUserId: (userId) => { const nextConfig = getEffectiveConfig(); nextConfig.allowedUserId = userId; setEffectiveConfig(nextConfig); }, load: async () => { config = await readTelegramConfig(configPath, { onInvalidConfig: (recovery) => { options.recordRuntimeEvent?.("config", recovery.error, { phase: "load", configPath: recovery.configPath, recoveryPath: recovery.recoveryPath, }); }, }); persistedConfig = cloneTelegramConfig(config); mutationVersion += 1; }, persist: (nextConfig = getEffectiveConfig()) => { const profileName = activeProfileName; const desiredConfig = storeTelegramEffectiveConfig( config, cloneTelegramConfig(nextConfig), profileName, ); const baseConfig = cloneTelegramConfig(persistedConfig); const capturedMutationVersion = mutationVersion; const persist = persistQueue.then(() => { const mergedConfig = withTelegramFileTransaction( `${configPath}.transaction`, () => { const latestConfig = readTelegramConfigForTransaction(configPath); const merged = mergeTelegramConfigDelta( baseConfig as Record, desiredConfig as Record, latestConfig as Record, ) as TelegramConfig; writeTelegramConfigInTransaction(agentDir, configPath, merged); return merged; }, ); persistedConfig = cloneTelegramConfig(mergedConfig); if (mutationVersion === capturedMutationVersion) { config = cloneTelegramConfig(mergedConfig); } else { config = mergeTelegramConfigDelta( baseConfig as Record, config as Record, mergedConfig as Record, ) as TelegramConfig; } }); persistQueue = persist.catch(() => undefined); return persist; }, }; } export function createTelegramProactivePushChecker( configStore: Pick, ): () => boolean { return () => configStore.get().proactivePush ?? false; } export function createTelegramProactivePushSetter( configStore: TelegramMutableConfigStore, ): (enabled: boolean) => Promise { return async (enabled) => { await loadLatestTelegramConfig(configStore); const config = { ...configStore.get(), proactivePush: enabled }; configStore.set(config); await configStore.persist(config); }; } export function createTelegramDraftPreviewsChecker( configStore: Pick, ): () => boolean { return () => { const config = configStore.get(); return ( config.assistant?.draftPreviews ?? config.draftPreviews ?? config.richDraftPreviews ?? false ); }; } export function createTelegramDraftPreviewsSetter( configStore: TelegramMutableConfigStore, ): (enabled: boolean) => Promise { return async (enabled) => { await loadLatestTelegramConfig(configStore); const { draftPreviews: _legacyDraftPreviews, richDraftPreviews: _legacyRichDraftPreviews, ...current } = configStore.get(); const config = { ...current, assistant: { ...current.assistant, draftPreviews: enabled }, }; configStore.set(config); await configStore.persist(config); }; } export function createTelegramAssistantRenderingModeGetter( configStore: Pick, ): () => TelegramAssistantRenderingMode { return () => { const config = configStore.get(); const mode = config.assistant?.rendering ?? config.assistantRendering; return mode === "html" ? "html" : "rich"; }; } export function createTelegramAssistantRenderingModeSetter( configStore: TelegramMutableConfigStore, ): (mode: TelegramAssistantRenderingMode) => Promise { return async (mode) => { await loadLatestTelegramConfig(configStore); const { assistantRendering: _legacyAssistantRendering, ...current } = configStore.get(); const config = { ...current, assistant: { ...current.assistant, rendering: mode }, }; configStore.set(config); await configStore.persist(config); }; } export function createTelegramVoiceReplyModeGetter( configStore: Pick, ): () => "manual" | "mirror" | "always" { return () => { const mode = configStore.get().voice?.replyMode; return mode === "mirror" || mode === "always" || mode === "manual" ? mode : "manual"; }; } export function createTelegramVoiceReplyModeConfiguredChecker( configStore: Pick, ): () => boolean { return () => { const mode = configStore.get().voice?.replyMode; return mode === "mirror" || mode === "always" || mode === "manual"; }; } export function createTelegramVoiceReplyModeSetter( configStore: TelegramMutableConfigStore, ): (replyMode: "manual" | "mirror" | "always" | undefined) => Promise { return async (replyMode) => { await loadLatestTelegramConfig(configStore); const current = configStore.get(); if (replyMode === undefined) { const { replyMode: _replyMode, ...remainingVoice } = current.voice ?? {}; const next = { ...current }; if (Object.keys(remainingVoice).length > 0) next.voice = remainingVoice; else delete next.voice; configStore.set(next); await configStore.persist(next); return; } const next = { ...current, voice: { ...(current.voice ?? {}), replyMode } }; configStore.set(next); await configStore.persist(next); }; } function getSystemTimezone(): string { try { const tz = Intl.DateTimeFormat().resolvedOptions().timeZone; return tz && tz.length > 0 ? tz : "UTC"; } catch { return "UTC"; } } export function resolveTelegramTimeConfig( raw: TelegramTimeConfig | undefined, ): ResolvedTelegramTimeConfig { const injectionMode: TelegramTimeMode = raw?.injectionMode === "always" || raw?.injectionMode === "interval" ? raw.injectionMode : "hidden"; const interval = typeof raw?.interval === "number" && Number.isFinite(raw.interval) && raw.interval > 0 ? raw.interval : 60 * 60 * 1000; const timezone = getSystemTimezone(); return { injectionMode, interval, timezone }; } export function createTelegramTimeConfigGetter( configStore: Pick, ): () => ResolvedTelegramTimeConfig { return () => resolveTelegramTimeConfig(configStore.get().time); } export function createTelegramTimeInjectionModeGetter( configStore: Pick, ): () => TelegramTimeMode { return () => resolveTelegramTimeConfig(configStore.get().time).injectionMode; } export function createTelegramTimeInjectionModeSetter( configStore: TelegramMutableConfigStore, ): (injectionMode: TelegramTimeMode) => Promise { return async (injectionMode) => { await loadLatestTelegramConfig(configStore); const current = configStore.get(); if (injectionMode === "hidden") { const { injectionMode: _injectionMode, ...remainingTime } = current.time ?? {}; const next = { ...current }; if (Object.keys(remainingTime).length > 0) next.time = remainingTime; else delete next.time; configStore.set(next); await configStore.persist(next); return; } const next = { ...current, time: { ...(current.time ?? {}), injectionMode }, }; configStore.set(next); await configStore.persist(next); }; } export interface TelegramProactivePushTarget { chatId: number; threadId?: number; } export function createTelegramProactivePushChatIdGetter(deps: { getActiveTurnChatId: () => number | undefined; getAllowedUserId: () => number | undefined; }): () => number | undefined { return () => deps.getActiveTurnChatId() ?? deps.getAllowedUserId(); } export function createTelegramProactivePushTargetGetter(deps: { getActiveTurnTarget: () => TelegramProactivePushTarget | undefined; getAssignedTarget: () => TelegramProactivePushTarget | undefined; getAllowedUserId: () => number | undefined; }): () => TelegramProactivePushTarget | undefined { return () => { const activeTarget = deps.getActiveTurnTarget(); if (activeTarget) return activeTarget; const assignedTarget = deps.getAssignedTarget(); if (assignedTarget) return assignedTarget; const chatId = deps.getAllowedUserId(); return typeof chatId === "number" ? { chatId } : undefined; }; } export function createTelegramConfigControls( configStore: TelegramMutableConfigStore, ) { return { isProactivePushEnabled: createTelegramProactivePushChecker(configStore), setProactivePushEnabled: createTelegramProactivePushSetter(configStore), areDraftPreviewsEnabled: createTelegramDraftPreviewsChecker(configStore), setDraftPreviewsEnabled: createTelegramDraftPreviewsSetter(configStore), getAssistantRenderingMode: createTelegramAssistantRenderingModeGetter(configStore), setAssistantRenderingMode: createTelegramAssistantRenderingModeSetter(configStore), getVoiceReplyMode: createTelegramVoiceReplyModeGetter(configStore), isVoiceReplyModeConfigured: createTelegramVoiceReplyModeConfiguredChecker(configStore), setVoiceReplyMode: createTelegramVoiceReplyModeSetter(configStore), getTimeInjectionMode: createTelegramTimeInjectionModeGetter(configStore), setTimeInjectionMode: createTelegramTimeInjectionModeSetter(configStore), }; } export type TelegramAuthorizationState = { kind: "pair"; userId: number } | { kind: "allow" } | { kind: "deny" }; export interface TelegramUserPairingDeps { allowedUserId?: number; ctx: TContext; setAllowedUserId: (userId: number) => void; persistConfig: () => Promise; updateStatus: (ctx: TContext) => void; } export interface TelegramUserPairingRuntimeDeps { getAllowedUserId: () => number | undefined; setAllowedUserId: (userId: number) => void; persistConfig: () => Promise; updateStatus: (ctx: TContext) => void; /** Called once the user was accepted as owner and the config persisted. */ onPaired?: (userId: number, ctx: TContext) => void; } export interface TelegramUserPairingRuntime { pairIfNeeded: (userId: number, ctx: TContext) => Promise; } export function getTelegramAuthorizationState( userId: number, allowedUserId?: number, ): TelegramAuthorizationState { if (allowedUserId === undefined) { return { kind: "pair", userId }; } if (userId === allowedUserId) { return { kind: "allow" }; } return { kind: "deny" }; } function isTelegramStaleContextError(error: unknown): boolean { return ( error instanceof Error && (error.message.includes("stale after session") || error.message.includes("stale ctx")) ); } export async function pairTelegramUserIfNeeded( userId: number, deps: TelegramUserPairingDeps, ): Promise { const authorization = getTelegramAuthorizationState( userId, deps.allowedUserId, ); if (authorization.kind !== "pair") return false; deps.setAllowedUserId(authorization.userId); await deps.persistConfig(); try { deps.updateStatus(deps.ctx); } catch (error) { if (!isTelegramStaleContextError(error)) throw error; } return true; } export function createTelegramUserPairingRuntime( deps: TelegramUserPairingRuntimeDeps, ): TelegramUserPairingRuntime { return { pairIfNeeded: async (userId, ctx) => { const paired = await pairTelegramUserIfNeeded(userId, { allowedUserId: deps.getAllowedUserId(), ctx, setAllowedUserId: deps.setAllowedUserId, persistConfig: deps.persistConfig, updateStatus: deps.updateStatus, }); if (paired) deps.onPaired?.(userId, ctx); return paired; }, }; }