diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1ba84d2d..c7b0bb07 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -91,11 +91,13 @@ jobs: - run: bun install --frozen-lockfile - name: Build internal TypeScript dependencies - run: bun run --filter @wateaminbox/shared build && bun run --filter @wateaminbox/database build + run: bun run --filter @wateaminbox/shared build && bun run --filter @wateaminbox/database build && bun run --filter @wateaminbox/adapter-telegram build - name: API unit tests run: cd apps/api && bun test - name: Web unit tests run: cd apps/web && bun test + - name: Telegram adapter unit tests + run: cd packages/adapter-telegram && bun test - name: Shared unit tests run: cd packages/shared && bun test @@ -154,7 +156,7 @@ jobs: - run: bun install --frozen-lockfile - name: Build internal TypeScript dependencies - run: bun run --filter @wateaminbox/shared build && bun run --filter @wateaminbox/database build + run: bun run --filter @wateaminbox/shared build && bun run --filter @wateaminbox/database build && bun run --filter @wateaminbox/adapter-telegram build - run: bun run db:migrate - name: Run all TypeScript integration tests (recursive discovery) diff --git a/apps/api/Dockerfile b/apps/api/Dockerfile index b5799a36..fa7751c8 100644 --- a/apps/api/Dockerfile +++ b/apps/api/Dockerfile @@ -6,6 +6,7 @@ WORKDIR /app COPY package.json bun.lock turbo.json ./ COPY apps/api/package.json apps/api/package.json COPY apps/web/package.json apps/web/package.json +COPY packages/adapter-telegram/package.json packages/adapter-telegram/package.json COPY packages/database/package.json packages/database/package.json COPY packages/shared/package.json packages/shared/package.json COPY packages/ui/package.json packages/ui/package.json @@ -16,10 +17,12 @@ RUN --mount=type=cache,target=/root/.bun/install/cache \ FROM workspace AS build COPY apps/api apps/api +COPY packages/adapter-telegram packages/adapter-telegram COPY packages/database packages/database COPY packages/shared packages/shared RUN bun run --filter @wateaminbox/shared build \ && bun run --filter @wateaminbox/database build \ + && bun run --filter @wateaminbox/adapter-telegram build \ && bun run --filter @wateaminbox/api build FROM workspace AS production-dependencies diff --git a/apps/api/package.json b/apps/api/package.json index 235a91d5..9517a700 100644 --- a/apps/api/package.json +++ b/apps/api/package.json @@ -17,6 +17,7 @@ "@hono/mcp": "^0.3.2", "@hono/zod-validator": "^0.9.1", "@modelcontextprotocol/sdk": "^1.30.0", + "@wateaminbox/adapter-telegram": "workspace:*", "@wateaminbox/database": "workspace:*", "@wateaminbox/shared": "workspace:*", "bcrypt": "^6.0.0", diff --git a/apps/api/src/channel-spine/application/adapter-conformance.test.ts b/apps/api/src/channel-spine/application/adapter-conformance.test.ts index 295e99c2..4db4b365 100644 --- a/apps/api/src/channel-spine/application/adapter-conformance.test.ts +++ b/apps/api/src/channel-spine/application/adapter-conformance.test.ts @@ -8,8 +8,10 @@ import { type ProviderSendResult, type ResolvedCapabilities, } from "@wateaminbox/shared"; -import { TelegramBotAdapter } from "../providers/telegram-bot/adapter"; -import { classifyTelegramSendFailure } from "../providers/telegram-bot/transport"; +import { + classifyTelegramSendFailure, + TelegramBotAdapter, +} from "@wateaminbox/adapter-telegram"; import { WhatsAppLinkedDeviceAdapter } from "../providers/whatsapp-linked-device/adapter"; import { storedContentType } from "../../services/channel-attachment-fetch.service"; import { diff --git a/apps/api/src/channel-spine/providers/telegram-bot/ports.integration.test.ts b/apps/api/src/channel-spine/providers/telegram-bot/ports.integration.test.ts new file mode 100644 index 00000000..b6abf017 --- /dev/null +++ b/apps/api/src/channel-spine/providers/telegram-bot/ports.integration.test.ts @@ -0,0 +1,255 @@ +import { describe, expect, test } from "bun:test"; +import { db } from "@wateaminbox/database"; +import { TelegramLocalFailureError } from "@wateaminbox/adapter-telegram"; +import { ChannelCredentialCipher } from "../../../services/channel-credential.service.js"; +import { sql } from "kysely"; +import { + canStoreChannelCredentials, + storeChannelCredential, +} from "../../../services/channel-credential.service.js"; +import { + clearTenantConnection, + createTenantSchema, + getSchemaName, + getTenantConnection, +} from "../../../services/tenant.service.js"; +import { telegramTransportPorts } from "./ports.js"; + +const integrationTest = + process.env.RUN_DB_INTEGRATION === "1" && canStoreChannelCredentials() + ? test + : test.skip; + +const token = `12345:${"a".repeat(30)}`; + +/** + * The application half of the adapter's outbound port. + * + * Every refusal here has to arrive as a closed local failure code, because + * those are the only outcomes the transport may classify permanent without + * hearing from Telegram. Anything else degrades to "uncertain" and strands + * the send instead of failing it. + */ +describe("telegramTransportPorts.resolveOutboundContext", () => { + integrationTest( + "resolves a bot's chat target and refuses every unsendable conversation by code", + async () => { + const companyId = crypto.randomUUID(); + const schemaName = getSchemaName(companyId); + try { + await db + .insertInto("companies") + .values({ + id: companyId, + name: "Telegram ports test", + schema_name: schemaName, + status: "active", + }) + .execute(); + await createTenantSchema(companyId); + const tenantDb = getTenantConnection(companyId); + + const accountId = crypto.randomUUID(); + await tenantDb + .insertInto("channel_accounts") + .values({ + id: accountId, + channel: "telegram", + provider: "telegram_bot", + display_name: "Support bot", + status: "connected", + }) + .execute(); + await storeChannelCredential( + tenantDb, + companyId, + accountId, + "telegram_bot_token", + token, + ); + + const conversationId = async (externalThreadId: string | null) => + ( + await tenantDb + .insertInto("conversations") + .values({ + channel_account_id: accountId, + client_thread_key: `telegram:${crypto.randomUUID()}`, + external_thread_id: externalThreadId, + kind: "direct", + subject: null, + legacy_contact_id: null, + }) + .returning("id") + .executeTakeFirstOrThrow() + ).id; + + // A forum topic carries its thread id in the same column, and the Bot + // API needs it split back out or the reply lands in the wrong topic. + const topic = await conversationId("-1001234567890:thread:42"); + expect( + await telegramTransportPorts.resolveOutboundContext({ + companyId, + channelAccountId: accountId, + conversationId: topic, + }), + ).toEqual({ + token, + chatId: "-1001234567890", + messageThreadId: 42, + externalThreadId: "-1001234567890:thread:42", + }); + + const direct = await conversationId("987654321"); + const resolved = await telegramTransportPorts.resolveOutboundContext({ + companyId, + channelAccountId: accountId, + conversationId: direct, + }); + expect(resolved.chatId).toBe("987654321"); + expect(resolved.messageThreadId).toBeUndefined(); + + // A thread id the Bot API cannot address is a permanent refusal, not + // an attempt against a chat id parsed out of a malformed string. + const malformed = await conversationId("not-a-chat-id"); + await expect( + telegramTransportPorts.resolveOutboundContext({ + companyId, + channelAccountId: accountId, + conversationId: malformed, + }), + ).rejects.toThrow( + new TelegramLocalFailureError("telegram_conversation_invalid"), + ); + + // A conversation the spine never gave an external thread has nowhere + // to send, and so does one on an archived conversation. + const unmapped = await conversationId(null); + await expect( + telegramTransportPorts.resolveOutboundContext({ + companyId, + channelAccountId: accountId, + conversationId: unmapped, + }), + ).rejects.toThrow( + new TelegramLocalFailureError("telegram_conversation_unavailable"), + ); + await tenantDb + .updateTable("conversations") + .set({ archived_at: new Date() }) + .where("id", "=", direct) + .execute(); + await expect( + telegramTransportPorts.resolveOutboundContext({ + companyId, + channelAccountId: accountId, + conversationId: direct, + }), + ).rejects.toThrow( + new TelegramLocalFailureError("telegram_conversation_unavailable"), + ); + + // A paused bot still owns its conversations and its credential. Sending + // through one anyway would reach Telegram over a webhook we revoked. + await tenantDb + .updateTable("channel_accounts") + .set({ status: "disabled" }) + .where("id", "=", accountId) + .execute(); + await expect( + telegramTransportPorts.resolveOutboundContext({ + companyId, + channelAccountId: accountId, + conversationId: topic, + }), + ).rejects.toThrow( + new TelegramLocalFailureError("telegram_conversation_unavailable"), + ); + await tenantDb + .updateTable("channel_accounts") + .set({ status: "connected" }) + .where("id", "=", accountId) + .execute(); + + // Another account's conversation is not reachable with this account's + // token, even inside the same workspace. + const otherAccountId = crypto.randomUUID(); + await tenantDb + .insertInto("channel_accounts") + .values({ + id: otherAccountId, + channel: "telegram", + provider: "telegram_bot", + display_name: "Other bot", + status: "connected", + }) + .execute(); + await storeChannelCredential( + tenantDb, + companyId, + otherAccountId, + "telegram_bot_token", + token, + ); + await expect( + telegramTransportPorts.resolveOutboundContext({ + companyId, + channelAccountId: otherAccountId, + conversationId: topic, + }), + ).rejects.toThrow( + new TelegramLocalFailureError("telegram_conversation_unavailable"), + ); + + // An account whose credential is gone reports exactly that, so the + // operator is told to reconnect rather than shown a generic failure. + await tenantDb + .deleteFrom("channel_account_credentials") + .where("channel_account_id", "=", accountId) + .execute(); + await expect( + telegramTransportPorts.resolveOutboundContext({ + companyId, + channelAccountId: accountId, + conversationId: topic, + }), + ).rejects.toThrow( + new TelegramLocalFailureError("telegram_credential_unavailable"), + ); + // A credential stored under a key version this process no longer holds + // must surface as the adapter's own code. Untranslated it would reach + // classifyTelegramSendFailure as an unrecognised error and be called + // "uncertain", which is never retried - parking the send for ever over + // a fault an operator can fix. + const strandedKey = new ChannelCredentialCipher( + `retired:${Buffer.alloc(32, 9).toString("base64")}`, + "retired", + ); + await storeChannelCredential( + tenantDb, + companyId, + accountId, + "telegram_bot_token", + token, + strandedKey, + ); + await expect( + telegramTransportPorts.resolveOutboundContext({ + companyId, + channelAccountId: accountId, + conversationId: topic, + }), + ).rejects.toThrow( + new TelegramLocalFailureError("telegram_credential_key_unavailable"), + ); + } finally { + await clearTenantConnection(companyId); + await sql + .raw(`DROP SCHEMA IF EXISTS "${schemaName}" CASCADE`) + .execute(db); + await db.deleteFrom("companies").where("id", "=", companyId).execute(); + } + }, + 120_000, + ); +}); diff --git a/apps/api/src/channel-spine/providers/telegram-bot/ports.ts b/apps/api/src/channel-spine/providers/telegram-bot/ports.ts new file mode 100644 index 00000000..feccd9d1 --- /dev/null +++ b/apps/api/src/channel-spine/providers/telegram-bot/ports.ts @@ -0,0 +1,117 @@ +import { + parseTelegramThreadTarget, + TelegramLocalFailureError, + type TelegramOutboundContext, + type TelegramOutboundTarget, + type TelegramTransportPorts, +} from "@wateaminbox/adapter-telegram"; +import { + getPresignedUrl, + resolveMediaKeyForCompany, +} from "../../../lib/storage.js"; +import { + ChannelCredentialKeyError, + readChannelCredential, +} from "../../../services/channel-credential.service.js"; +import { getTenantConnection } from "../../../services/tenant.service.js"; +import type { TenantDatabase } from "@wateaminbox/database"; +import type { Kysely } from "kysely"; + +/** + * The application half of Telegram outbound sending. + * + * The adapter package holds no tenant database or media storage knowledge, so + * the credential read, the conversation lookup, and media presigning are bound + * here and injected into the transport by the registry. + */ +export const telegramTransportPorts: TelegramTransportPorts = { + resolveOutboundContext, + resolveAttachmentUrl, +}; + +async function resolveOutboundContext( + target: TelegramOutboundTarget, +): Promise { + const tenantDb = await getTenantConnection(target.companyId); + const [token, conversation] = await Promise.all([ + readBotToken(tenantDb, target), + tenantDb + .selectFrom("conversations as conversation") + .innerJoin( + "channel_accounts as account", + "account.id", + "conversation.channel_account_id", + ) + .select([ + "conversation.external_thread_id", + "account.status as account_status", + ]) + .where("conversation.id", "=", target.conversationId) + .where("conversation.channel_account_id", "=", target.channelAccountId) + .where("conversation.archived_at", "is", null) + .where("account.archived_at", "is", null) + .executeTakeFirst(), + ]); + if (!token) { + throw new TelegramLocalFailureError("telegram_credential_unavailable"); + } + if ( + !conversation?.external_thread_id || + conversation.account_status !== "connected" + ) { + throw new TelegramLocalFailureError("telegram_conversation_unavailable"); + } + const { chatId, messageThreadId } = parseTelegramThreadTarget( + conversation.external_thread_id, + ); + return { + token, + chatId, + messageThreadId, + externalThreadId: conversation.external_thread_id, + }; +} + +/** + * The bot token, with a missing keyring translated into the adapter's protocol. + * + * The adapter classifies by code and knows nothing of this application's error + * types. Left untranslated, a key the process was started without falls through + * to "uncertain" - never retried, because Telegram has no idempotency key - so + * an operator-fixable fault would park the send for ever. + */ +async function readBotToken( + tenantDb: Kysely, + target: TelegramOutboundTarget, +): Promise { + try { + return await readChannelCredential( + tenantDb, + target.companyId, + target.channelAccountId, + "telegram_bot_token", + ); + } catch (error) { + if (error instanceof ChannelCredentialKeyError) { + throw new TelegramLocalFailureError( + "telegram_credential_key_unavailable", + ); + } + throw error; + } +} + +async function resolveAttachmentUrl( + companyId: string, + attachment: Record | undefined, +): Promise { + const storageUri = + attachment && typeof attachment.storageUri === "string" + ? attachment.storageUri + : undefined; + if (!storageUri) { + throw new TelegramLocalFailureError("telegram_attachment_missing"); + } + const key = resolveMediaKeyForCompany(storageUri, companyId); + return getPresignedUrl(key, 10 * 60); +} diff --git a/apps/api/src/channel-spine/registry.ts b/apps/api/src/channel-spine/registry.ts index 4f1b97cb..01938610 100644 --- a/apps/api/src/channel-spine/registry.ts +++ b/apps/api/src/channel-spine/registry.ts @@ -1,7 +1,13 @@ +import { + TelegramBotAdapter, + TelegramBotApiTransport, +} from "@wateaminbox/adapter-telegram"; import { ChannelAdapterRegistry } from "./application/adapter-registry.js"; -import { resolveTelegramWebhookSecret } from "../services/channel-credential.service.js"; -import { TelegramBotAdapter } from "./providers/telegram-bot/adapter.js"; -import { TelegramBotApiTransport } from "./providers/telegram-bot/transport.js"; +import { + ChannelCredentialKeyError, + resolveTelegramWebhookSecret, +} from "../services/channel-credential.service.js"; +import { telegramTransportPorts } from "./providers/telegram-bot/ports.js"; import { WhatsAppLinkedDeviceAdapter } from "./providers/whatsapp-linked-device/adapter.js"; import { LinkedDeviceNatsTransport } from "./providers/whatsapp-linked-device/transport.js"; @@ -13,6 +19,8 @@ channelAdapterRegistry.register( channelAdapterRegistry.register( new TelegramBotAdapter({ resolveWebhookSecret: resolveTelegramWebhookSecret, - outboundTransport: new TelegramBotApiTransport(), + isCredentialUnavailable: (error) => + error instanceof ChannelCredentialKeyError, + outboundTransport: new TelegramBotApiTransport(telegramTransportPorts), }), ); diff --git a/apps/api/src/routes/channel-accounts.ts b/apps/api/src/routes/channel-accounts.ts index 990db1fc..d4ca4ea1 100644 --- a/apps/api/src/routes/channel-accounts.ts +++ b/apps/api/src/routes/channel-accounts.ts @@ -10,7 +10,7 @@ import { configureTelegramWebhook, getTelegramBotIdentity, removeTelegramWebhook, -} from "../channel-spine/providers/telegram-bot/api.js"; +} from "@wateaminbox/adapter-telegram"; import { channelAdapterRegistry } from "../channel-spine/registry.js"; import { env } from "../lib/env.js"; import { conflict, forbidden, notFound } from "../lib/errors.js"; diff --git a/apps/api/src/services/channel-attachment-fetch.service.ts b/apps/api/src/services/channel-attachment-fetch.service.ts index fdd5fea6..d5333a10 100644 --- a/apps/api/src/services/channel-attachment-fetch.service.ts +++ b/apps/api/src/services/channel-attachment-fetch.service.ts @@ -1,6 +1,6 @@ import { db } from "@wateaminbox/database"; import { sql } from "kysely"; -import { downloadTelegramFile } from "../channel-spine/providers/telegram-bot/api.js"; +import { downloadTelegramFile } from "@wateaminbox/adapter-telegram"; import { createLogger, formatError } from "../lib/logger.js"; import { deleteMedia, uploadMedia } from "../lib/storage.js"; import { diff --git a/apps/api/src/services/channel-endpoint-avatar.service.ts b/apps/api/src/services/channel-endpoint-avatar.service.ts index a56509d5..071ef9a1 100644 --- a/apps/api/src/services/channel-endpoint-avatar.service.ts +++ b/apps/api/src/services/channel-endpoint-avatar.service.ts @@ -3,7 +3,7 @@ import { sql } from "kysely"; import { downloadTelegramFile, getTelegramProfilePhotoFileId, -} from "../channel-spine/providers/telegram-bot/api.js"; +} from "@wateaminbox/adapter-telegram"; import { createLogger, formatError } from "../lib/logger.js"; import { uploadMedia } from "../lib/storage.js"; import { readChannelCredential } from "./channel-credential.service.js"; diff --git a/apps/web/Dockerfile b/apps/web/Dockerfile index ea7d8c33..3a4c6229 100644 --- a/apps/web/Dockerfile +++ b/apps/web/Dockerfile @@ -6,6 +6,7 @@ WORKDIR /app COPY package.json bun.lock turbo.json ./ COPY apps/web/package.json apps/web/package.json COPY apps/api/package.json apps/api/package.json +COPY packages/adapter-telegram/package.json packages/adapter-telegram/package.json COPY packages/database/package.json packages/database/package.json COPY packages/shared/package.json packages/shared/package.json COPY packages/ui/package.json packages/ui/package.json diff --git a/bun.lock b/bun.lock index 9e8cdeeb..0fc8c930 100644 --- a/bun.lock +++ b/bun.lock @@ -1,6 +1,5 @@ { "lockfileVersion": 1, - "configVersion": 0, "workspaces": { "": { "name": "wateaminbox", @@ -20,6 +19,7 @@ "@hono/mcp": "^0.3.2", "@hono/zod-validator": "^0.9.1", "@modelcontextprotocol/sdk": "^1.30.0", + "@wateaminbox/adapter-telegram": "workspace:*", "@wateaminbox/database": "workspace:*", "@wateaminbox/shared": "workspace:*", "bcrypt": "^6.0.0", @@ -90,6 +90,18 @@ "vite": "^7.3.6", }, }, + "packages/adapter-telegram": { + "name": "@wateaminbox/adapter-telegram", + "version": "0.0.1", + "dependencies": { + "@wateaminbox/shared": "workspace:*", + }, + "devDependencies": { + "@types/bun": "latest", + "tsup": "^8.0.1", + "typescript": "^5.7.2", + }, + }, "packages/database": { "name": "@wateaminbox/database", "version": "0.0.1", @@ -674,6 +686,8 @@ "@vitejs/plugin-react": ["@vitejs/plugin-react@5.2.0", "", { "dependencies": { "@babel/core": "^7.29.0", "@babel/plugin-transform-react-jsx-self": "^7.27.1", "@babel/plugin-transform-react-jsx-source": "^7.27.1", "@rolldown/pluginutils": "1.0.0-rc.3", "@types/babel__core": "^7.20.5", "react-refresh": "^0.18.0" }, "peerDependencies": { "vite": "^4.2.0 || ^5.0.0 || ^6.0.0 || ^7.0.0 || ^8.0.0" } }, "sha512-YmKkfhOAi3wsB1PhJq5Scj3GXMn3WvtQ/JC0xoopuHoXSdmtdStOpFrYaT1kie2YgFBcIe64ROzMYRjCrYOdYw=="], + "@wateaminbox/adapter-telegram": ["@wateaminbox/adapter-telegram@workspace:packages/adapter-telegram"], + "@wateaminbox/api": ["@wateaminbox/api@workspace:apps/api"], "@wateaminbox/database": ["@wateaminbox/database@workspace:packages/database"], diff --git a/knip.json b/knip.json index cb47d4d8..8e151af9 100644 --- a/knip.json +++ b/knip.json @@ -24,6 +24,10 @@ ], "project": ["src/**/*.ts"] }, + "packages/adapter-telegram": { + "entry": ["src/**/*.test.ts"], + "project": ["src/**/*.ts"] + }, "packages/shared": { "entry": ["src/**/*.test.ts"], "project": ["src/**/*.ts"] diff --git a/package.json b/package.json index b155cd73..82a95034 100644 --- a/package.json +++ b/package.json @@ -11,7 +11,7 @@ "preview:connection-emails": "bun --preload ./apps/api/src/test-env.ts scripts/preview-connection-emails.ts", "dev": "turbo dev", "lint": "biome lint apps/api apps/web packages && ./scripts/lint-go.sh", - "typecheck": "bun run --filter @wateaminbox/shared build && bun run --filter @wateaminbox/database build && tsc -p apps/api/tsconfig.json --noEmit && tsc -p apps/web/tsconfig.json --noEmit", + "typecheck": "bun run --filter @wateaminbox/shared build && bun run --filter @wateaminbox/database build && bun run --filter @wateaminbox/adapter-telegram build && tsc -p apps/api/tsconfig.json --noEmit && tsc -p apps/web/tsconfig.json --noEmit", "check:unused": "knip", "format": "biome format --write apps/api apps/web packages", "test": "./scripts/run-tests.sh", diff --git a/packages/adapter-telegram/package.json b/packages/adapter-telegram/package.json new file mode 100644 index 00000000..28377e5e --- /dev/null +++ b/packages/adapter-telegram/package.json @@ -0,0 +1,31 @@ +{ + "name": "@wateaminbox/adapter-telegram", + "version": "0.0.1", + "private": true, + "description": "Telegram Bot API channel adapter for the channel-neutral spine", + "main": "./dist/index.js", + "module": "./dist/index.mjs", + "types": "./dist/index.d.ts", + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.mjs", + "require": "./dist/index.js" + } + }, + "scripts": { + "build": "tsup src/index.ts --format cjs,esm --dts", + "dev": "tsup src/index.ts --format cjs,esm --dts --watch", + "lint": "bunx biome lint src", + "format": "bunx biome format --write src", + "test": "bun test" + }, + "dependencies": { + "@wateaminbox/shared": "workspace:*" + }, + "devDependencies": { + "@types/bun": "latest", + "tsup": "^8.0.1", + "typescript": "^5.7.2" + } +} diff --git a/apps/api/src/channel-spine/providers/telegram-bot/adapter.test.ts b/packages/adapter-telegram/src/adapter.test.ts similarity index 100% rename from apps/api/src/channel-spine/providers/telegram-bot/adapter.test.ts rename to packages/adapter-telegram/src/adapter.test.ts diff --git a/apps/api/src/channel-spine/providers/telegram-bot/adapter.ts b/packages/adapter-telegram/src/adapter.ts similarity index 89% rename from apps/api/src/channel-spine/providers/telegram-bot/adapter.ts rename to packages/adapter-telegram/src/adapter.ts index e4916e95..43e2fa90 100644 --- a/apps/api/src/channel-spine/providers/telegram-bot/adapter.ts +++ b/packages/adapter-telegram/src/adapter.ts @@ -9,7 +9,6 @@ import type { ProviderSendResult, ResolvedCapabilities, } from "@wateaminbox/shared"; -import { ChannelCredentialKeyError } from "../../../services/channel-credential.service.js"; import { normalizeTelegramUpdate } from "./normalize.js"; const SECRET_HEADER = "x-telegram-bot-api-secret-token"; @@ -32,6 +31,16 @@ export interface TelegramBotOutboundTransport { export interface TelegramBotAdapterDependencies { resolveWebhookSecret: TelegramWebhookSecretResolver; outboundTransport?: TelegramBotOutboundTransport; + /** + * Whether a resolver failure is the host's fault rather than a bad signature. + * + * The adapter cannot recognise the host's error types, and the distinction + * matters: a key this process was started without must reach the caller + * intact so it can answer 503, while anything else is a verification + * failure. Without this, a deployment fault is reported to the provider as + * "your secret is wrong" and to the operator as nothing at all. + */ + isCredentialUnavailable?: (error: unknown) => boolean; } export class TelegramIngressVerificationError extends Error { @@ -47,10 +56,13 @@ export class TelegramBotAdapter implements ChannelAdapter { readonly #resolveWebhookSecret: TelegramWebhookSecretResolver; readonly #outboundTransport?: TelegramBotOutboundTransport; + readonly #isCredentialUnavailable: (error: unknown) => boolean; constructor(dependencies: TelegramBotAdapterDependencies) { this.#resolveWebhookSecret = dependencies.resolveWebhookSecret; this.#outboundTransport = dependencies.outboundTransport; + this.#isCredentialUnavailable = + dependencies.isCredentialUnavailable ?? (() => false); } async verifyAndNormalizeIngress(input: ProviderIngress) { @@ -60,8 +72,9 @@ export class TelegramBotAdapter implements ChannelAdapter { } catch (error) { // A key this process was started without cannot be reported as a failed // signature check: the sender is probably legitimate and the fix is an - // operator's, not a retry's. - if (error instanceof ChannelCredentialKeyError) throw error; + // operator's, not a retry's. The host recognises its own error and the + // original instance is rethrown, so the route can answer 503. + if (this.#isCredentialUnavailable(error)) throw error; throw new TelegramIngressVerificationError(); } const presented = headerValue(input.headers, SECRET_HEADER); diff --git a/apps/api/src/channel-spine/providers/telegram-bot/api.test.ts b/packages/adapter-telegram/src/api.test.ts similarity index 100% rename from apps/api/src/channel-spine/providers/telegram-bot/api.test.ts rename to packages/adapter-telegram/src/api.test.ts diff --git a/apps/api/src/channel-spine/providers/telegram-bot/api.ts b/packages/adapter-telegram/src/api.ts similarity index 100% rename from apps/api/src/channel-spine/providers/telegram-bot/api.ts rename to packages/adapter-telegram/src/api.ts diff --git a/packages/adapter-telegram/src/index.ts b/packages/adapter-telegram/src/index.ts new file mode 100644 index 00000000..41ca33f0 --- /dev/null +++ b/packages/adapter-telegram/src/index.ts @@ -0,0 +1,10 @@ +// Telegram Bot API channel adapter. +// +// Bot API knowledge only: webhook verification, update normalization, outbound +// method mapping, and capabilities. Credential storage, tenant database +// access, and media storage are injected by the host application. + +export * from "./adapter.js"; +export * from "./api.js"; +export * from "./normalize.js"; +export * from "./transport.js"; diff --git a/apps/api/src/channel-spine/providers/telegram-bot/normalize.test.ts b/packages/adapter-telegram/src/normalize.test.ts similarity index 100% rename from apps/api/src/channel-spine/providers/telegram-bot/normalize.test.ts rename to packages/adapter-telegram/src/normalize.test.ts diff --git a/apps/api/src/channel-spine/providers/telegram-bot/normalize.ts b/packages/adapter-telegram/src/normalize.ts similarity index 100% rename from apps/api/src/channel-spine/providers/telegram-bot/normalize.ts rename to packages/adapter-telegram/src/normalize.ts diff --git a/apps/api/src/channel-spine/providers/telegram-bot/transport.test.ts b/packages/adapter-telegram/src/transport.test.ts similarity index 83% rename from apps/api/src/channel-spine/providers/telegram-bot/transport.test.ts rename to packages/adapter-telegram/src/transport.test.ts index 3fa44b74..6a83f9c7 100644 --- a/apps/api/src/channel-spine/providers/telegram-bot/transport.test.ts +++ b/packages/adapter-telegram/src/transport.test.ts @@ -1,6 +1,8 @@ import { describe, expect, test } from "bun:test"; -import { ChannelCredentialKeyError } from "../../../services/channel-credential.service"; -import { classifyTelegramSendFailure } from "./transport"; +import { + classifyTelegramSendFailure, + TelegramLocalFailureError, +} from "./transport"; describe("Telegram send outcome classification", () => { test("fails a missing credential key outright instead of calling it unknown", () => { @@ -11,9 +13,7 @@ describe("Telegram send outcome classification", () => { // operator-fixable fault behind the label for "we could not tell". expect( classifyTelegramSendFailure( - new ChannelCredentialKeyError( - "active channel credential key is unavailable", - ), + new TelegramLocalFailureError("telegram_credential_key_unavailable"), ), ).toEqual({ outcome: "permanent_failure", diff --git a/apps/api/src/channel-spine/providers/telegram-bot/transport.ts b/packages/adapter-telegram/src/transport.ts similarity index 66% rename from apps/api/src/channel-spine/providers/telegram-bot/transport.ts rename to packages/adapter-telegram/src/transport.ts index 387e3c3e..81191e84 100644 --- a/apps/api/src/channel-spine/providers/telegram-bot/transport.ts +++ b/packages/adapter-telegram/src/transport.ts @@ -4,15 +4,6 @@ import type { ProviderActionResult, ProviderSendResult, } from "@wateaminbox/shared"; -import { - getPresignedUrl, - resolveMediaKeyForCompany, -} from "../../../lib/storage.js"; -import { - ChannelCredentialKeyError, - readChannelCredential, -} from "../../../services/channel-credential.service.js"; -import { getTenantConnection } from "../../../services/tenant.service.js"; import type { TelegramBotOutboundTransport } from "./adapter.js"; import { telegramBotRequest } from "./api.js"; @@ -20,10 +11,97 @@ interface TelegramMessageResult { message_id: number; } +/** + * Failures raised before Telegram is contacted at all. + * + * These are the only outcomes that may be classified permanent without + * hearing from the provider, so the set is closed and the host's port + * implementations raise from it rather than inventing their own codes. + */ +export const TELEGRAM_LOCAL_FAILURE_CODES = [ + "telegram_credential_unavailable", + // A key the process was started without is a configuration fault: the stored + // credential is intact and Telegram is fine, so the send certainly never + // happened. Reporting it as an unknown outcome parks the intent for ever - + // uncertain outcomes are deliberately never retried, because Telegram has no + // idempotency key - and hides an operator-fixable problem behind the label + // reserved for "we could not tell what happened". + "telegram_credential_key_unavailable", + "telegram_conversation_unavailable", + "telegram_conversation_invalid", + "telegram_attachment_missing", +] as const; + +export type TelegramLocalFailureCode = + (typeof TELEGRAM_LOCAL_FAILURE_CODES)[number]; + +export class TelegramLocalFailureError extends Error { + constructor(readonly code: TelegramLocalFailureCode) { + super(code); + this.name = "TelegramLocalFailureError"; + } +} + +/** Where one conversation's outbound traffic goes, resolved by the host. */ +export interface TelegramOutboundContext { + token: string; + chatId: string; + messageThreadId?: number; + externalThreadId: string; +} + +export interface TelegramOutboundTarget { + companyId: string; + channelAccountId: string; + conversationId: string; +} + +/** + * The host-owned half of outbound sending. + * + * Credential decryption, tenant database access, and media storage stay with + * the application; this package keeps only Bot API knowledge. + */ +export interface TelegramTransportPorts { + resolveOutboundContext( + target: TelegramOutboundTarget, + ): Promise; + resolveAttachmentUrl( + companyId: string, + attachment: Record | undefined, + ): Promise; +} + +/** + * Split an `external_thread_id` into its Bot API target. + * + * The encoding is Telegram's own, so it is parsed here rather than by the + * host that happens to store the column. + */ +export function parseTelegramThreadTarget(externalThreadId: string): { + chatId: string; + messageThreadId?: number; +} { + const match = /^(-?\d+)(?::thread:(\d+))?$/.exec(externalThreadId); + if (!match) { + throw new TelegramLocalFailureError("telegram_conversation_invalid"); + } + return { + chatId: match[1]!, + messageThreadId: match[2] ? Number(match[2]) : undefined, + }; +} + export class TelegramBotApiTransport implements TelegramBotOutboundTransport { + readonly #ports: TelegramTransportPorts; + + constructor(ports: TelegramTransportPorts) { + this.#ports = ports; + } + async send(intent: OutboundMessageIntent): Promise { try { - const context = await resolveContext(intent); + const context = await this.#ports.resolveOutboundContext(intent); const payload = intent.normalizedPayload; const messageType = stringValue(payload.messageType) ?? "text"; const text = stringValue(payload.textContent) ?? ""; @@ -41,27 +119,42 @@ export class TelegramBotApiTransport implements TelegramBotOutboundTransport { break; case "image": method = "sendPhoto"; - request.photo = await attachmentUrl(intent.companyId, attachment); + request.photo = await this.#attachmentUrl( + intent.companyId, + attachment, + ); request.caption = text || undefined; break; case "video": method = "sendVideo"; - request.video = await attachmentUrl(intent.companyId, attachment); + request.video = await this.#attachmentUrl( + intent.companyId, + attachment, + ); request.caption = text || undefined; break; case "audio": method = "sendAudio"; - request.audio = await attachmentUrl(intent.companyId, attachment); + request.audio = await this.#attachmentUrl( + intent.companyId, + attachment, + ); request.caption = text || undefined; break; case "voice": method = "sendVoice"; - request.voice = await attachmentUrl(intent.companyId, attachment); + request.voice = await this.#attachmentUrl( + intent.companyId, + attachment, + ); request.caption = text || undefined; break; case "document": method = "sendDocument"; - request.document = await attachmentUrl(intent.companyId, attachment); + request.document = await this.#attachmentUrl( + intent.companyId, + attachment, + ); request.caption = text || undefined; break; default: @@ -87,7 +180,7 @@ export class TelegramBotApiTransport implements TelegramBotOutboundTransport { async perform(action: ChannelActionIntent): Promise { try { - const context = await resolveContext(action); + const context = await this.#ports.resolveOutboundContext(action); const messageId = numberValue(action.payload.externalMessageId); if (!messageId) { return { @@ -139,60 +232,16 @@ export class TelegramBotApiTransport implements TelegramBotOutboundTransport { }; } } -} -async function resolveContext(intent: { - companyId: string; - channelAccountId: string; - conversationId: string; -}): Promise<{ - token: string; - chatId: string; - messageThreadId?: number; - externalThreadId: string; -}> { - const tenantDb = await getTenantConnection(intent.companyId); - const [token, conversation] = await Promise.all([ - readChannelCredential( - tenantDb, - intent.companyId, - intent.channelAccountId, - "telegram_bot_token", - ), - tenantDb - .selectFrom("conversations as conversation") - .innerJoin( - "channel_accounts as account", - "account.id", - "conversation.channel_account_id", - ) - .select([ - "conversation.external_thread_id", - "account.status as account_status", - ]) - .where("conversation.id", "=", intent.conversationId) - .where("conversation.channel_account_id", "=", intent.channelAccountId) - .where("conversation.archived_at", "is", null) - .where("account.archived_at", "is", null) - .executeTakeFirst(), - ]); - if (!token) throw new Error("telegram_credential_unavailable"); - if ( - !conversation?.external_thread_id || - conversation.account_status !== "connected" - ) { - throw new Error("telegram_conversation_unavailable"); + async #attachmentUrl( + companyId: string, + attachment?: Record, + ): Promise { + if (!attachment) { + throw new TelegramLocalFailureError("telegram_attachment_missing"); + } + return this.#ports.resolveAttachmentUrl(companyId, attachment); } - const match = /^(-?\d+)(?::thread:(\d+))?$/.exec( - conversation.external_thread_id, - ); - if (!match) throw new Error("telegram_conversation_invalid"); - return { - token, - chatId: match[1]!, - messageThreadId: match[2] ? Number(match[2]) : undefined, - externalThreadId: conversation.external_thread_id, - }; } export function classifyTelegramSendFailure( @@ -201,26 +250,8 @@ export function classifyTelegramSendFailure( ProviderSendResult, { outcome: "transient_failure" | "permanent_failure" | "uncertain" } > { - // A key the process was started without is a configuration fault: the stored - // credential is intact and Telegram is fine, so the send certainly never - // happened. Reporting it as an unknown outcome parks the intent for ever - - // uncertain outcomes are deliberately never retried, because Telegram has no - // idempotency key - and hides an operator-fixable problem behind the label - // reserved for "we could not tell what happened". - if (error instanceof ChannelCredentialKeyError) { - return { - outcome: "permanent_failure", - errorCode: "telegram_credential_key_unavailable", - }; - } const message = error instanceof Error ? error.message : ""; - const localFailureCodes = new Set([ - "telegram_credential_unavailable", - "telegram_conversation_unavailable", - "telegram_conversation_invalid", - "telegram_attachment_missing", - ]); - if (localFailureCodes.has(message)) { + if ((TELEGRAM_LOCAL_FAILURE_CODES as readonly string[]).includes(message)) { return { outcome: "permanent_failure", errorCode: message }; } if (message === "Telegram Bot API rejected the request") { @@ -248,16 +279,6 @@ function firstAttachment(value: unknown): Record | undefined { : undefined; } -async function attachmentUrl( - companyId: string, - attachment?: Record, -): Promise { - const storageUri = attachment && stringValue(attachment.storageUri); - if (!storageUri) throw new Error("telegram_attachment_missing"); - const key = resolveMediaKeyForCompany(storageUri, companyId); - return getPresignedUrl(key, 10 * 60); -} - function stringValue(value: unknown): string | undefined { return typeof value === "string" ? value : undefined; } diff --git a/packages/adapter-telegram/tsconfig.json b/packages/adapter-telegram/tsconfig.json new file mode 100644 index 00000000..c82dcf13 --- /dev/null +++ b/packages/adapter-telegram/tsconfig.json @@ -0,0 +1,25 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "ESNext", + "moduleResolution": "bundler", + "lib": ["ES2022", "DOM"], + "declaration": true, + "declarationMap": true, + "sourceMap": true, + "strict": true, + "noImplicitAny": true, + "strictNullChecks": true, + "noUnusedLocals": true, + "noUnusedParameters": true, + "noFallthroughCasesInSwitch": true, + "esModuleInterop": true, + "skipLibCheck": true, + "forceConsistentCasingInFileNames": true, + "outDir": "./dist", + "rootDir": "./src", + "types": ["bun-types"] + }, + "include": ["src/**/*"], + "exclude": ["node_modules", "dist"] +} diff --git a/scripts/run-integration-tests-ts.sh b/scripts/run-integration-tests-ts.sh index c4bab763..96ac0f43 100755 --- a/scripts/run-integration-tests-ts.sh +++ b/scripts/run-integration-tests-ts.sh @@ -37,6 +37,11 @@ export DATABASE_URL="${DATABASE_URL:-postgresql://postgres:postgres@localhost:54 # validation and are never used for signing or authentication in tests. export JWT_SECRET="${JWT_SECRET:-integration-test-only-not-a-secret-at-least-32-chars}" export CENTRIFUGO_TOKEN_HMAC_SECRET="${CENTRIFUGO_TOKEN_HMAC_SECRET:-integration-test-only-realtime-not-a-secret-32-chars}" +# Synthetic channel-credential keyring: the provider adapters read encrypted +# credentials through the configured cipher, so suites covering them need a +# well-formed ring. Fixed, non-secret, and never used outside tests. +export CHANNEL_CREDENTIAL_ENCRYPTION_KEYS="${CHANNEL_CREDENTIAL_ENCRYPTION_KEYS:-v1:BwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwc=}" +export CHANNEL_CREDENTIAL_ACTIVE_KEY_VERSION="${CHANNEL_CREDENTIAL_ACTIVE_KEY_VERSION:-v1}" # Run each file in its own Bun process. Several integration suites intentionally # cache membership/configuration at module scope; combining every file in one diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index de7f88e8..e0035e21 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -13,6 +13,7 @@ run() { run "API unit tests" bash -c "cd '$ROOT/apps/api' && bun test" run "Web unit tests" bash -c "cd '$ROOT/apps/web' && bun test" run "Shared TypeScript unit tests" bash -c "cd '$ROOT/packages/shared' && bun test" +run "Telegram adapter unit tests" bash -c "cd '$ROOT/packages/adapter-telegram' && bun test" run "Shared Go tests" bash -c "cd '$ROOT/services/shared' && go test -short -timeout 5m ./..." run "Orchestrator Go tests" bash -c "cd '$ROOT/services/orchestrator' && go test -short -timeout 5m ./..." run "WhatsApp Go tests" bash -c "cd '$ROOT/services/whatsapp' && go test -short -timeout 5m ./..."