diff --git a/.github/workflows/telegram-e2e.yml b/.github/workflows/telegram-e2e.yml new file mode 100644 index 0000000..d366f69 --- /dev/null +++ b/.github/workflows/telegram-e2e.yml @@ -0,0 +1,57 @@ +name: Telegram Live E2E + +on: + workflow_dispatch: + inputs: + delete_webhook: + description: Delete an existing webhook before polling + required: true + type: boolean + default: false + drop_pending_updates: + description: Drop queued updates when deleting the webhook + required: true + type: boolean + default: false + +concurrency: + group: telegram-live-e2e + cancel-in-progress: false + +permissions: + contents: read + +jobs: + telegram-e2e: + name: Telegram live long-polling E2E + runs-on: ubuntu-latest + environment: telegram-e2e + timeout-minutes: 5 + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version-file: go.mod + + - name: Validate live E2E secrets + shell: bash + env: + TELEGRAM_BOT_TOKEN: ${{ secrets.TELEGRAM_BOT_TOKEN }} + TELEGRAM_SENDER_BOT_TOKEN: ${{ secrets.TELEGRAM_SENDER_BOT_TOKEN }} + run: | + test -n "$TELEGRAM_BOT_TOKEN" || { echo "::error::telegram-e2e Environment is missing TELEGRAM_BOT_TOKEN"; exit 1; } + test -n "$TELEGRAM_SENDER_BOT_TOKEN" || { echo "::error::telegram-e2e Environment is missing TELEGRAM_SENDER_BOT_TOKEN"; exit 1; } + + - name: Run live Telegram E2E + run: go run ./examples/telegram-e2e + env: + TELEGRAM_BOT_TOKEN: ${{ secrets.TELEGRAM_BOT_TOKEN }} + TELEGRAM_SENDER_BOT_TOKEN: ${{ secrets.TELEGRAM_SENDER_BOT_TOKEN }} + TELEGRAM_DELETE_WEBHOOK: ${{ inputs.delete_webhook }} + TELEGRAM_DROP_PENDING_UPDATES: ${{ inputs.drop_pending_updates }} + TELEGRAM_TEST_MESSAGE: telegram-e2e-${{ github.run_id }}-${{ github.run_attempt }} + TELEGRAM_TIMEOUT: 90s + TELEGRAM_POLL_TIMEOUT: 5s diff --git a/README.md b/README.md index 59de3a8..2b7509a 100644 --- a/README.md +++ b/README.md @@ -174,6 +174,7 @@ - [ ] 实现 Runner Event 到文本、流式消息和卡片消息的转换 - [ ] 接入企业微信或微信相关通道 - [x] 接入 Telegram long polling 文本通道（Issue #31；单 Binding、Gateway Dispatch、进程内幂等） +- [x] 增加真实 Telegram live E2E 示例与手动 CI workflow（Issue #33；根目录 `examples/telegram-e2e`） - [ ] 接入 Telegram webhook、媒体/rich update 或其他 IM 通道 - [ ] 实现 webhook 验签、账号与租户绑定、用户身份映射 - [ ] 使用 `tenant + channel + message_id` 实现幂等去重和缓存回复 @@ -232,6 +233,9 @@ - Issue #31 的 `trpcservice/channels/telegram` 提供单 Binding、`getMe` 身份校验、普通文本 long polling、Gateway Dispatch、进程内幂等和脱敏分段回复；具体边界以 `docs/docs/telegram.md` 为准。 +- Issue #33 的 `examples/telegram-e2e` 使用真实 Telegram Bot API 和确定性 Dispatcher 验证 + `getMe -> getUpdates -> sendMessage`；live workflow 只手动触发并使用受保护 Environment， + 不替代完整模型供应商或生产控制面 E2E。 - Issue #26 的 fake candidate resolver/verifier 与 proof-bearing routing 边界有独立测试， 但这不代表 WeCom/Telegram webhook、媒体能力或持久化消息能力已满足 README 原验收要求。 @@ -253,6 +257,8 @@ | |-- start.sh # 启动服务 | `-- stop.sh # 停止服务 |-- data # 服务运行时数据 +|-- examples # 可运行的外部集成示例 +| `-- telegram-e2e # Telegram live long-polling E2E |-- docs # 各模块说明与架构设计文档 |-- cmd | `-- trpc-service # 命令行入口，可直接启动服务 diff --git a/codecov.yml b/codecov.yml index ed7826e..036772c 100644 --- a/codecov.yml +++ b/codecov.yml @@ -1,6 +1,11 @@ # Codecov status checks (shown on PRs). # The hard CI gate lives in scripts/coverage.sh --min (see .github/workflows/ci.yml); # this file makes the same 85% target visible in Codecov reports and PR comments. +# The live example entrypoint is exercised only by the opt-in Telegram workflow; +# its credential-free unit tests cover the pure helpers and boundary setup. +ignore: + - "examples/telegram-e2e/main.go" + coverage: status: project: diff --git a/docs/docs/index.md b/docs/docs/index.md index 6017a9b..93ff0aa 100644 --- a/docs/docs/index.md +++ b/docs/docs/index.md @@ -22,6 +22,8 @@ 限流、幂等和服务生命周期契约。 - [Telegram 长轮询 Adapter](telegram.md)：Issue #31 的文档先行契约，固定单 Binding、Bot 身份校验、普通文本映射、Dispatch 聚合回复和生命周期边界。 +- [Telegram live E2E 示例](https://github.com/XnLemon/trpc-agent-service/tree/main/examples/telegram-e2e)： + Issue #33 的真实 Bot API 传输冒烟测试和手动 CI 运行说明。 ## 快速开始 diff --git a/docs/docs/telegram.md b/docs/docs/telegram.md index 548e4c2..e8c05b2 100644 --- a/docs/docs/telegram.md +++ b/docs/docs/telegram.md @@ -148,3 +148,20 @@ README 和 MkDocs 状态应明确区分已交付与后续能力： 参考：[Telegram Bot API](https://core.telegram.org/bots/api)、 [getUpdates](https://core.telegram.org/bots/api#getting-updates)、 [github.com/go-telegram/bot](https://github.com/go-telegram/bot)。 + +## 7. 真实 Telegram E2E + +Issue #33 提供根目录 `examples/telegram-e2e/` 示例和手动触发的 CI 工作流， +用于验证真实的 `getMe -> getUpdates -> sendMessage` 边界。示例内部使用确定性 +`DispatchService`，因此不会把模型供应商凭据和 Telegram 传输冒烟测试混在一起。 + +本地运行只需要在进程环境中提供 `TELEGRAM_BOT_TOKEN`；Token 不得进入仓库、日志、 +trace 或错误。CI 使用受保护的 `telegram-e2e` Environment，至少配置接收 Bot 的 +`TELEGRAM_BOT_TOKEN`，并在需要完全自动化入站消息时配置第二个受控测试 Bot 的 +`TELEGRAM_SENDER_BOT_TOKEN`。一个 Bot Token 不能模拟普通用户向自己发送入站消息， +所以当前 workflow 必须显式配置第二个受控测试 Bot；本地人工运行可以不配置发送者。 + +示例和 CI 都只验证普通文本；命令、媒体、rich update、Webhook、持久化 outbox 和 +生产模型供应商仍不属于该 E2E 范围。详见 +[Telegram live E2E example](https://github.com/XnLemon/trpc-agent-service/tree/main/examples/telegram-e2e) +和 Issue #33。 diff --git a/examples/telegram-e2e/README.md b/examples/telegram-e2e/README.md new file mode 100644 index 0000000..c78e29c --- /dev/null +++ b/examples/telegram-e2e/README.md @@ -0,0 +1,73 @@ +# Telegram live E2E example + +This example exercises the real Telegram Bot API boundary around the +tenant-scoped long-polling adapter: + +```text +getMe -> getUpdates -> trusted Telegram Adapter -> DispatchService -> sendMessage +``` + +It deliberately uses a deterministic `DispatchService`, so this example tests +Telegram transport, trusted target construction, update normalization, reply +delivery, cancellation, and secret handling without requiring a production +LLM provider. + +## Local run + +Create a dedicated test Bot with `@BotFather`, revoke any token that has been +shared outside a secret store, and set the replacement token in the process +environment. Do not place it in this repository or print it. + +PowerShell: + +```powershell +$env:TELEGRAM_BOT_TOKEN = '' +go run ./examples/telegram-e2e +``` + +The command prints a unique ordinary-text marker. Open the receiver Bot in +Telegram, send that marker, and confirm the `telegram-e2e-ok:` reply. Commands, +media, and rich updates are intentionally outside this first E2E. Press +`Ctrl+C` to stop the local polling process cleanly. + +If PowerShell can reach `api.telegram.org` but this command reports +`telegram E2E getMe network failure`, Go is using a different HTTPS path. Go +uses the standard `HTTPS_PROXY`/`HTTP_PROXY` environment variables; it does not +automatically import every Windows system-proxy setting. Configure the proxy +for the same PowerShell process, without printing credentials, and rerun the +command. + +If the Bot has a webhook, either remove it before starting long polling or set +`TELEGRAM_DELETE_WEBHOOK=true`. Pending updates are preserved by default; set +`TELEGRAM_DROP_PENDING_UPDATES=true` only when discarding them is intentional. + +Optional local settings: + +| Variable | Meaning | Default | +| --- | --- | --- | +| `TELEGRAM_TEST_MESSAGE` | Exact marker to wait for | generated per run | +| `TELEGRAM_TIMEOUT` | Maximum run duration | `2m` | +| `TELEGRAM_POLL_TIMEOUT` | Telegram long-poll timeout | `5s` | +| `TELEGRAM_DELETE_WEBHOOK` | Delete an existing webhook | `false` | +| `TELEGRAM_DROP_PENDING_UPDATES` | Drop queued updates when deleting webhook | `false` | + +## CI run + +The live workflow is intentionally manual and references a protected GitHub +Environment named `telegram-e2e`: + +- `TELEGRAM_BOT_TOKEN`: required secret for the receiving test Bot. +- `TELEGRAM_SENDER_BOT_TOKEN`: required secret for a second controlled test Bot + in CI; it sends the unique marker and receives the expected reply. This + sender secret is optional only for local human-driven runs. + +For a fully automatic message round trip, enable Telegram Bot-to-Bot +Communication Mode for both dedicated test Bots. A single Bot API token cannot +act as a normal user sending an inbound message to itself. Without the sender +secret, the example remains suitable for a local human-driven run but the CI +job will eventually time out waiting for the marker. + +The workflow uses one concurrency group so two runs cannot poll the same test +Bot at the same time. It must not be changed to run automatically on arbitrary +pull requests: live Bot credentials and external Telegram side effects are +intentionally outside the offline PR checks. diff --git a/examples/telegram-e2e/main.go b/examples/telegram-e2e/main.go new file mode 100644 index 0000000..606557a --- /dev/null +++ b/examples/telegram-e2e/main.go @@ -0,0 +1,570 @@ +package main + +import ( + "context" + cryptorand "crypto/rand" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os" + "os/signal" + "strconv" + "strings" + "sync" + "syscall" + "time" + + "github.com/XnLemon/trpc-agent-service/trpcservice/agent" + "github.com/XnLemon/trpc-agent-service/trpcservice/channels" + channelsinmemory "github.com/XnLemon/trpc-agent-service/trpcservice/channels/inmemory" + "github.com/XnLemon/trpc-agent-service/trpcservice/channels/telegram" + "github.com/XnLemon/trpc-agent-service/trpcservice/gateway" + "github.com/XnLemon/trpc-agent-service/trpcservice/tenant" + "github.com/go-telegram/bot" + "github.com/go-telegram/bot/models" +) + +const ( + defaultRunTimeout = 2 * time.Minute + defaultPollTimeout = 5 * time.Second + shutdownTimeout = 5 * time.Second + e2eReply = "telegram-e2e-ok" +) + +var ( + errConfiguration = errors.New("invalid Telegram E2E configuration") + errPreflight = errors.New("telegram E2E preflight failed") + errPreflightClient = errors.New("telegram E2E bot client preflight failed") + errPreflightGetMeNetwork = errors.New("telegram E2E getMe network failure") + errPreflightGetMeTimeout = errors.New("telegram E2E getMe timeout") + errPreflightGetMeAPI = errors.New("telegram E2E getMe Telegram API rejected the request") + errPreflightGetMeReply = errors.New("telegram E2E getMe response was invalid") + errPreflightWebhook = errors.New("telegram E2E webhook preflight failed") + errWebhookConfigured = errors.New("telegram webhook is configured; remove it or enable TELEGRAM_DELETE_WEBHOOK") + errAdapterConfiguration = errors.New("telegram E2E adapter configuration failed") + errAdapterInitialization = errors.New("telegram E2E adapter initialization failed") + errAdapterIdentity = errors.New("telegram E2E adapter identity check failed") + errAdapterRun = errors.New("telegram E2E adapter stopped unexpectedly") + errAdapterClose = errors.New("telegram E2E adapter close failed") + errRunTimeout = errors.New("telegram E2E timed out waiting for the test message") + errSender = errors.New("telegram E2E sender failed") + errSenderStopped = errors.New("telegram E2E sender stopped unexpectedly") +) + +type runConfig struct { + botToken string + senderBotToken string + testMessage string + runTimeout time.Duration + pollTimeout time.Duration + deleteWebhook bool + dropPendingUpdate bool +} + +type webhookClient interface { + GetWebhookInfo(context.Context) (*models.WebhookInfo, error) + DeleteWebhook(context.Context, *bot.DeleteWebhookParams) (bool, error) +} + +type prepareBotFunc func(context.Context, string, time.Duration, bool, bool) (*models.User, error) + +type deterministicDispatcher struct { + marker string + reply string + seen chan gateway.InboundMessage + once sync.Once +} + +var _ gateway.DispatchService = (*deterministicDispatcher)(nil) + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + if err := run(ctx, os.Getenv, os.Stdout, os.Stderr); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func run(ctx context.Context, lookup func(string) string, stdout, stderr io.Writer) error { + return runWithPreflight(ctx, lookup, stdout, stderr, prepareBot) +} + +func runWithPreflight(ctx context.Context, lookup func(string) string, stdout, stderr io.Writer, prepare prepareBotFunc) error { + if ctx == nil || lookup == nil || stdout == nil || stderr == nil { + return errConfiguration + } + if prepare == nil { + return errConfiguration + } + configuration, err := loadConfig(lookup) + if err != nil { + return err + } + runContext, cancel := context.WithTimeout(ctx, configuration.runTimeout) + defer cancel() + correlationID, err := newCorrelationID() + if err != nil { + return errConfiguration + } + reply := e2eReplyFor(correlationID) + + receiver, err := prepare(runContext, configuration.botToken, configuration.pollTimeout, configuration.deleteWebhook, configuration.dropPendingUpdate) + if err != nil { + return classifyPreflightResult(ctx.Err(), runContext.Err(), err) + } + target, err := newTrustedTarget(strconv.FormatInt(receiver.ID, 10)) + if err != nil { + return errConfiguration + } + dispatcher := newDeterministicDispatcher(configuration.testMessage, reply) + adapter, err := telegramAdapter(runContext, configuration, target, dispatcher, stderr) + if err != nil { + return classifyPreflightResult(ctx.Err(), runContext.Err(), err) + } + + runDone := make(chan error, 1) + go func() { + runDone <- adapter.Run(runContext) + }() + + _, _ = fmt.Fprintf(stdout, "Telegram E2E receiver @%s (%d) is listening.\n", receiver.Username, receiver.ID) + _, _ = fmt.Fprintf(stdout, "Send this ordinary text: %s\n", configuration.testMessage) + + var result error + if configuration.senderBotToken != "" { + senderResult := runAutomatedSender(runContext, configuration.senderBotToken, configuration.pollTimeout, receiver, configuration.testMessage, reply, configuration.deleteWebhook, configuration.dropPendingUpdate) + result = classifyAutomatedSenderResult(ctx.Err(), runContext.Err(), senderResult) + cancel() + if stopErr := waitForAdapter(runDone); stopErr != nil && result == nil { + result = stopErr + } + } else { + select { + case err := <-runDone: + result = classifyManualRunResult(ctx.Err(), runContext.Err(), err) + case <-runContext.Done(): + if stopErr := waitForAdapter(runDone); stopErr != nil { + result = stopErr + } else { + result = classifyManualRunResult(ctx.Err(), runContext.Err(), nil) + } + } + } + cancel() + if err := adapter.Close(); err != nil && result == nil { + result = errAdapterClose + } + return result +} + +func classifyPreflightResult(parentErr, runContextErr, preflightErr error) error { + if parentErr != nil { + return nil + } + if errors.Is(runContextErr, context.DeadlineExceeded) { + return errRunTimeout + } + return preflightErr +} + +func classifyManualRunResult(parentErr, runContextErr, adapterErr error) error { + if parentErr != nil { + return nil + } + if errors.Is(runContextErr, context.DeadlineExceeded) { + return errRunTimeout + } + if errors.Is(runContextErr, context.Canceled) || errors.Is(adapterErr, context.Canceled) || errors.Is(adapterErr, context.DeadlineExceeded) { + return nil + } + return errAdapterRun +} + +func classifyAutomatedSenderResult(parentErr, runContextErr, senderErr error) error { + if parentErr != nil { + return nil + } + if errors.Is(runContextErr, context.DeadlineExceeded) { + return errRunTimeout + } + if errors.Is(runContextErr, context.Canceled) { + return nil + } + return senderErr +} + +func loadConfig(lookup func(string) string) (runConfig, error) { + if lookup == nil { + return runConfig{}, errConfiguration + } + botToken := strings.TrimSpace(lookup("TELEGRAM_BOT_TOKEN")) + if botToken == "" || hasControl(botToken) { + return runConfig{}, errConfiguration + } + senderToken := strings.TrimSpace(lookup("TELEGRAM_SENDER_BOT_TOKEN")) + if hasControl(senderToken) || (senderToken != "" && senderToken == botToken) { + return runConfig{}, errConfiguration + } + message := strings.TrimSpace(lookup("TELEGRAM_TEST_MESSAGE")) + if message == "" { + message = fmt.Sprintf("telegram-e2e-%d", time.Now().UTC().UnixNano()) + } + if hasControl(message) || len([]rune(message)) > 4096 || strings.Contains(message, botToken) || (senderToken != "" && strings.Contains(message, senderToken)) { + return runConfig{}, errConfiguration + } + runTimeout, err := readDuration(lookup, "TELEGRAM_TIMEOUT", defaultRunTimeout) + if err != nil || runTimeout <= 0 { + return runConfig{}, errConfiguration + } + pollTimeout, err := readDuration(lookup, "TELEGRAM_POLL_TIMEOUT", defaultPollTimeout) + if err != nil || pollTimeout < 2*time.Second || pollTimeout > 10*time.Minute { + return runConfig{}, errConfiguration + } + deleteWebhook, err := readBool(lookup, "TELEGRAM_DELETE_WEBHOOK", false) + if err != nil { + return runConfig{}, errConfiguration + } + dropPending, err := readBool(lookup, "TELEGRAM_DROP_PENDING_UPDATES", false) + if err != nil { + return runConfig{}, errConfiguration + } + return runConfig{ + botToken: botToken, senderBotToken: senderToken, testMessage: message, + runTimeout: runTimeout, pollTimeout: pollTimeout, + deleteWebhook: deleteWebhook, dropPendingUpdate: dropPending, + }, nil +} + +func readDuration(lookup func(string) string, name string, fallback time.Duration) (time.Duration, error) { + value := strings.TrimSpace(lookup(name)) + if value == "" { + return fallback, nil + } + return time.ParseDuration(value) +} + +func readBool(lookup func(string) string, name string, fallback bool) (bool, error) { + value := strings.TrimSpace(lookup(name)) + if value == "" { + return fallback, nil + } + return strconv.ParseBool(value) +} + +func prepareBot(ctx context.Context, token string, pollTimeout time.Duration, deleteWebhook, dropPending bool) (*models.User, error) { + client, err := bot.New(token, bot.WithSkipGetMe(), bot.WithHTTPClient(pollTimeout, preflightHTTPClient(pollTimeout))) + if err != nil { + return nil, errPreflightClient + } + me, err := client.GetMe(ctx) + if err != nil { + return nil, classifyGetMeError(err) + } + if me == nil || !me.IsBot || me.ID <= 0 { + return nil, errPreflightGetMeReply + } + if err := prepareLongPolling(ctx, client, deleteWebhook, dropPending); err != nil { + return nil, err + } + return me, nil +} + +func classifyGetMeError(err error) error { + if err == nil { + return errPreflightGetMeReply + } + if errors.Is(err, context.DeadlineExceeded) { + return errPreflightGetMeTimeout + } + if errors.Is(err, bot.ErrorForbidden) || errors.Is(err, bot.ErrorBadRequest) || errors.Is(err, bot.ErrorUnauthorized) || errors.Is(err, bot.ErrorNotFound) || errors.Is(err, bot.ErrorConflict) || errors.Is(err, bot.ErrorTooManyRequests) { + return errPreflightGetMeAPI + } + var tooManyRequestsError *bot.TooManyRequestsError + if errors.As(err, &tooManyRequestsError) { + return errPreflightGetMeAPI + } + var requestError *url.Error + if errors.As(err, &requestError) { + if requestError.Timeout() { + return errPreflightGetMeTimeout + } + return errPreflightGetMeNetwork + } + return errPreflightGetMeReply +} + +func classifyAdapterError(err error) error { + switch { + case errors.Is(err, telegram.ErrInvalid): + return errAdapterConfiguration + case errors.Is(err, telegram.ErrBotIdentityMismatch): + return errAdapterIdentity + case errors.Is(err, telegram.ErrInitialization): + return errAdapterInitialization + default: + return errPreflight + } +} + +func preflightHTTPClient(pollTimeout time.Duration) *http.Client { + return &http.Client{Timeout: pollTimeout + 5*time.Second} +} + +func newCorrelationID() (string, error) { + var nonce [8]byte + if _, err := cryptorand.Read(nonce[:]); err != nil { + return "", err + } + return hex.EncodeToString(nonce[:]), nil +} + +func e2eReplyFor(correlationID string) string { + return e2eReply + ":" + correlationID +} + +func prepareLongPolling(ctx context.Context, client webhookClient, deleteWebhook, dropPending bool) error { + if client == nil { + return errPreflightWebhook + } + info, err := client.GetWebhookInfo(ctx) + if err != nil { + return errPreflightWebhook + } + if info == nil || info.URL == "" { + return nil + } + if !deleteWebhook { + return errWebhookConfigured + } + if _, err := client.DeleteWebhook(ctx, &bot.DeleteWebhookParams{DropPendingUpdates: dropPending}); err != nil { + return errPreflightWebhook + } + return nil +} + +func newTrustedTarget(providerAccountID string) (channels.RoutingTarget, error) { + accountID, err := strconv.ParseInt(providerAccountID, 10, 64) + if err != nil || accountID <= 0 || strconv.FormatInt(accountID, 10) != providerAccountID { + return channels.RoutingTarget{}, errConfiguration + } + root, err := tenant.NewTenant(tenant.CreateInput{ + TenantKey: "telegram-e2e", DisplayName: "Telegram E2E Tenant", + AuditRetentionDays: 30, LogMaskingLevel: tenant.MaskingStrict, TraceSamplingRate: 1, + }) + if err != nil { + return channels.RoutingTarget{}, errConfiguration + } + snapshot, err := tenant.NewConfigurationSnapshot(root) + if err != nil { + return channels.RoutingTarget{}, errConfiguration + } + app, err := agent.NewApp(agent.CreateInput{ + TenantID: root.TenantID, AppKey: "telegram-e2e", DisplayName: "Telegram E2E", Description: "Deterministic Telegram transport test", + }) + if err != nil { + return channels.RoutingTarget{}, errConfiguration + } + revision := int64(1) + app.Status = agent.StatusActive + app.CurrentRevision = &revision + app.Version++ + app.UpdatedAt = app.CreatedAt.Add(time.Second) + if err := app.Validate(); err != nil { + return channels.RoutingTarget{}, errConfiguration + } + routeDigest, err := channels.DigestPublicRouteKey(channels.ChannelTelegram, "telegram-e2e") + if err != nil { + return channels.RoutingTarget{}, errConfiguration + } + repository := channelsinmemory.NewRepository() + secret := "telegram-e2e-verifier-secret" + binding, _, err := repository.Create(context.Background(), channels.CreateInput{ + TenantID: root.TenantID, BindingKey: "telegram-e2e", Channel: channels.ChannelTelegram, + ProviderAccountID: providerAccountID, PublicRouteKeyDigest: routeDigest, AppID: app.AppID, + SecretRef: "examples/telegram-e2e", Status: channels.StatusActive, + Protocol: channels.ProtocolConfiguration{Telegram: &channels.TelegramProtocolConfiguration{}}, + Metadata: exampleMetadata(), + }) + if err != nil { + return channels.RoutingTarget{}, errConfiguration + } + resolver := channelsinmemory.NewFakeCandidateResolver(repository, map[channels.SecretScope]string{{TenantID: binding.TenantID, SecretRef: binding.SecretRef}: secret}) + candidates, err := repository.LookupCandidates(context.Background(), channels.ChannelTelegram, routeDigest) + if err != nil || len(candidates) != 1 { + return channels.RoutingTarget{}, errConfiguration + } + handle, err := resolver.ResolveCandidate(context.Background(), channels.CandidateSecretRequest{ + Candidate: candidates[0], Purpose: channels.PurposeWebhookVerification, + }) + if err != nil { + return channels.RoutingTarget{}, errConfiguration + } + digest := sha256.Sum256([]byte("telegram-e2e-trusted-target")) + verification := channels.VerificationRequest{ + Purpose: channels.PurposeWebhookVerification, Timestamp: time.Now().UTC(), + Nonce: "telegram-e2e-target", MessageDigest: hex.EncodeToString(digest[:]), + } + verification.Signature = channelsinmemory.SignFakeRequest(secret, verification) + verified, err := resolver.Verify(context.Background(), handle, verification) + if err != nil { + return channels.RoutingTarget{}, errConfiguration + } + target, err := channels.NewRoutingTarget(snapshot, binding, app, verified) + if err != nil { + return channels.RoutingTarget{}, errConfiguration + } + return target, nil +} + +func exampleMetadata() channels.ChangeMetadata { + return channels.ChangeMetadata{ + ActorType: "example", ActorID: "telegram-e2e", Reason: "live Telegram transport test", + CorrelationID: "telegram-e2e", + } +} + +func telegramAdapter(ctx context.Context, configuration runConfig, target channels.RoutingTarget, dispatcher gateway.DispatchService, stderr io.Writer) (*telegram.Adapter, error) { + adapter, err := telegram.New(ctx, telegram.Config{ + BotToken: configuration.botToken, Target: target, Dispatcher: dispatcher, + PollTimeout: configuration.pollTimeout, + ErrorHook: func(event telegram.ErrorEvent) { + _, _ = fmt.Fprintf(stderr, "telegram %s failed: %v\n", event.Operation, event.Err) + }, + }) + if err != nil { + return nil, classifyAdapterError(err) + } + return adapter, nil +} + +func newDeterministicDispatcher(marker, reply string) *deterministicDispatcher { + return &deterministicDispatcher{marker: marker, reply: reply, seen: make(chan gateway.InboundMessage, 1)} +} + +func (dispatcher *deterministicDispatcher) Dispatch(ctx context.Context, request gateway.DispatchRequest) (<-chan gateway.DispatchEvent, error) { + if dispatcher == nil || ctx == nil { + return nil, errConfiguration + } + if err := ctx.Err(); err != nil { + return nil, err + } + if request.Message.Content == dispatcher.marker { + dispatcher.once.Do(func() { + select { + case dispatcher.seen <- request.Message: + default: + } + }) + } + events := make(chan gateway.DispatchEvent, 2) + if request.Message.Content == dispatcher.marker { + events <- gateway.DispatchEvent{Type: gateway.DispatchEventMessage, RequestID: request.RequestID, Text: dispatcher.reply} + } + events <- gateway.DispatchEvent{Type: gateway.DispatchEventDone, RequestID: request.RequestID, Status: "complete", Done: true} + close(events) + return events, nil +} + +func runAutomatedSender(ctx context.Context, token string, pollTimeout time.Duration, receiver *models.User, marker, reply string, deleteWebhook, dropPending bool) error { + if receiver == nil || receiver.ID <= 0 || receiver.Username == "" { + return errSender + } + replyReceived := make(chan struct{}, 1) + pollingFailed := make(chan struct{}, 1) + sender, err := bot.New(token, + bot.WithSkipGetMe(), + bot.WithDefaultHandler(func(_ context.Context, _ *bot.Bot, update *models.Update) { + if update == nil || update.Message == nil || update.Message.From == nil { + return + } + if isExpectedAutomatedReply(update, receiver.ID, reply) { + select { + case replyReceived <- struct{}{}: + default: + } + } + }), + bot.WithNotAsyncHandlers(), + bot.WithErrorsHandler(func(error) { + select { + case pollingFailed <- struct{}{}: + default: + } + }), + bot.WithHTTPClient(pollTimeout, preflightHTTPClient(pollTimeout)), + ) + if err != nil { + return errSender + } + senderUser, err := sender.GetMe(ctx) + if err != nil || senderUser == nil || !senderUser.IsBot || senderUser.ID <= 0 || senderUser.ID == receiver.ID { + return errSender + } + if err := prepareLongPolling(ctx, sender, deleteWebhook, dropPending); err != nil { + return errSender + } + + senderContext, cancel := context.WithCancel(ctx) + senderDone := make(chan struct{}) + go func() { + sender.Start(senderContext) + close(senderDone) + }() + if _, err := sender.SendMessage(ctx, &bot.SendMessageParams{ChatID: "@" + receiver.Username, Text: marker}); err != nil { + cancel() + waitForSender(senderDone) + return errSender + } + select { + case <-replyReceived: + cancel() + if !waitForSender(senderDone) { + return errSenderStopped + } + return nil + case <-pollingFailed: + cancel() + waitForSender(senderDone) + return errSender + case <-senderDone: + cancel() + return errSenderStopped + case <-ctx.Done(): + cancel() + waitForSender(senderDone) + return errSender + } +} + +func isExpectedAutomatedReply(update *models.Update, receiverID int64, reply string) bool { + return update != nil && update.Message != nil && update.Message.From != nil && + update.Message.From.ID == receiverID && update.Message.Chat.ID == receiverID && update.Message.Text == reply +} + +func waitForSender(done <-chan struct{}) bool { + select { + case <-done: + return true + case <-time.After(shutdownTimeout): + return false + } +} + +func waitForAdapter(done <-chan error) error { + if err := <-done; err != nil { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return nil + } + return errAdapterRun + } + return nil +} + +func hasControl(value string) bool { + return strings.ContainsAny(value, "\r\n\x00") +} diff --git a/examples/telegram-e2e/main_test.go b/examples/telegram-e2e/main_test.go new file mode 100644 index 0000000..1d32917 --- /dev/null +++ b/examples/telegram-e2e/main_test.go @@ -0,0 +1,353 @@ +package main + +import ( + "context" + "errors" + "fmt" + "io" + "net/url" + "strings" + "testing" + "time" + + "github.com/XnLemon/trpc-agent-service/trpcservice/channels" + "github.com/XnLemon/trpc-agent-service/trpcservice/channels/telegram" + "github.com/XnLemon/trpc-agent-service/trpcservice/gateway" + "github.com/go-telegram/bot" + "github.com/go-telegram/bot/models" +) + +func TestLoadConfigUsesSafeDefaults(t *testing.T) { + values := map[string]string{"TELEGRAM_BOT_TOKEN": "receiver-token"} + configuration, err := loadConfig(func(name string) string { return values[name] }) + if err != nil { + t.Fatal(err) + } + if configuration.botToken != values["TELEGRAM_BOT_TOKEN"] || configuration.senderBotToken != "" { + t.Fatalf("unexpected token configuration: %+v", configuration) + } + if configuration.testMessage == "" || !strings.HasPrefix(configuration.testMessage, "telegram-e2e-") { + t.Fatalf("generated marker = %q", configuration.testMessage) + } + if configuration.runTimeout != defaultRunTimeout || configuration.pollTimeout != defaultPollTimeout { + t.Fatalf("unexpected defaults: %+v", configuration) + } + if configuration.deleteWebhook || configuration.dropPendingUpdate { + t.Fatalf("destructive webhook defaults must be false: %+v", configuration) + } +} + +func TestLoadConfigRejectsSecretBearingOrUnsafeValuesWithoutEchoingThem(t *testing.T) { + tests := []map[string]string{ + {"TELEGRAM_BOT_TOKEN": "receiver-token", "TELEGRAM_SENDER_BOT_TOKEN": "receiver-token"}, + {"TELEGRAM_BOT_TOKEN": "receiver-token", "TELEGRAM_TEST_MESSAGE": "contains-receiver-token"}, + {"TELEGRAM_BOT_TOKEN": "receiver-token", "TELEGRAM_TIMEOUT": "not-a-duration"}, + {"TELEGRAM_BOT_TOKEN": "receiver-token", "TELEGRAM_POLL_TIMEOUT": "1s"}, + {"TELEGRAM_BOT_TOKEN": "receiver-token", "TELEGRAM_DELETE_WEBHOOK": "not-a-bool"}, + } + for _, values := range tests { + _, err := loadConfig(func(name string) string { return values[name] }) + if !errors.Is(err, errConfiguration) { + t.Fatalf("values=%v error=%v, want errConfiguration", values, err) + } + for _, value := range values { + if value != "" && strings.Contains(err.Error(), value) { + t.Fatalf("error %q echoed configured value %q", err, value) + } + } + } +} + +func TestLoadConfigParsesExplicitSettings(t *testing.T) { + values := map[string]string{ + "TELEGRAM_BOT_TOKEN": "receiver-token", + "TELEGRAM_SENDER_BOT_TOKEN": "sender-token", + "TELEGRAM_TEST_MESSAGE": "telegram-e2e-marker", + "TELEGRAM_TIMEOUT": "45s", + "TELEGRAM_POLL_TIMEOUT": "3s", + "TELEGRAM_DELETE_WEBHOOK": "true", + "TELEGRAM_DROP_PENDING_UPDATES": "true", + } + configuration, err := loadConfig(func(name string) string { return values[name] }) + if err != nil { + t.Fatal(err) + } + if configuration.testMessage != values["TELEGRAM_TEST_MESSAGE"] || configuration.runTimeout != 45*time.Second || configuration.pollTimeout != 3*time.Second || !configuration.deleteWebhook || !configuration.dropPendingUpdate { + t.Fatalf("explicit settings were not parsed: %+v", configuration) + } +} + +func TestRunTimeoutCoversBlockingPreflight(t *testing.T) { + values := map[string]string{ + "TELEGRAM_BOT_TOKEN": "receiver-token", + "TELEGRAM_TIMEOUT": "20ms", + "TELEGRAM_POLL_TIMEOUT": "2s", + } + var observedContextErr error + prepare := func(ctx context.Context, _ string, _ time.Duration, _, _ bool) (*models.User, error) { + <-ctx.Done() + observedContextErr = ctx.Err() + return nil, errPreflightGetMeTimeout + } + + err := runWithPreflight(context.Background(), func(name string) string { return values[name] }, io.Discard, io.Discard, prepare) + if !errors.Is(err, errRunTimeout) { + t.Fatalf("runWithPreflight() error = %v, want run timeout", err) + } + if !errors.Is(observedContextErr, context.DeadlineExceeded) { + t.Fatalf("preflight context error = %v, want deadline exceeded", observedContextErr) + } +} + +func TestRunWithPreflightTreatsParentCancellationAsClean(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + prepare := func(ctx context.Context, _ string, _ time.Duration, _, _ bool) (*models.User, error) { + return nil, ctx.Err() + } + if err := runWithPreflight(ctx, func(name string) string { + if name == "TELEGRAM_BOT_TOKEN" { + return "receiver-token" + } + return "" + }, io.Discard, io.Discard, prepare); err != nil { + t.Fatalf("runWithPreflight() error = %v, want clean cancellation", err) + } +} + +func TestClassifyManualRunResultTreatsCancellationAsClean(t *testing.T) { + tests := []struct { + name string + parentErr error + runContextErr error + adapterErr error + want error + }{ + {name: "parent cancellation", parentErr: context.Canceled, runContextErr: context.Canceled, want: nil}, + {name: "run timeout", runContextErr: context.DeadlineExceeded, want: errRunTimeout}, + {name: "adapter cancellation", adapterErr: context.Canceled, want: nil}, + {name: "unexpected stop", want: errAdapterRun}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := classifyManualRunResult(test.parentErr, test.runContextErr, test.adapterErr); !errors.Is(got, test.want) { + t.Fatalf("classifyManualRunResult() = %v, want %v", got, test.want) + } + }) + } +} + +func TestClassifyAutomatedSenderResultTreatsCancellationAndTimeoutAsClean(t *testing.T) { + tests := []struct { + name string + parentErr error + runContextErr error + senderErr error + want error + }{ + {name: "parent cancellation", parentErr: context.Canceled, runContextErr: context.Canceled, senderErr: errSender, want: nil}, + {name: "run timeout", runContextErr: context.DeadlineExceeded, senderErr: errSender, want: errRunTimeout}, + {name: "run cancellation", runContextErr: context.Canceled, senderErr: errSender, want: nil}, + {name: "sender failure", senderErr: errSender, want: errSender}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := classifyAutomatedSenderResult(test.parentErr, test.runContextErr, test.senderErr); !errors.Is(got, test.want) { + t.Fatalf("classifyAutomatedSenderResult() = %v, want %v", got, test.want) + } + }) + } +} + +func TestPrepareLongPollingHandlesWebhookSafely(t *testing.T) { + noWebhook := &fakeWebhookClient{} + if err := prepareLongPolling(context.Background(), noWebhook, false, false); err != nil { + t.Fatal(err) + } + if noWebhook.deleted { + t.Fatal("did not expect DeleteWebhook without a webhook") + } + + configured := &fakeWebhookClient{info: &models.WebhookInfo{URL: "https://example.test/telegram"}} + if err := prepareLongPolling(context.Background(), configured, false, false); !errors.Is(err, errWebhookConfigured) { + t.Fatalf("configured webhook error = %v", err) + } + if configured.deleted { + t.Fatal("must not delete webhook without explicit permission") + } + + configured = &fakeWebhookClient{info: &models.WebhookInfo{URL: "https://example.test/telegram"}} + if err := prepareLongPolling(context.Background(), configured, true, true); err != nil { + t.Fatal(err) + } + if !configured.deleted || !configured.dropPending { + t.Fatalf("DeleteWebhook options were not preserved: %+v", configured) + } +} + +func TestClassifyGetMeErrorRedactsProviderDetails(t *testing.T) { + secret := "bot-secret" + tests := []struct { + name string + err error + want error + }{ + {name: "nil error", want: errPreflightGetMeReply}, + {name: "context timeout", err: context.DeadlineExceeded, want: errPreflightGetMeTimeout}, + {name: "api rejection", err: fmt.Errorf("provider response: %w", bot.ErrorUnauthorized), want: errPreflightGetMeAPI}, + {name: "rate limit", err: &bot.TooManyRequestsError{Message: "provider response", RetryAfter: 1}, want: errPreflightGetMeAPI}, + {name: "network error", err: &url.Error{Op: "POST", URL: "https://api.telegram.org/bot" + secret + "/getMe", Err: errors.New("dial failed")}, want: errPreflightGetMeNetwork}, + {name: "invalid response", err: errors.New("provider response could not be decoded"), want: errPreflightGetMeReply}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got := classifyGetMeError(test.err) + if !errors.Is(got, test.want) { + t.Fatalf("classifyGetMeError(%v) = %v, want %v", test.err, got, test.want) + } + if strings.Contains(got.Error(), secret) { + t.Fatalf("classification error %q leaked provider secret", got) + } + }) + } +} + +func TestClassifyAdapterErrorRedactsProviderDetails(t *testing.T) { + secret := "bot-secret" + tests := []struct { + name string + err error + want error + }{ + {name: "invalid configuration", err: fmt.Errorf("provider token %s: %w", secret, telegram.ErrInvalid), want: errAdapterConfiguration}, + {name: "identity mismatch", err: telegram.ErrBotIdentityMismatch, want: errAdapterIdentity}, + {name: "initialization", err: fmt.Errorf("provider detail: %w", telegram.ErrInitialization), want: errAdapterInitialization}, + {name: "fallback", err: errors.New("unexpected provider error"), want: errPreflight}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got := classifyAdapterError(test.err) + if !errors.Is(got, test.want) { + t.Fatalf("classifyAdapterError(%v) = %v, want %v", test.err, got, test.want) + } + if strings.Contains(got.Error(), secret) { + t.Fatalf("classification error %q leaked provider secret", got) + } + }) + } +} + +func TestNewTrustedTargetUsesTheTrustedBoundary(t *testing.T) { + target, err := newTrustedTarget("123456789") + if err != nil { + t.Fatal(err) + } + if err := target.Validate(); err != nil { + t.Fatal(err) + } + if target.Channel != channels.ChannelTelegram || target.ProviderAccountID != "123456789" { + t.Fatalf("unexpected target: %+v", target) + } +} + +func TestNewTrustedTargetRejectsNonCanonicalAccountID(t *testing.T) { + for _, value := range []string{"", "0", "+123456789", "0123456789", "bot"} { + if _, err := newTrustedTarget(value); !errors.Is(err, errConfiguration) { + t.Fatalf("provider account %q error = %v", value, err) + } + } +} + +func TestDeterministicDispatcherEmitsCompleteReplyAndMarksInput(t *testing.T) { + reply := e2eReplyFor("correlation") + dispatcher := newDeterministicDispatcher("marker", reply) + stream, err := dispatcher.Dispatch(context.Background(), gateway.DispatchRequest{ + Message: gateway.InboundMessage{Content: "marker"}, RequestID: "request-id", + }) + if err != nil { + t.Fatal(err) + } + var events []gateway.DispatchEvent + for event := range stream { + events = append(events, event) + } + if len(events) != 2 || events[0].Type != gateway.DispatchEventMessage || events[0].Text != reply || !events[1].Done { + t.Fatalf("unexpected dispatch events: %+v", events) + } + select { + case message := <-dispatcher.seen: + if message.Content != "marker" { + t.Fatalf("seen message = %+v", message) + } + default: + t.Fatal("dispatcher did not mark the expected message") + } +} + +func TestDeterministicDispatcherDoesNotReplyToNonMarker(t *testing.T) { + dispatcher := newDeterministicDispatcher("marker", e2eReplyFor("correlation")) + stream, err := dispatcher.Dispatch(context.Background(), gateway.DispatchRequest{ + Message: gateway.InboundMessage{Content: "old-message"}, RequestID: "request-id", + }) + if err != nil { + t.Fatal(err) + } + var events []gateway.DispatchEvent + for event := range stream { + events = append(events, event) + } + if len(events) != 1 || !events[0].Done || events[0].Type != gateway.DispatchEventDone { + t.Fatalf("unexpected non-marker events: %+v", events) + } + select { + case message := <-dispatcher.seen: + t.Fatalf("non-marker was recorded as seen: %+v", message) + default: + } +} + +func TestExpectedAutomatedReplyRequiresCorrelationAndPrivatePeer(t *testing.T) { + receiverID := int64(42) + reply := e2eReplyFor("correlation") + tests := []struct { + name string + update *models.Update + want bool + }{ + {name: "valid reply", update: &models.Update{Message: &models.Message{From: &models.User{ID: receiverID}, Chat: models.Chat{ID: receiverID}, Text: reply}}, want: true}, + {name: "stale correlation", update: &models.Update{Message: &models.Message{From: &models.User{ID: receiverID}, Chat: models.Chat{ID: receiverID}, Text: e2eReplyFor("stale")}}}, + {name: "unexpected chat", update: &models.Update{Message: &models.Message{From: &models.User{ID: receiverID}, Chat: models.Chat{ID: 99}, Text: reply}}}, + {name: "unexpected sender", update: &models.Update{Message: &models.Message{From: &models.User{ID: 99}, Chat: models.Chat{ID: receiverID}, Text: reply}}}, + {name: "missing message", update: &models.Update{}}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := isExpectedAutomatedReply(test.update, receiverID, reply); got != test.want { + t.Fatalf("isExpectedAutomatedReply() = %v, want %v", got, test.want) + } + }) + } +} + +func TestPreflightHTTPClientUsesPollTimeoutBudget(t *testing.T) { + client := preflightHTTPClient(3 * time.Second) + if client.Timeout != 8*time.Second { + t.Fatalf("preflight HTTP timeout = %s, want 8s", client.Timeout) + } +} + +type fakeWebhookClient struct { + info *models.WebhookInfo + deleted bool + dropPending bool +} + +func (client *fakeWebhookClient) GetWebhookInfo(context.Context) (*models.WebhookInfo, error) { + return client.info, nil +} + +func (client *fakeWebhookClient) DeleteWebhook(_ context.Context, params *bot.DeleteWebhookParams) (bool, error) { + client.deleted = true + client.dropPending = params != nil && params.DropPendingUpdates + return true, nil +}