import assert from 'node:assert/strict'; import { execFile } from 'node:child_process'; import { createServer } from 'node:http'; import { promisify } from 'node:util'; import { resolve } from 'node:path'; import test from 'node:test'; import { TelegramApi } from '../../../src/channels/telegram/telegram-api.mjs'; import { createTelegramHttpTransport } from '../../../src/channels/telegram/telegram-http.mjs'; import { TelegramRuntime } from '../../../src/channels/telegram/telegram-runtime.mjs'; const TOKEN = '123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZabcdef123456'; const execFileAsync = promisify(execFile); const packageRoot = resolve(import.meta.dirname, '../../..'); function deferred() { let resolvePromise; let rejectPromise; const promise = new Promise((resolveValue, rejectValue) => { resolvePromise = resolveValue; rejectPromise = rejectValue; }); return { promise, resolve: resolvePromise, reject: rejectPromise }; } async function bounded(promise, message, timeoutMs = 1_000) { let timer; try { return await Promise.race([ promise, new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(message)), timeoutMs); }), ]); } finally { clearTimeout(timer); } } async function listen(server) { await new Promise((resolvePromise, rejectPromise) => { server.once('error', rejectPromise); server.listen(0, '127.0.0.1', resolvePromise); }); const address = server.address(); return `http://127.0.0.1:${address.port}/`; } async function close(server) { await new Promise((resolvePromise, rejectPromise) => { server.close((error) => (error ? rejectPromise(error) : resolvePromise())); }); } function withLocalhostProxyBypass() { const names = ['http_proxy', 'HTTP_PROXY', 'https_proxy', 'HTTPS_PROXY', 'no_proxy', 'NO_PROXY']; const previous = Object.fromEntries(names.map((name) => [name, process.env[name]])); for (const name of names) delete process.env[name]; process.env.no_proxy = '127.0.0.1,localhost'; process.env.NO_PROXY = '127.0.0.1,localhost'; return () => { for (const name of names) { if (previous[name] === undefined) delete process.env[name]; else process.env[name] = previous[name]; } }; } test('Telegram private transport sends while a long poll remains open', async () => { const pollStarted = deferred(); const releasePoll = deferred(); const events = []; const server = createServer((request, response) => { request.resume(); if (request.url.endsWith('/getUpdates')) { events.push('poll-started'); pollStarted.resolve(); void releasePoll.promise.then(() => { events.push('poll-released'); response.setHeader('content-type', 'application/json'); response.end(JSON.stringify({ ok: true, result: [] })); }); return; } events.push('send-completed'); response.setHeader('content-type', 'application/json'); response.end(JSON.stringify({ ok: true, result: true })); }); const baseUrl = await listen(server); const restoreProxy = withLocalhostProxyBypass(); const transport = createTelegramHttpTransport(); const api = new TelegramApi({ token: TOKEN, baseUrl, fetchImpl: transport.fetchImpl, FormDataImpl: transport.FormDataImpl, }); let pollSettled = false; const poll = api.getUpdates().finally(() => { pollSettled = true; }); try { await bounded(pollStarted.promise, 'Telegram long poll did not start'); assert.equal(await bounded( api.sendChatAction({ chatId: 42 }), 'Telegram send was blocked behind its long poll', ), true); assert.equal(pollSettled, false); assert.deepEqual(events, ['poll-started', 'send-completed']); releasePoll.resolve(); assert.deepEqual(await poll, []); assert.deepEqual(events, ['poll-started', 'send-completed', 'poll-released']); } finally { releasePoll.resolve(); await poll.catch(() => undefined); await transport.destroy(); restoreProxy(); await close(server); } }); test('Telegram private transport is independent from a saturated global dispatcher', async () => { const source = String.raw` import { createServer } from 'node:http'; import { Agent, fetch, setGlobalDispatcher } from 'undici'; import { createTelegramHttpTransport } from './src/channels/telegram/telegram-http.mjs'; let releaseHold; let markHoldStarted; const holdStarted = new Promise((resolve) => { markHoldStarted = resolve; }); const holdReleased = new Promise((resolve) => { releaseHold = resolve; }); const server = createServer((request, response) => { request.resume(); if (request.url === '/hold') { markHoldStarted(); void holdReleased.then(() => response.end('held')); return; } response.end('private'); }); await new Promise((resolve, reject) => { server.once('error', reject); server.listen(0, '127.0.0.1', resolve); }); const origin = 'http://127.0.0.1:' + server.address().port; const globalAgent = new Agent({ connections: 1 }); setGlobalDispatcher(globalAgent); const occupied = fetch(origin + '/hold'); await holdStarted; const transport = createTelegramHttpTransport(); let timeout; try { const response = await Promise.race([ transport.fetchImpl(origin + '/send'), new Promise((_, reject) => { timeout = setTimeout( () => reject(new Error('private transport used the saturated global dispatcher')), 1_000, ); }), ]); if (await response.text() !== 'private') throw new Error('unexpected private response'); process.stdout.write('independent'); } finally { clearTimeout(timeout); releaseHold(); await occupied; await transport.destroy(); await globalAgent.destroy(); await new Promise((resolve, reject) => server.close( (error) => error ? reject(error) : resolve(), )); } `; const childEnv = { ...process.env }; for (const name of ['http_proxy', 'HTTP_PROXY', 'https_proxy', 'HTTPS_PROXY']) { delete childEnv[name]; } childEnv.no_proxy = '127.0.0.1,localhost'; childEnv.NO_PROXY = '127.0.0.1,localhost'; const { stdout, stderr } = await execFileAsync(process.execPath, [ '--input-type=module', '--eval', source, ], { cwd: packageRoot, env: childEnv, timeout: 5_000, }); assert.equal(stderr, ''); assert.equal(stdout, 'independent'); }); test('Telegram API sends real undici multipart data without losing metadata or bytes', async () => { const received = deferred(); const server = createServer((request, response) => { const chunks = []; request.on('data', (chunk) => chunks.push(chunk)); request.on('end', () => { received.resolve({ contentType: request.headers['content-type'], body: Buffer.concat(chunks), }); response.setHeader('content-type', 'application/json'); response.end(JSON.stringify({ ok: true, result: { message_id: 901 } })); }); }); const baseUrl = await listen(server); const restoreProxy = withLocalhostProxyBypass(); const transport = createTelegramHttpTransport(); const bytes = Buffer.from([0, 1, 2, 3, 0xff, 0x80, 0x41]); const api = new TelegramApi({ token: TOKEN, baseUrl, fetchImpl: transport.fetchImpl, FormDataImpl: transport.FormDataImpl, }); try { assert.deepEqual(await api.sendDocument({ chatId: -100123, replyToMessageId: 44, messageThreadId: 55, file: { fileName: 'result.bin', mediaType: 'application/octet-stream', bytes, }, }), { message_id: 901 }); const request = await bounded(received.promise, 'Telegram multipart request was not received'); assert.match(request.contentType, /^multipart\/form-data; boundary=/); assert.ok(request.body.includes(Buffer.from('name="chat_id"\r\n\r\n-100123'))); assert.ok(request.body.includes(Buffer.from('name="message_thread_id"\r\n\r\n55'))); assert.ok(request.body.includes(Buffer.from('name="reply_parameters"'))); assert.ok(request.body.includes(Buffer.from('"message_id":44'))); assert.ok(request.body.includes(Buffer.from( 'name="document"; filename="result.bin"\r\nContent-Type: application/octet-stream', ))); assert.ok(request.body.includes(bytes)); } finally { await transport.destroy(); restoreProxy(); await close(server); } }); function runtimeState(cursor = 0) { return { cursor: () => cursor, setCursor: async (value) => { cursor = value; }, hasSeen: () => false, markSeen: async () => undefined, sessionFor: () => null, }; } function runtimeConfig() { return { botId: 'telegram_transport_test', platformId: '123456789', username: 'HarnessBot', }; } function healthyApi() { return { getMe: async () => ({ id: 123456789, is_bot: true }), getWebhookInfo: async () => ({ url: '' }), setMyCommands: async () => true, setChatMenuButton: async () => true, getUpdates: async ({ signal }) => new Promise((_, reject) => { if (signal.aborted) reject(signal.reason); else signal.addEventListener('abort', () => reject(signal.reason), { once: true }); }), }; } function fakeTransport() { let destroyCalls = 0; class FakeFormData {} const fetchImpl = async () => undefined; return { fetchImpl, FormDataImpl: FakeFormData, destroy: async () => { destroyCalls += 1; }, get destroyCalls() { return destroyCalls; }, }; } test('Telegram Runtime injects and destroys one private transport exactly once', async () => { const transport = fakeTransport(); let apiOptions; const runtime = new TelegramRuntime({ config: runtimeConfig(), token: TOKEN, harness: { ensureRunning: async () => true }, state: runtimeState(), createHttpTransport: () => transport, createApi: (options) => { apiOptions = options; return healthyApi(); }, }); await runtime.start(); assert.equal(apiOptions.fetchImpl, transport.fetchImpl); assert.equal(apiOptions.FormDataImpl, transport.FormDataImpl); await runtime.stop(); await runtime.stop(); assert.equal(transport.destroyCalls, 1); }); for (const failurePoint of ['getMe', 'createApi']) { test(`Telegram Runtime destroys its private transport when ${failurePoint} fails`, async () => { const transport = fakeTransport(); const failure = new Error(`${failurePoint} failed`); const runtime = new TelegramRuntime({ config: runtimeConfig(), token: TOKEN, harness: { ensureRunning: async () => true }, state: runtimeState(), createHttpTransport: () => transport, createApi: failurePoint === 'createApi' ? () => { throw failure; } : () => ({ ...healthyApi(), getMe: async () => { throw failure; } }), }); await assert.rejects(runtime.start(), (error) => error === failure); assert.equal(transport.destroyCalls, 1); await runtime.stop(); assert.equal(transport.destroyCalls, 1); }); }