/** * Telegram bus-aware API runtime * Zones: multi-instance bus, telegram api transport, live instance routing * Wraps the direct Telegram Bot API runtime so follower instances can route outbound calls through the bus leader */ import { markTelegramBusCrossTargetDelivery, stripTelegramBusApiMetadata, } from "./bus.ts"; import { isTelegramMessageNotModifiedError } from "./telegram-api.ts"; import type { TelegramAnswerGuestQueryOptions, TelegramApiCallOptions, TelegramBridgeApiRuntime, TelegramEditMessageTextBody, TelegramSendMessageBody, TelegramSendMessageDraftBody, TelegramSendRichMessageBody, TelegramSendRichMessageDraftBody, TelegramSentMessage, TelegramUpdate, } from "./telegram-api.ts"; export type TelegramBusApiCall = ( method: string, args: unknown[], ) => Promise; export interface TelegramBusAwareApiRuntimeDeps { directRuntime: TelegramBridgeApiRuntime; ownsDirect: () => boolean; callFollowerApi: TelegramBusApiCall; getDefaultTarget?: () => { chatId: number; threadId?: number } | undefined; } function asRecord(value: unknown): Record { return value && typeof value === "object" && !Array.isArray(value) ? (value as Record) : {}; } function asBoolean(value: unknown): boolean { return Boolean(value); } function asSentMessage(value: unknown): TelegramSentMessage { return asRecord(value) as unknown as TelegramSentMessage; } function withDefaultThreadTarget( body: Record, target: { chatId: number; threadId?: number } | undefined, ): Record { if (target?.threadId === undefined) return body; if (body.message_thread_id !== undefined) return body; return body.chat_id === target.chatId ? { ...body, message_thread_id: target.threadId } : body; } function markFollowerCrossTargetDelivery>( body: T, defaultTarget: { chatId: number; threadId?: number } | undefined, ): T { if (!defaultTarget || body.chat_id !== defaultTarget.chatId) return body; const threadId = body.message_thread_id; const isDifferentTarget = threadId === undefined ? defaultTarget.threadId !== undefined : threadId !== defaultTarget.threadId; return isDifferentTarget ? markTelegramBusCrossTargetDelivery(body) : body; } function rejectTelegramDirectOwnership(method: string): Promise { return Promise.reject( new Error(`Telegram ${method} requires direct transport ownership.`), ); } export function createTelegramAggregateTypingActionSender( runtime: Pick, ): (chatId: number) => Promise { return (chatId) => runtime.call("sendChatAction", { chat_id: chatId, action: "typing" }); } export function createTelegramBusAwareApiRuntime( deps: TelegramBusAwareApiRuntimeDeps, ): TelegramBridgeApiRuntime { return { call( method: string, body: Record, options?: TelegramApiCallOptions, ): Promise { return deps.ownsDirect() ? deps.directRuntime.call(method, body, options) : (deps.callFollowerApi("call", [ method, body, options, ]) as Promise); }, callMultipart( method: string, fields: Record, fileField: string, filePath: string, fileName: string, options?: TelegramApiCallOptions, ): Promise { return deps.ownsDirect() ? deps.directRuntime.callMultipart( method, fields, fileField, filePath, fileName, options, ) : (deps.callFollowerApi("callMultipart", [ method, fields, fileField, filePath, fileName, options, ]) as Promise); }, downloadFile(fileId: string, suggestedName: string): Promise { return deps.ownsDirect() ? deps.directRuntime.downloadFile(fileId, suggestedName) : (deps.callFollowerApi("downloadFile", [ fileId, suggestedName, ]) as Promise); }, deleteWebhook(signal?: AbortSignal): Promise { return deps.ownsDirect() ? deps.directRuntime.deleteWebhook(signal) : rejectTelegramDirectOwnership("deleteWebhook"); }, getUpdates( body: Record, signal?: AbortSignal, ): Promise { return deps.ownsDirect() ? deps.directRuntime.getUpdates(body, signal) : rejectTelegramDirectOwnership("getUpdates"); }, setMyCommands(commands): Promise { return deps.ownsDirect() ? deps.directRuntime.setMyCommands(commands) : deps .callFollowerApi("call", ["setMyCommands", { commands }]) .then(asBoolean); }, sendChatAction( chatId: number, action: string, options?: { message_thread_id?: number }, ): Promise { const body = withDefaultThreadTarget( { chat_id: chatId, action, ...(options?.message_thread_id !== undefined ? { message_thread_id: options.message_thread_id } : {}), }, deps.getDefaultTarget?.(), ); return deps.ownsDirect() ? deps.directRuntime.sendChatAction(chatId, action, options) : deps .callFollowerApi("call", ["sendChatAction", body]) .then(asBoolean); }, sendTypingAction( chatId: number, options?: { message_thread_id?: number }, ): Promise { const body = withDefaultThreadTarget( { chat_id: chatId, action: "typing", ...(options?.message_thread_id !== undefined ? { message_thread_id: options.message_thread_id } : {}), }, deps.getDefaultTarget?.(), ); return deps.ownsDirect() ? deps.directRuntime.sendTypingAction(chatId, options) : deps .callFollowerApi("call", ["sendChatAction", body]) .then(asBoolean); }, sendRecordVoiceAction( chatId: number, options?: { message_thread_id?: number }, ): Promise { const body = withDefaultThreadTarget( { chat_id: chatId, action: "record_voice", ...(options?.message_thread_id !== undefined ? { message_thread_id: options.message_thread_id } : {}), }, deps.getDefaultTarget?.(), ); return deps.ownsDirect() ? deps.directRuntime.sendRecordVoiceAction(chatId, options) : deps .callFollowerApi("call", ["sendChatAction", body]) .then(asBoolean); }, sendMessageDraft( chatId: number, draftId: number, text?: string, options?: TelegramSendMessageDraftBody extends Record ? { parse_mode?: string; entities?: unknown[]; message_thread_id?: number; } : never, ): Promise { const body: Record = { chat_id: chatId, draft_id: draftId, }; if (text !== undefined) body.text = text; if (options?.parse_mode !== undefined) body.parse_mode = options.parse_mode; if (options?.entities !== undefined) body.entities = options.entities; if (options?.message_thread_id !== undefined) { body.message_thread_id = options.message_thread_id; } const scopedBody = withDefaultThreadTarget( body, deps.getDefaultTarget?.(), ); return deps.ownsDirect() ? deps.directRuntime.sendMessageDraft(chatId, draftId, text, options) : deps .callFollowerApi("call", ["sendMessageDraft", scopedBody]) .then(asBoolean); }, sendMessage(body: TelegramSendMessageBody): Promise { return deps.ownsDirect() ? deps.directRuntime.sendMessage(stripTelegramBusApiMetadata(body)) : deps .callFollowerApi("call", [ "sendMessage", markFollowerCrossTargetDelivery( body, deps.getDefaultTarget?.(), ), ]) .then(asSentMessage); }, sendRichMessage( body: TelegramSendRichMessageBody, ): Promise { return deps.ownsDirect() ? deps.directRuntime.sendRichMessage(stripTelegramBusApiMetadata(body)) : deps .callFollowerApi("call", [ "sendRichMessage", markFollowerCrossTargetDelivery( body, deps.getDefaultTarget?.(), ), ]) .then(asSentMessage); }, sendRichMessageDraft( body: TelegramSendRichMessageDraftBody, ): Promise { return deps.ownsDirect() ? deps.directRuntime.sendRichMessageDraft(body) : deps .callFollowerApi("call", ["sendRichMessageDraft", body]) .then(asBoolean); }, async editMessageText( body: TelegramEditMessageTextBody, ): Promise<"edited" | "unchanged"> { if (deps.ownsDirect()) return deps.directRuntime.editMessageText(body); try { await deps.callFollowerApi("call", ["editMessageText", body]); return "edited"; } catch (error) { if (isTelegramMessageNotModifiedError(error)) return "unchanged"; throw error; } }, async editMessageReplyMarkup( chatId: number, messageId: number, replyMarkup: unknown, ): Promise { if (deps.ownsDirect()) { await deps.directRuntime.editMessageReplyMarkup( chatId, messageId, replyMarkup, ); return; } await deps.callFollowerApi("call", [ "editMessageReplyMarkup", { chat_id: chatId, message_id: messageId, reply_markup: replyMarkup }, ]); }, async answerCallbackQuery( callbackQueryId: string, text?: string, ): Promise { if (deps.ownsDirect()) { await deps.directRuntime.answerCallbackQuery(callbackQueryId, text); return; } await deps.callFollowerApi("call", [ "answerCallbackQuery", { callback_query_id: callbackQueryId, ...(text !== undefined ? { text } : {}), }, ]); }, async answerGuestQuery( guestQueryId: string, text?: string, options?: TelegramAnswerGuestQueryOptions, ): Promise { if (deps.ownsDirect()) { await deps.directRuntime.answerGuestQuery(guestQueryId, text, options); return; } const body: Record = { guest_query_id: guestQueryId }; if (options?.result) { body.result = options.result; } else if (text !== undefined || options?.richMessage) { const inputContent: Record = options?.richMessage ? { rich_message: options.richMessage } : { message_text: text }; if (!options?.richMessage && options?.parseMode) { inputContent.parse_mode = options.parseMode; } body.result = { type: "article", id: "1", title: "Response", input_message_content: inputContent, }; } await deps.callFollowerApi("call", ["answerGuestQuery", body]); }, async deleteMessage(chatId: number, messageId: number): Promise { if (deps.ownsDirect()) return deps.directRuntime.deleteMessage(chatId, messageId); await deps.callFollowerApi("call", [ "deleteMessage", { chat_id: chatId, message_id: messageId, }, ]); }, prepareTempDir(): Promise { return deps.directRuntime.prepareTempDir(); }, }; }