From 3ca1d2c4370e8e79aba3940d5c8fe115960b75a8 Mon Sep 17 00:00:00 2001 From: Murilo Ijanc Date: Sun, 26 Jul 2026 15:18:21 -0300 Subject: Initial import --- .dockerignore | 6 + .gitignore | 6 + Dockerfile | 16 ++ LICENSE | 15 ++ Makefile | 56 +++++ README.md | 99 ++++++++ go.mod | 5 + go.sum | 2 + internal/app/app.go | 96 ++++++++ internal/app/app_test.go | 510 +++++++++++++++++++++++++++++++++++++++ internal/app/fakes_test.go | 285 ++++++++++++++++++++++ internal/bot/bot.go | 288 ++++++++++++++++++++++ internal/bot/command.go | 54 +++++ internal/bot/command_test.go | 59 +++++ internal/bot/format.go | 225 +++++++++++++++++ internal/bot/format_test.go | 126 ++++++++++ internal/bot/ratelimit.go | 139 +++++++++++ internal/bot/ratelimit_test.go | 128 ++++++++++ internal/config/config.go | 153 ++++++++++++ internal/config/config_test.go | 76 ++++++ internal/telegram/client.go | 359 +++++++++++++++++++++++++++ internal/telegram/client_test.go | 209 ++++++++++++++++ internal/telegram/poller.go | 198 +++++++++++++++ internal/telegram/poller_test.go | 247 +++++++++++++++++++ internal/telegram/types.go | 152 ++++++++++++ main.go | 50 ++++ 26 files changed, 3559 insertions(+) create mode 100644 .dockerignore create mode 100644 .gitignore create mode 100644 Dockerfile create mode 100644 LICENSE create mode 100644 Makefile create mode 100644 README.md create mode 100644 go.mod create mode 100644 go.sum create mode 100644 internal/app/app.go create mode 100644 internal/app/app_test.go create mode 100644 internal/app/fakes_test.go create mode 100644 internal/bot/bot.go create mode 100644 internal/bot/command.go create mode 100644 internal/bot/command_test.go create mode 100644 internal/bot/format.go create mode 100644 internal/bot/format_test.go create mode 100644 internal/bot/ratelimit.go create mode 100644 internal/bot/ratelimit_test.go create mode 100644 internal/config/config.go create mode 100644 internal/config/config_test.go create mode 100644 internal/telegram/client.go create mode 100644 internal/telegram/client_test.go create mode 100644 internal/telegram/poller.go create mode 100644 internal/telegram/poller_test.go create mode 100644 internal/telegram/types.go create mode 100644 main.go diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..e5857c8 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,6 @@ +.git +bin +dist +*.md +LICENSE +Makefile diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..6b05d27 --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +/bin/ +/dist/ +*.exe +*.test +*.out +coverage.html diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..e7bb4a7 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,16 @@ +FROM golang:1.24-alpine AS build + +WORKDIR /src + +COPY go.mod go.sum ./ +RUN go mod download + +COPY . . +RUN CGO_ENABLED=0 go build -trimpath -ldflags '-s -w' -o /out/cepis-telegram . + +FROM gcr.io/distroless/static-debian12:nonroot + +COPY --from=build /out/cepis-telegram /cepis-telegram + +USER nonroot:nonroot +ENTRYPOINT ["/cepis-telegram"] diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..5e9104d --- /dev/null +++ b/LICENSE @@ -0,0 +1,15 @@ +ISC License + +Copyright (c) 2026 Murilo Ijanc + +Permission to use, copy, modify, and/or distribute this software for any +purpose with or without fee is hereby granted, provided that the above +copyright notice and this permission notice appear in all copies. + +THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +PERFORMANCE OF THIS SOFTWARE. diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..2eeccc6 --- /dev/null +++ b/Makefile @@ -0,0 +1,56 @@ +# Development tasks. Everything runs through the `go` tool, so there is +# nothing to install first. +# +# make list the targets +# make test run the suite +# make lint gofmt check + go vet, no files touched +# make build compile the binary into bin/ +GO ?= go +BINARY ?= bin/cepis-telegram +IMAGE ?= cepis-telegram:dev + +.DEFAULT_GOAL := help +.PHONY: help test test-race lint format build image check clean + +help: + @echo 'Targets:' + @echo ' test run the test suite' + @echo ' test-race run it under the race detector' + @echo ' lint gofmt -l + go vet (read-only)' + @echo ' format apply gofmt (writes)' + @echo ' build compile the binary into $(BINARY)' + @echo ' image build the container image as $(IMAGE)' + @echo ' check lint + test + build, in that order' + @echo ' clean remove build and test artefacts' + +test: + $(GO) test ./... + +test-race: + $(GO) test -race -count=1 ./... + +# Read-only by design, so it is safe in CI and in a pre-commit hook: it reports +# what `make format` would change without changing anything. +lint: + @files=$$(gofmt -l .); \ + if [ -n "$$files" ]; then \ + echo 'gofmt would rewrite:'; echo "$$files"; exit 1; \ + fi + $(GO) vet ./... + +format: + gofmt -w . + +build: + CGO_ENABLED=0 $(GO) build -trimpath -ldflags '-s -w' -o $(BINARY) . + +image: + docker build -t $(IMAGE) . + +# The order matters: lint is the fastest to fail, so a broken run reports the +# cheapest problem first. +check: lint test build + +clean: + $(GO) clean -testcache + rm -rf bin diff --git a/README.md b/README.md new file mode 100644 index 0000000..b81b553 --- /dev/null +++ b/README.md @@ -0,0 +1,99 @@ +# cepis-telegram + +Bot de Telegram para consulta de CEP, sobre a API pública da +[cep.is](https://cep.is) — base oficial dos Correios (DNE), atualizada a cada +quinzena. + +Escrito em Go, usa o cliente oficial [`cep.is/go`](https://cep.is/go) e recebe +as mensagens por *long polling*: não precisa de porta aberta, de hostname +público nem de segredo de webhook. + +## O que ele faz + +| onde | como se usa | +| --- | --- | +| conversa privada | manda o CEP direto: `01001000` ou `01001-000` | +| grupos e canais | `/cep 01001000` — e também `/cep@seubot 01001000` | +| qualquer conversa | `@seubot 01001000` (modo inline, sem precisar adicionar o bot) | + +Também responde `/start` e `/help`. + +A resposta traz o endereço formatado e, quando o CEP tem coordenada, distingue +os dois casos: com precisão de logradouro sai um link de mapa; quando o CEP é +geral do município, sai um aviso com o raio da área em vez de um pino, porque o +ponto é o centroide da cidade e pode errar dezenas de quilómetros. + +## Criando o bot + +O username e o token são criados por você no +[@BotFather](https://t.me/BotFather) e pertencem a você — este repositório não +contém nem nunca deve conter um token. + +1. `/newbot` no @BotFather, escolha nome e username. Guarde o token. +2. `/setinline` no @BotFather, para o mesmo bot, com um texto de placeholder + (por exemplo `01001000`). Sem isso o modo inline fica desligado. +3. Opcional, `/setcommands`: + + ``` + cep - consulta um CEP + help - como usar + ``` + +O *privacy mode* pode continuar ligado (é o padrão). Em grupos o bot só recebe +os comandos endereçados a ele, que é exatamente o que ele precisa. + +## Rodando + +```sh +export TELEGRAM_BOT_TOKEN=... # o token do @BotFather +make build +./bin/cepis-telegram +``` + +Em container: + +```sh +docker build -t cepis-telegram . +docker run --rm -e TELEGRAM_BOT_TOKEN=... cepis-telegram +``` + +Só uma instância por token: o Telegram entrega cada update uma única vez, e dois +processos com o mesmo token disputam os mesmos updates. + +## Configuração + +Tudo vem do ambiente. + +| variável | padrão | o que faz | +| --- | --- | --- | +| `TELEGRAM_BOT_TOKEN` | — | obrigatória; o processo aborta na partida sem ela | +| `CEPIS_BASE_URL` | `https://cep.is` | aponta as consultas para outra instância da API | +| `TELEGRAM_API_BASE_URL` | `https://api.telegram.org` | aponta o bot para outro host da Bot API | +| `LOG_LEVEL` | `info` | `debug`, `info`, `warn` ou `error` | + +Os logs saem em JSON no stdout. O token nunca aparece neles, e o conteúdo das +mensagens dos usuários não é registrado. + +Há um limite por usuário de 5 consultas em rajada e uma a cada 3 segundos +depois disso. + +## Desenvolvimento + +```sh +make # lista os alvos +make test # sobe um Telegram falso e uma cep.is falsa, em processo +make lint # gofmt -l + go vet, não escreve nada +make format # aplica o gofmt +make check # lint + test + build +``` + +Para testar contra uma cep.is local (o repositório da API sobe uma com +`just up`): + +```sh +CEPIS_BASE_URL=http://localhost:8080 TELEGRAM_BOT_TOKEN=... ./bin/cepis-telegram +``` + +## Licença + +ISC. Veja [LICENSE](LICENSE). diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..b2681be --- /dev/null +++ b/go.mod @@ -0,0 +1,5 @@ +module cep.is/telegrambot + +go 1.23 + +require cep.is/go v0.0.0-20260726172041-323d9a8422f0 diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..dde6d3c --- /dev/null +++ b/go.sum @@ -0,0 +1,2 @@ +cep.is/go v0.0.0-20260726172041-323d9a8422f0 h1:dafV2FEOBQi4/hLSn6c+8NWh4PN6s+80AB0iAyLz6JQ= +cep.is/go v0.0.0-20260726172041-323d9a8422f0/go.mod h1:scg4o+dKixPVgqiEkccyZnwkNdzt1tRfdLf0bzaGAj4= diff --git a/internal/app/app.go b/internal/app/app.go new file mode 100644 index 0000000..0b78baf --- /dev/null +++ b/internal/app/app.go @@ -0,0 +1,96 @@ +// Package app wires the configuration, the cep.is client, the Telegram client +// and the bot into a single runnable unit. +// +// It exists so that main stays a thin shell around signal handling, and so the +// test suite can run the whole thing against a fake Telegram and a fake cep.is +// without going through the process environment. +package app + +import ( + "context" + "fmt" + "log/slog" + + cepis "cep.is/go" + + "cep.is/telegrambot/internal/bot" + "cep.is/telegrambot/internal/config" + "cep.is/telegrambot/internal/telegram" +) + +// App is one configured bot process. +type App struct { + poller *telegram.Poller + username string + logger *slog.Logger +} + +// New builds the application and identifies the bot. +// +// The getMe call is not a formality. It fails fast on a bad token, and it is the +// only way to learn the username the operator chose with @BotFather, which the +// help text quotes and which "/cep@username" has to match. +// +// The whole-call timeout of the cep.is client is the shutdown grace on purpose: +// handling one update is what a termination signal waits for, so a lookup must +// not be allowed to outlast that budget. +func New(ctx context.Context, cfg config.Config, logger *slog.Logger) (*App, error) { + if logger == nil { + logger = slog.Default() + } + + lookups, err := cepis.New( + cepis.WithBaseURL(cfg.CepisBaseURL), + cepis.WithTimeout(cfg.RequestTimeout), + cepis.WithMaxAttempts(cfg.MaxAttempts), + cepis.WithBackoff(cfg.BackoffBase, cfg.BackoffMax), + cepis.WithTotalTimeout(cfg.ShutdownGrace), + cepis.WithUserAgent("cepis-telegram/0.1.0 (+https://cep.is)"), + ) + if err != nil { + return nil, fmt.Errorf("app: cep.is client: %w", err) + } + + api, err := telegram.New(cfg.Token, + telegram.WithBaseURL(cfg.TelegramAPIBaseURL), + telegram.WithTimeout(cfg.RequestTimeout), + telegram.WithMaxAttempts(cfg.MaxAttempts), + telegram.WithBackoff(cfg.BackoffBase, cfg.BackoffMax), + ) + if err != nil { + return nil, fmt.Errorf("app: telegram client: %w", err) + } + + me, err := api.GetMe(ctx) + if err != nil { + return nil, fmt.Errorf("app: getMe: %w", err) + } + if me.Username == "" { + return nil, fmt.Errorf("app: getMe returned no username") + } + + handler := bot.New(lookups, api, me.Username, bot.WithLogger(logger)) + poller := telegram.NewPoller(api, handler, + telegram.WithLogger(logger), + telegram.WithPollTimeout(cfg.PollTimeout), + telegram.WithShutdownGrace(cfg.ShutdownGrace), + telegram.WithErrorBackoff(cfg.BackoffBase, cfg.BackoffMax), + ) + + return &App{poller: poller, username: me.Username, logger: logger}, nil +} + +// Username is the bot's own username, as reported by getMe. +func (a *App) Username() string { return a.username } + +// Offset is the identifier of the next update expected. It is meant for tests +// and diagnostics. +func (a *App) Offset() int64 { return a.poller.Offset() } + +// Run polls until ctx is cancelled and then shuts down cleanly. +func (a *App) Run(ctx context.Context) error { + a.logger.Info("bot started", slog.String("username", a.username)) + err := a.poller.Run(ctx) + a.logger.Info("bot stopped", slog.Int64("offset", a.poller.Offset())) + return err +} diff --git a/internal/app/app_test.go b/internal/app/app_test.go new file mode 100644 index 0000000..e770060 --- /dev/null +++ b/internal/app/app_test.go @@ -0,0 +1,510 @@ +package app + +import ( + "context" + "io" + "log/slog" + "strings" + "testing" + "time" + + "cep.is/telegrambot/internal/config" + "cep.is/telegrambot/internal/telegram" +) + +func testConfig(tg *fakeTelegram, api *fakeCEPIS) config.Config { + cfg := config.Defaults() + cfg.Token = fakeToken + cfg.TelegramAPIBaseURL = tg.URL() + cfg.CepisBaseURL = api.URL() + cfg.PollTimeout = time.Second + cfg.RequestTimeout = 2 * time.Second + cfg.MaxAttempts = 2 + cfg.BackoffBase = 5 * time.Millisecond + cfg.BackoffMax = 20 * time.Millisecond + cfg.ShutdownGrace = 3 * time.Second + return cfg +} + +func quietLogger() *slog.Logger { + return slog.New(slog.NewJSONHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelError})) +} + +type running struct { + app *App + stop context.CancelFunc + done chan error +} + +func (r *running) wait(t *testing.T) { + t.Helper() + r.stop() + select { + case err := <-r.done: + r.done <- err + case <-time.After(10 * time.Second): + t.Fatal("the bot did not shut down") + } +} + +func start(t *testing.T, cfg config.Config) *running { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + + application, err := New(ctx, cfg, quietLogger()) + if err != nil { + cancel() + t.Fatalf("New: %v", err) + } + + done := make(chan error, 1) + go func() { done <- application.Run(ctx) }() + + r := &running{app: application, stop: cancel, done: done} + t.Cleanup(func() { + cancel() + select { + case <-done: + case <-time.After(10 * time.Second): + } + }) + return r +} + +func privateMessage(id int64, userID int64, text string) telegram.Update { + return telegram.Update{ + UpdateID: id, + Message: &telegram.Message{ + MessageID: id, + From: &telegram.User{ID: userID, FirstName: "Tester"}, + Chat: telegram.Chat{ID: userID, Type: "private"}, + Text: text, + }, + } +} + +func groupMessage(id int64, text string) telegram.Update { + return telegram.Update{ + UpdateID: id, + Message: &telegram.Message{ + MessageID: id, + From: &telegram.User{ID: 777, FirstName: "Tester"}, + Chat: telegram.Chat{ID: -100200, Type: "supergroup", Title: "Grupo"}, + Text: text, + }, + } +} + +func channelPost(id int64, text string) telegram.Update { + return telegram.Update{ + UpdateID: id, + ChannelPost: &telegram.Message{ + MessageID: id, + Chat: telegram.Chat{ID: -100999, Type: "channel", Title: "Canal"}, + Text: text, + }, + } +} + +func inlineQuery(id int64, userID int64, query string) telegram.Update { + return telegram.Update{ + UpdateID: id, + InlineQuery: &telegram.InlineQuery{ + ID: "iq-" + query, + From: telegram.User{ID: userID, FirstName: "Tester"}, + Query: query, + }, + } +} + +func waitForMessages(t *testing.T, tg *fakeTelegram, n int) []telegram.SendMessageRequest { + t.Helper() + waitFor(t, "messages", 10*time.Second, func() bool { return len(tg.sentMessages()) >= n }) + return tg.sentMessages() +} + +func TestGetMeIdentifiesTheBot(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + r := start(t, testConfig(tg, api)) + if r.app.Username() != fakeUsername { + t.Fatalf("username = %q, want %q", r.app.Username(), fakeUsername) + } +} + +func TestPrivateBareCEP(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push(privateMessage(1, 555, "01001-000")) + start(t, testConfig(tg, api)) + + sent := waitForMessages(t, tg, 1) + if !strings.Contains(sent[0].Text, "Praça da Sé") { + t.Fatalf("reply does not carry the address: %q", sent[0].Text) + } + if sent[0].ChatID != 555 { + t.Fatalf("chat id = %d, want 555", sent[0].ChatID) + } + if sent[0].ReplyParameters != nil { + t.Fatal("a private answer should not quote the message") + } +} + +func TestGroupCommand(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push(groupMessage(1, "/cep 01001000")) + start(t, testConfig(tg, api)) + + sent := waitForMessages(t, tg, 1) + if !strings.Contains(sent[0].Text, "Praça da Sé") { + t.Fatalf("reply does not carry the address: %q", sent[0].Text) + } + if sent[0].ReplyParameters == nil || sent[0].ReplyParameters.MessageID != 1 { + t.Fatal("a group answer should quote the message that asked") + } +} + +func TestGroupCommandAddressedToTheBot(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push(groupMessage(1, "/cep@"+fakeUsername+" 01001000")) + start(t, testConfig(tg, api)) + + sent := waitForMessages(t, tg, 1) + if !strings.Contains(sent[0].Text, "Praça da Sé") { + t.Fatalf("reply does not carry the address: %q", sent[0].Text) + } +} + +func TestCommandAddressedToAnotherBotIsIgnored(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push( + groupMessage(1, "/cep@outrobot 01001000"), + groupMessage(2, "/cep 01001000"), + ) + start(t, testConfig(tg, api)) + + sent := waitForMessages(t, tg, 1) + time.Sleep(200 * time.Millisecond) + if got := len(tg.sentMessages()); got != 1 { + t.Fatalf("sent %d messages, want exactly 1", got) + } + if sent[0].ReplyParameters.MessageID != 2 { + t.Fatalf("answered message %d, want 2", sent[0].ReplyParameters.MessageID) + } +} + +func TestChannelPost(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push(channelPost(1, "/cep@"+fakeUsername+" 01001000")) + start(t, testConfig(tg, api)) + + sent := waitForMessages(t, tg, 1) + if sent[0].ChatID != -100999 { + t.Fatalf("chat id = %d, want -100999", sent[0].ChatID) + } + if !strings.Contains(sent[0].Text, "Praça da Sé") { + t.Fatalf("reply does not carry the address: %q", sent[0].Text) + } +} + +func TestGroupTextWithoutCommandIsIgnored(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push( + groupMessage(1, "01001000"), + groupMessage(2, "/cep 01001000"), + ) + start(t, testConfig(tg, api)) + + waitForMessages(t, tg, 1) + time.Sleep(200 * time.Millisecond) + if got := len(tg.sentMessages()); got != 1 { + t.Fatalf("sent %d messages, want exactly 1", got) + } +} + +func TestInlineQuery(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push(inlineQuery(1, 555, "01001000")) + start(t, testConfig(tg, api)) + + waitFor(t, "inline answer", 10*time.Second, func() bool { return len(tg.inlineAnswers()) >= 1 }) + answer := tg.inlineAnswers()[0] + if len(answer.Results) != 1 { + t.Fatalf("got %d results, want 1", len(answer.Results)) + } + result := answer.Results[0] + if result.Type != "article" { + t.Fatalf("result type = %q, want article", result.Type) + } + if !strings.Contains(result.Title, "01001-000") { + t.Fatalf("title = %q", result.Title) + } + if !strings.Contains(result.InputMessageContent.MessageText, "Praça da Sé") { + t.Fatalf("message content = %q", result.InputMessageContent.MessageText) + } +} + +func TestInlineQueryWhileStillTypingCostsNothing(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push(inlineQuery(1, 555, "0100")) + start(t, testConfig(tg, api)) + + waitFor(t, "inline answer", 10*time.Second, func() bool { return len(tg.inlineAnswers()) >= 1 }) + if got := len(tg.inlineAnswers()[0].Results); got != 0 { + t.Fatalf("got %d results for a partial CEP, want 0", got) + } + if got := api.requestCount(); got != 0 { + t.Fatalf("the API was called %d times for a partial CEP, want 0", got) + } +} + +func TestStartAndHelp(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push( + privateMessage(1, 555, "/start"), + privateMessage(2, 555, "/help"), + ) + start(t, testConfig(tg, api)) + + sent := waitForMessages(t, tg, 2) + for _, message := range sent { + for _, needed := range []string{"https://cep.is", "/cep 01001000", "@" + fakeUsername} { + if !strings.Contains(message.Text, needed) { + t.Fatalf("help text is missing %q: %q", needed, message.Text) + } + } + } + if !strings.Contains(sent[0].Text, "Olá") { + t.Fatalf("/start should greet: %q", sent[0].Text) + } +} + +func TestMalformedCEP(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push(groupMessage(1, "/cep 123")) + start(t, testConfig(tg, api)) + + sent := waitForMessages(t, tg, 1) + if !strings.Contains(sent[0].Text, "inválido") { + t.Fatalf("reply = %q, want an invalid-CEP message", sent[0].Text) + } + if got := api.requestCount(); got != 0 { + t.Fatalf("the API was called %d times for a malformed CEP, want 0", got) + } +} + +func TestCEPNotFound(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push(privateMessage(1, 555, "99999999")) + start(t, testConfig(tg, api)) + + sent := waitForMessages(t, tg, 1) + if !strings.Contains(sent[0].Text, "não encontrado") { + t.Fatalf("reply = %q, want a not-found message", sent[0].Text) + } +} + +func TestAPIServerError(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push(privateMessage(1, 555, "50000000")) + start(t, testConfig(tg, api)) + + sent := waitForMessages(t, tg, 1) + if !strings.Contains(sent[0].Text, "respondeu com erro") { + t.Fatalf("reply = %q, want a server-error message", sent[0].Text) + } +} + +func TestAPINetworkFailure(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + cfg := testConfig(tg, api) + cfg.MaxAttempts = 1 + api.server.Close() + + tg.push(privateMessage(1, 555, "01001000")) + start(t, cfg) + + sent := waitForMessages(t, tg, 1) + if !strings.Contains(sent[0].Text, "Não consegui falar com a cep.is") { + t.Fatalf("reply = %q, want a network-failure message", sent[0].Text) + } +} + +func TestRateLimiterLimits(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + + var updates []telegram.Update + for i := 1; i <= 8; i++ { + updates = append(updates, privateMessage(int64(i), 555, "01001000")) + } + tg.push(updates...) + start(t, testConfig(tg, api)) + + waitFor(t, "the limiter to kick in", 10*time.Second, func() bool { + for _, message := range tg.sentMessages() { + if strings.Contains(message.Text, "muitas consultas") { + return true + } + } + return false + }) + time.Sleep(300 * time.Millisecond) + + sent := tg.sentMessages() + addresses, warnings := 0, 0 + for _, message := range sent { + switch { + case strings.Contains(message.Text, "Praça da Sé"): + addresses++ + case strings.Contains(message.Text, "muitas consultas"): + warnings++ + } + } + if addresses != 5 { + t.Fatalf("answered %d lookups, want 5 (the burst)", addresses) + } + if warnings != 1 { + t.Fatalf("sent %d warnings, want exactly 1 per window", warnings) + } + if got := api.requestCount(); got != 5 { + t.Fatalf("the API saw %d requests, want 5", got) + } +} + +func TestTelegram429RetryAfterIsHonoured(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.failNextSends(cannedFailure{ + status: 429, code: 429, description: "Too Many Requests: retry after 1", retryAfter: 1, + }) + tg.push(privateMessage(1, 555, "01001000")) + + started := time.Now() + start(t, testConfig(tg, api)) + + sent := waitForMessages(t, tg, 1) + elapsed := time.Since(started) + if elapsed < time.Second { + t.Fatalf("retried after %v, want at least the 1s Telegram asked for", elapsed) + } + if !strings.Contains(sent[0].Text, "Praça da Sé") { + t.Fatalf("reply = %q", sent[0].Text) + } +} + +func TestOffsetAdvancesAcrossPolls(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push(privateMessage(10, 555, "01001000")) + r := start(t, testConfig(tg, api)) + + waitForMessages(t, tg, 1) + tg.push(privateMessage(11, 555, "99999999")) + waitForMessages(t, tg, 2) + + waitFor(t, "the second offset to be confirmed", 5*time.Second, func() bool { + for _, offset := range tg.requestedOffsets() { + if offset == 12 { + return true + } + } + return false + }) + + offsets := tg.requestedOffsets() + if offsets[0] != 0 { + t.Fatalf("first poll asked for offset %d, want 0", offsets[0]) + } + seen11 := false + for _, offset := range offsets { + if offset == 11 { + seen11 = true + } + if offset > 12 { + t.Fatalf("offset ran past the last update: %v", offsets) + } + } + if !seen11 { + t.Fatalf("offset never confirmed the first update: %v", offsets) + } + r.wait(t) + if r.app.Offset() != 12 { + t.Fatalf("final offset = %d, want 12", r.app.Offset()) + } +} + +func TestGracefulShutdownKeepsTheOffset(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push( + privateMessage(1, 555, "01001000"), + privateMessage(2, 555, "99999999"), + ) + r := start(t, testConfig(tg, api)) + + waitForMessages(t, tg, 2) + r.wait(t) + + offsets := tg.requestedOffsets() + last := offsets[len(offsets)-1] + if last != 3 { + t.Fatalf("last offset sent was %d, want 3 so the handled updates stay confirmed", last) + } + if got := tg.pendingCount(); got != 0 { + t.Fatalf("%d updates are still unconfirmed and would be replayed", got) + } +} + +func TestPrecisionChangesTheReply(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push( + privateMessage(1, 555, "01001000"), + privateMessage(2, 556, "76200000"), + ) + start(t, testConfig(tg, api)) + + sent := waitForMessages(t, tg, 2) + street, municipality := sent[0].Text, sent[1].Text + + if street == municipality { + t.Fatal("both precisions produced the same reply") + } + if !strings.Contains(street, "openstreetmap.org") { + t.Fatalf("a street-level CEP should carry a map link: %q", street) + } + if !strings.Contains(street, "128 m") { + t.Fatalf("a street-level CEP should state its radius: %q", street) + } + + if strings.Contains(municipality, "openstreetmap.org") { + t.Fatalf("a municipality CEP must never carry a map link: %q", municipality) + } + if !strings.Contains(municipality, "CEP geral de município") { + t.Fatalf("a municipality CEP must say so: %q", municipality) + } + if !strings.Contains(municipality, "18,4 km") { + t.Fatalf("a municipality CEP must state its radius: %q", municipality) + } + if !strings.Contains(municipality, "centroide") { + t.Fatalf("a municipality CEP must name the point: %q", municipality) + } +} + +func TestAddressWithoutCoordinate(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + tg.push(privateMessage(1, 555, "29902000")) + start(t, testConfig(tg, api)) + + sent := waitForMessages(t, tg, 1) + if strings.Contains(sent[0].Text, "openstreetmap.org") { + t.Fatalf("no coordinate should mean no map link: %q", sent[0].Text) + } + if !strings.Contains(sent[0].Text, "Sem coordenada") { + t.Fatalf("reply = %q", sent[0].Text) + } +} + +func TestBadTokenFailsAtStartup(t *testing.T) { + tg, api := newFakeTelegram(t), newFakeCEPIS(t) + cfg := testConfig(tg, api) + cfg.Token = "424242:a-different-token" + + if _, err := New(context.Background(), cfg, quietLogger()); err == nil { + t.Fatal("New accepted a token the API rejects") + } +} diff --git a/internal/app/fakes_test.go b/internal/app/fakes_test.go new file mode 100644 index 0000000..011d4af --- /dev/null +++ b/internal/app/fakes_test.go @@ -0,0 +1,285 @@ +package app + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "cep.is/telegrambot/internal/telegram" +) + +const fakeToken = "424242:not-a-real-token-generated-for-tests" + +const fakeUsername = "cepisbot" + +type cannedFailure struct { + status int + code int + description string + retryAfter int +} + +// fakeTelegram is a real HTTP server speaking the Bot API, including the +// confirmation semantics of getUpdates: an update stays pending until a later +// call arrives with a higher offset. Without that, an offset bug in the poller +// would go unnoticed. +type fakeTelegram struct { + server *httptest.Server + username string + + mu sync.Mutex + pending []telegram.Update + offsets []int64 + sent []telegram.SendMessageRequest + inline []telegram.AnswerInlineQueryRequest + sendFailures []cannedFailure + getMeCalls int +} + +func newFakeTelegram(t *testing.T) *fakeTelegram { + t.Helper() + f := &fakeTelegram{username: fakeUsername} + f.server = httptest.NewServer(f) + t.Cleanup(f.server.Close) + return f +} + +func (f *fakeTelegram) URL() string { return f.server.URL } + +func (f *fakeTelegram) ServeHTTP(w http.ResponseWriter, r *http.Request) { + prefix, method, ok := strings.Cut(strings.TrimPrefix(r.URL.Path, "/"), "/") + if !ok || prefix != "bot"+fakeToken { + writeEnvelope(w, http.StatusUnauthorized, envelope{ + OK: false, ErrorCode: 401, Description: "Unauthorized", + }) + return + } + + switch method { + case "getMe": + f.mu.Lock() + f.getMeCalls++ + f.mu.Unlock() + writeResult(w, telegram.User{ID: 42, IsBot: true, FirstName: "cep.is", Username: f.username}) + case "getUpdates": + f.getUpdates(w, r) + case "sendMessage": + f.sendMessage(w, r) + case "answerInlineQuery": + var req telegram.AnswerInlineQueryRequest + decode(r, &req) + f.mu.Lock() + f.inline = append(f.inline, req) + f.mu.Unlock() + writeResult(w, true) + default: + writeEnvelope(w, http.StatusNotFound, envelope{ + OK: false, ErrorCode: 404, Description: "Not Found: method not found", + }) + } +} + +func (f *fakeTelegram) getUpdates(w http.ResponseWriter, r *http.Request) { + var req telegram.GetUpdatesRequest + decode(r, &req) + + f.mu.Lock() + f.offsets = append(f.offsets, req.Offset) + if req.Offset > 0 { + kept := f.pending[:0] + for _, update := range f.pending { + if update.UpdateID >= req.Offset { + kept = append(kept, update) + } + } + f.pending = kept + } + f.mu.Unlock() + + deadline := time.Now().Add(time.Duration(req.Timeout) * time.Second) + for { + f.mu.Lock() + out := append([]telegram.Update(nil), f.pending...) + f.mu.Unlock() + + if len(out) > 0 || time.Now().After(deadline) { + if req.Limit > 0 && len(out) > req.Limit { + out = out[:req.Limit] + } + writeResult(w, out) + return + } + select { + case <-r.Context().Done(): + writeResult(w, []telegram.Update{}) + return + case <-time.After(5 * time.Millisecond): + } + } +} + +func (f *fakeTelegram) sendMessage(w http.ResponseWriter, r *http.Request) { + var req telegram.SendMessageRequest + decode(r, &req) + + f.mu.Lock() + if len(f.sendFailures) > 0 { + failure := f.sendFailures[0] + f.sendFailures = f.sendFailures[1:] + f.mu.Unlock() + body := envelope{OK: false, ErrorCode: failure.code, Description: failure.description} + if failure.retryAfter > 0 { + body.Parameters = &telegram.ResponseParameters{RetryAfter: failure.retryAfter} + } + writeEnvelope(w, failure.status, body) + return + } + f.sent = append(f.sent, req) + f.mu.Unlock() + + writeResult(w, map[string]any{"message_id": 1}) +} + +func (f *fakeTelegram) push(updates ...telegram.Update) { + f.mu.Lock() + defer f.mu.Unlock() + f.pending = append(f.pending, updates...) +} + +func (f *fakeTelegram) failNextSends(failures ...cannedFailure) { + f.mu.Lock() + defer f.mu.Unlock() + f.sendFailures = append(f.sendFailures, failures...) +} + +func (f *fakeTelegram) sentMessages() []telegram.SendMessageRequest { + f.mu.Lock() + defer f.mu.Unlock() + return append([]telegram.SendMessageRequest(nil), f.sent...) +} + +func (f *fakeTelegram) inlineAnswers() []telegram.AnswerInlineQueryRequest { + f.mu.Lock() + defer f.mu.Unlock() + return append([]telegram.AnswerInlineQueryRequest(nil), f.inline...) +} + +func (f *fakeTelegram) requestedOffsets() []int64 { + f.mu.Lock() + defer f.mu.Unlock() + return append([]int64(nil), f.offsets...) +} + +func (f *fakeTelegram) pendingCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.pending) +} + +// fakeCEPIS is a real HTTP server speaking the cep.is native contract, with +// enough of it to exercise every branch the bot has: an address with a street +// coordinate, one that is a whole municipality, a 404, a 400 and a 500. +type fakeCEPIS struct { + server *httptest.Server + + mu sync.Mutex + bodies map[string]string + statuses map[string]int + requests int +} + +func newFakeCEPIS(t *testing.T) *fakeCEPIS { + t.Helper() + f := &fakeCEPIS{ + bodies: map[string]string{ + "01001000": `{"cep":"01001000","logradouro":"Praça da Sé","complemento":"lado ímpar",` + + `"unidade":"","bairro":"Sé","localidade":"São Paulo","uf":"SP","estado":"São Paulo",` + + `"regiao":"Sudeste","ibge":"3550308","gia":"1004","ddd":"11","siafi":"7107",` + + `"geo":{"lat":-23.567202,"lon":-46.649659,"raio_m":128,"precisao":"logradouro"}}`, + "76200000": `{"cep":"76200000","logradouro":"","complemento":"","unidade":"",` + + `"bairro":"","localidade":"Iporá","uf":"GO","estado":"Goiás","regiao":"Centro-Oeste",` + + `"ibge":"5210000","gia":"","ddd":"64","siafi":"9439",` + + `"geo":{"lat":-16.441,"lon":-51.118,"raio_m":18400,"precisao":"localidade"}}`, + "29902000": `{"cep":"29902000","logradouro":"Rua Sem Coordenada","complemento":"",` + + `"unidade":"","bairro":"Centro","localidade":"Linhares","uf":"ES","estado":"Espírito Santo",` + + `"regiao":"Sudeste","ibge":"3203205","gia":"","ddd":"27","siafi":"5665"}`, + }, + statuses: map[string]int{ + "50000000": http.StatusInternalServerError, + }, + } + f.server = httptest.NewServer(http.HandlerFunc(f.handle)) + t.Cleanup(f.server.Close) + return f +} + +func (f *fakeCEPIS) URL() string { return f.server.URL } + +func (f *fakeCEPIS) handle(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + f.requests++ + cep := strings.TrimPrefix(r.URL.Path, "/") + status, forced := f.statuses[cep] + body, known := f.bodies[cep] + f.mu.Unlock() + + w.Header().Set("Content-Type", "application/json; charset=utf-8") + switch { + case forced: + w.WriteHeader(status) + w.Write([]byte(`{"erro":"interno","mensagem":"falha interna"}`)) + case known: + w.WriteHeader(http.StatusOK) + w.Write([]byte(body)) + case len(cep) != 8: + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte(`{"erro":"cep_invalido","mensagem":"CEP deve ter 8 dígitos"}`)) + default: + w.WriteHeader(http.StatusNotFound) + w.Write([]byte(`{"erro":"cep_nao_encontrado","mensagem":"CEP não existe na base do DNE"}`)) + } +} + +func (f *fakeCEPIS) requestCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.requests +} + +type envelope struct { + OK bool `json:"ok"` + Result any `json:"result,omitempty"` + ErrorCode int `json:"error_code,omitempty"` + Description string `json:"description,omitempty"` + Parameters *telegram.ResponseParameters `json:"parameters,omitempty"` +} + +func writeResult(w http.ResponseWriter, result any) { + writeEnvelope(w, http.StatusOK, envelope{OK: true, Result: result}) +} + +func writeEnvelope(w http.ResponseWriter, status int, body envelope) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + json.NewEncoder(w).Encode(body) +} + +func decode(r *http.Request, out any) { + json.NewDecoder(r.Body).Decode(out) +} + +func waitFor(t *testing.T, what string, timeout time.Duration, cond func() bool) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s", what) +} diff --git a/internal/bot/bot.go b/internal/bot/bot.go new file mode 100644 index 0000000..b29dc36 --- /dev/null +++ b/internal/bot/bot.go @@ -0,0 +1,288 @@ +// Package bot turns a Telegram update into an answer about a CEP. +// +// It knows nothing about how the update arrived. [Bot.Handle] satisfies +// telegram.Handler, which the long-polling loop drives today and which a +// webhook receiver could drive tomorrow without touching anything here. +package bot + +import ( + "context" + "errors" + "log/slog" + "strings" + + cepis "cep.is/go" + + "cep.is/telegrambot/internal/telegram" +) + +// Lookups is the subset of the cep.is client this bot needs. Depending on the +// interface rather than the concrete client keeps the seam narrow and visible. +type Lookups interface { + Lookup(ctx context.Context, cep string) (*cepis.Address, error) +} + +// Messenger is the subset of the Bot API this bot calls to answer. +type Messenger interface { + SendMessage(ctx context.Context, req telegram.SendMessageRequest) error + AnswerInlineQuery(ctx context.Context, req telegram.AnswerInlineQueryRequest) error +} + +// Cache times for inline answers, in seconds. A found address is stable for a +// day — the DNE database changes twice a month — so Telegram may serve it from +// its own cache and spare both this bot and the API. Anything else is not +// cached, so a transient failure is not pinned to a query for hours. +const ( + inlineCacheFound = 86400 + inlineCacheOther = 0 +) + +// Bot answers CEP lookups. +type Bot struct { + lookups Lookups + messenger Messenger + limiter *Limiter + username string + logger *slog.Logger +} + +// Option configures a [Bot]. +type Option func(*Bot) + +// WithLimiter replaces the per-user token bucket. +func WithLimiter(limiter *Limiter) Option { + return func(b *Bot) { + if limiter != nil { + b.limiter = limiter + } + } +} + +// WithLogger sets the logger. +func WithLogger(logger *slog.Logger) Option { + return func(b *Bot) { + if logger != nil { + b.logger = logger + } + } +} + +// New builds a bot. The username is the bot's own, as reported by getMe, and is +// what makes "/cep@username" and the inline examples in the help text correct. +func New(lookups Lookups, messenger Messenger, username string, opts ...Option) *Bot { + b := &Bot{ + lookups: lookups, + messenger: messenger, + limiter: NewLimiter(), + username: username, + logger: slog.Default(), + } + for _, opt := range opts { + opt(b) + } + return b +} + +// Handle dispatches one update. It implements telegram.Handler. +func (b *Bot) Handle(ctx context.Context, update telegram.Update) error { + switch { + case update.InlineQuery != nil: + return b.handleInline(ctx, update.InlineQuery) + case update.Message != nil: + return b.handleMessage(ctx, update.Message) + case update.ChannelPost != nil: + return b.handleMessage(ctx, update.ChannelPost) + } + return nil +} + +// handleMessage covers both messages and channel posts. They have the same +// shape, and the only difference that matters is that a channel post has no +// author, which the rate-limit key already accounts for. +func (b *Bot) handleMessage(ctx context.Context, message *telegram.Message) error { + if message.From != nil && message.From.IsBot { + return nil + } + + command, isCommand := ParseCommand(message.Text) + if isCommand { + if !command.AddressedTo(b.username) { + return nil + } + return b.handleCommand(ctx, message, command) + } + + if !message.Chat.IsPrivate() { + return nil + } + + text := strings.TrimSpace(message.Text) + if text == "" { + return nil + } + if _, err := cepis.NormalizeCEP(text); err != nil { + return b.reply(ctx, message, UnknownMessage) + } + return b.lookupAndReply(ctx, message, text) +} + +func (b *Bot) handleCommand(ctx context.Context, message *telegram.Message, command Command) error { + switch command.Name { + case "start": + return b.reply(ctx, message, StartMessage(b.username)) + case "help": + return b.reply(ctx, message, HelpMessage(b.username)) + case "cep": + if command.Argument == "" { + return b.reply(ctx, message, UsageMessage(b.username)) + } + return b.lookupAndReply(ctx, message, command.Argument) + } + + if message.Chat.IsPrivate() { + return b.reply(ctx, message, UnknownMessage) + } + return nil +} + +func (b *Bot) lookupAndReply(ctx context.Context, message *telegram.Message, raw string) error { + decision := b.limiter.Allow(rateKey(message)) + if !decision.Allowed { + b.logger.Debug("rate limited", slog.Int64("key", rateKey(message))) + if decision.Warn { + return b.reply(ctx, message, RateLimitMessage(decision.RetryAfter)) + } + return nil + } + + address, err := b.lookups.Lookup(ctx, raw) + if err != nil { + b.logger.Info("lookup failed", + slog.String("chat_type", message.Chat.Type), + slog.String("error", err.Error())) + return b.reply(ctx, message, ErrorMessage(err)) + } + return b.reply(ctx, message, FormatAddress(address)) +} + +func (b *Bot) reply(ctx context.Context, message *telegram.Message, text string) error { + req := telegram.SendMessageRequest{ + ChatID: message.Chat.ID, + Text: text, + ParseMode: ParseMode, + LinkPreviewOptions: &telegram.LinkPreviewOptions{IsDisabled: true}, + } + if !message.Chat.IsPrivate() { + req.ReplyParameters = &telegram.ReplyParameters{ + MessageID: message.MessageID, + AllowSendingWithoutReply: true, + } + } + return b.messenger.SendMessage(ctx, req) +} + +// handleInline answers an inline query. +// +// Telegram fires one query per keystroke, so a partial CEP must cost nothing: a +// query that is not eight digits is answered with an empty list, without +// touching the API and without spending a rate-limit token. Only a complete CEP +// becomes a lookup. +func (b *Bot) handleInline(ctx context.Context, query *telegram.InlineQuery) error { + text := strings.TrimSpace(query.Query) + if text == "" { + return b.answerInline(ctx, query, nil, inlineCacheOther) + } + + if _, err := cepis.NormalizeCEP(text); err != nil { + if looksPartial(text) { + return b.answerInline(ctx, query, nil, inlineCacheOther) + } + return b.answerInline(ctx, query, []telegram.InlineQueryResult{ + b.inlineNotice(query, "CEP inválido", "Use oito dígitos: 01001000", ErrorMessage(err)), + }, inlineCacheOther) + } + + decision := b.limiter.Allow(query.From.ID) + if !decision.Allowed { + return b.answerInline(ctx, query, nil, inlineCacheOther) + } + + address, err := b.lookups.Lookup(ctx, text) + if err != nil { + b.logger.Info("inline lookup failed", slog.String("error", err.Error())) + return b.answerInline(ctx, query, []telegram.InlineQueryResult{ + b.inlineNotice(query, inlineErrorTitle(err), "Toque para enviar o motivo", ErrorMessage(err)), + }, inlineCacheOther) + } + + return b.answerInline(ctx, query, []telegram.InlineQueryResult{{ + Type: "article", + ID: address.CEP, + Title: InlineTitle(address), + Description: InlineDescription(address), + InputMessageContent: telegram.InputMessageContent{ + MessageText: FormatAddress(address), + ParseMode: ParseMode, + LinkPreviewOptions: &telegram.LinkPreviewOptions{IsDisabled: true}, + }, + }}, inlineCacheFound) +} + +func (b *Bot) answerInline(ctx context.Context, query *telegram.InlineQuery, results []telegram.InlineQueryResult, cacheTime int) error { + if results == nil { + results = []telegram.InlineQueryResult{} + } + return b.messenger.AnswerInlineQuery(ctx, telegram.AnswerInlineQueryRequest{ + InlineQueryID: query.ID, + Results: results, + CacheTime: cacheTime, + }) +} + +func (b *Bot) inlineNotice(query *telegram.InlineQuery, title, description, text string) telegram.InlineQueryResult { + return telegram.InlineQueryResult{ + Type: "article", + ID: "notice-" + query.ID, + Title: title, + Description: description, + InputMessageContent: telegram.InputMessageContent{ + MessageText: text, + ParseMode: ParseMode, + LinkPreviewOptions: &telegram.LinkPreviewOptions{IsDisabled: true}, + }, + } +} + +func inlineErrorTitle(err error) string { + if isNotFound(err) { + return "CEP não encontrado" + } + return "Não consegui consultar agora" +} + +// rateKey is the token bucket's key: the author when there is one, and the chat +// otherwise. A channel post has no author, and channel ids are negative, so the +// two spaces cannot collide. +func rateKey(message *telegram.Message) int64 { + if message.From != nil { + return message.From.ID + } + return message.Chat.ID +} + +// looksPartial reports whether an inline query is plausibly a CEP still being +// typed, in which case answering with nothing is better than accusing the user +// of a mistake at every keystroke. +func looksPartial(text string) bool { + if len(text) > 9 { + return false + } + for _, r := range text { + if (r < '0' || r > '9') && r != '-' && r != '.' { + return false + } + } + return true +} + +func isNotFound(err error) bool { return errors.Is(err, cepis.ErrCEPNotFound) } diff --git a/internal/bot/command.go b/internal/bot/command.go new file mode 100644 index 0000000..9e13060 --- /dev/null +++ b/internal/bot/command.go @@ -0,0 +1,54 @@ +package bot + +import "strings" + +// Command is a parsed "/name@target argument" line. +type Command struct { + // Name is the command without the leading slash, lowercased. + Name string + + // Target is the bot username the command was addressed to, without the + // "@", or "" when the user did not address anyone. + Target string + + // Argument is everything after the command, trimmed. + Argument string +} + +// ParseCommand splits a message that starts with a slash. +// +// The "@username" suffix is not optional to handle. Telegram appends it +// automatically whenever more than one bot is in a chat, so in exactly the busy +// groups where the bot is most visible, every command arrives as +// "/cep@thebotname 01001000". A bot that only matches the bare form looks broken +// there. +// +// ok is false when the text is not a command at all. +func ParseCommand(text string) (Command, bool) { + trimmed := strings.TrimSpace(text) + if !strings.HasPrefix(trimmed, "/") || len(trimmed) == 1 { + return Command{}, false + } + + head, argument, _ := strings.Cut(trimmed[1:], " ") + name, target, _ := strings.Cut(head, "@") + if name == "" { + return Command{}, false + } + + return Command{ + Name: strings.ToLower(name), + Target: target, + Argument: strings.TrimSpace(argument), + }, true +} + +// AddressedTo reports whether the command is for the bot with this username. +// +// An unaddressed command counts as addressed: in a private chat nobody types +// the suffix, and in a group with a single bot Telegram does not add it either. +// A command addressed to some other bot is not ours, and answering it would put +// two replies under one message. +func (c Command) AddressedTo(username string) bool { + return c.Target == "" || strings.EqualFold(c.Target, username) +} diff --git a/internal/bot/command_test.go b/internal/bot/command_test.go new file mode 100644 index 0000000..fe1c895 --- /dev/null +++ b/internal/bot/command_test.go @@ -0,0 +1,59 @@ +package bot + +import "testing" + +func TestParseCommand(t *testing.T) { + cases := []struct { + text string + ok bool + name string + target string + argument string + }{ + {text: "/cep 01001000", ok: true, name: "cep", argument: "01001000"}, + {text: "/cep@cepisbot 01001-000", ok: true, name: "cep", target: "cepisbot", argument: "01001-000"}, + {text: " /CEP@CepisBot 01001000 ", ok: true, name: "cep", target: "CepisBot", argument: "01001000"}, + {text: "/start", ok: true, name: "start"}, + {text: "/help@cepisbot", ok: true, name: "help", target: "cepisbot"}, + {text: "01001000", ok: false}, + {text: "", ok: false}, + {text: "/", ok: false}, + {text: "não é comando /cep", ok: false}, + } + + for _, tc := range cases { + command, ok := ParseCommand(tc.text) + if ok != tc.ok { + t.Fatalf("ParseCommand(%q) ok = %v, want %v", tc.text, ok, tc.ok) + } + if !ok { + continue + } + if command.Name != tc.name || command.Target != tc.target || command.Argument != tc.argument { + t.Fatalf("ParseCommand(%q) = %+v, want {%q %q %q}", + tc.text, command, tc.name, tc.target, tc.argument) + } + } +} + +func TestAddressedTo(t *testing.T) { + cases := []struct { + text string + want bool + }{ + {"/cep 01001000", true}, + {"/cep@cepisbot 01001000", true}, + {"/cep@CEPISBOT 01001000", true}, + {"/cep@outrobot 01001000", false}, + } + + for _, tc := range cases { + command, ok := ParseCommand(tc.text) + if !ok { + t.Fatalf("ParseCommand(%q) did not parse", tc.text) + } + if got := command.AddressedTo("cepisbot"); got != tc.want { + t.Fatalf("AddressedTo(%q) = %v, want %v", tc.text, got, tc.want) + } + } +} diff --git a/internal/bot/format.go b/internal/bot/format.go new file mode 100644 index 0000000..9a9a22d --- /dev/null +++ b/internal/bot/format.go @@ -0,0 +1,225 @@ +package bot + +import ( + "errors" + "fmt" + "html" + "strconv" + "strings" + "time" + + cepis "cep.is/go" +) + +// ParseMode is the formatting Telegram applies to every message this bot sends. +// HTML is used rather than Markdown because address data is arbitrary text from +// the Correios database and escaping three characters is a complete defence, +// while Markdown has a dozen context-dependent ones. +const ParseMode = "HTML" + +// FormatAddress renders an address as the HTML body of a Telegram message. +// +// The coordinate section is where this function earns its keep. A CEP whose +// precisao is "logradouro" gets a map link: the point is a stretch of street +// and following it lands where the user expects. A CEP whose precisao is +// "localidade" gets no link at all, because the CEP is the entire municipality +// and its point is the centroid — dropping a pin there looks exactly like an +// address and is wrong by kilometres. That case is labelled, given its radius +// in kilometres, and told plainly not to be used as a delivery point. +func FormatAddress(address *cepis.Address) string { + var b strings.Builder + + b.WriteString("📮 " + esc(address.FormattedCEP()) + "\n") + + if line := streetLine(address); line != "" { + b.WriteString(esc(line) + "\n") + } + if address.Unidade != "" { + b.WriteString(esc(address.Unidade) + "\n") + } + b.WriteString(esc(placeLine(address)) + "\n") + if line := metaLine(address); line != "" { + b.WriteString("" + esc(line) + "\n") + } + + b.WriteString(geoSection(address)) + return strings.TrimRight(b.String(), "\n") +} + +// InlineTitle is the one-line label shown in the inline results list. +func InlineTitle(address *cepis.Address) string { + if address.Logradouro != "" { + return address.FormattedCEP() + " — " + address.Logradouro + } + return address.FormattedCEP() + " — " + address.Localidade +} + +// InlineDescription is the second line of an inline result. It repeats the +// municipality warning, so someone picking a result from the list sees it +// before sending the message rather than after. +func InlineDescription(address *cepis.Address) string { + parts := []string{placeLine(address)} + if address.Geo != nil && !address.Geo.IsPrecise() { + parts = append(parts, "CEP geral do município") + } + return strings.Join(parts, " · ") +} + +func streetLine(address *cepis.Address) string { + parts := []string{} + if address.Logradouro != "" { + parts = append(parts, address.Logradouro) + } + if address.Complemento != "" { + parts = append(parts, address.Complemento) + } + return strings.Join(parts, ", ") +} + +func placeLine(address *cepis.Address) string { + place := address.Localidade + if address.UF != "" { + place += "/" + address.UF + } + if address.Bairro != "" { + return address.Bairro + " — " + place + } + return place +} + +func metaLine(address *cepis.Address) string { + parts := []string{} + if address.Regiao != "" { + parts = append(parts, address.Regiao) + } + if address.DDD != "" { + parts = append(parts, "DDD "+address.DDD) + } + if address.IBGE != "" { + parts = append(parts, "IBGE "+address.IBGE) + } + return strings.Join(parts, " · ") +} + +func geoSection(address *cepis.Address) string { + geo := address.Geo + if geo == nil { + return "\nSem coordenada para este CEP na base do IBGE.\n" + } + + if geo.IsPrecise() { + return "\n🗺 Ver no mapa\n" + + "Precisão de logradouro — o ponto é o trecho de rua (raio aprox. " + + esc(formatRadius(geo.RaioM)) + ").\n" + } + + return "\n⚠️ CEP geral de município — não corresponde a um endereço.\n" + + "A única coordenada disponível é o centroide de " + esc(address.Localidade) + + ", com raio aproximado de " + esc(formatRadius(geo.RaioM)) + + ". Não serve como ponto de entrega nem para calcular frete, " + + "por isso não mando link de mapa.\n" + + "" + esc(coordText(geo.Lat, geo.Lon)) + "\n" +} + +func mapURL(lat, lon float64) string { + latText := strconv.FormatFloat(lat, 'f', 6, 64) + lonText := strconv.FormatFloat(lon, 'f', 6, 64) + return "https://www.openstreetmap.org/?mlat=" + latText + "&mlon=" + lonText + + "#map=17/" + latText + "/" + lonText +} + +func coordText(lat, lon float64) string { + return "Centroide: " + strconv.FormatFloat(lat, 'f', 6, 64) + ", " + + strconv.FormatFloat(lon, 'f', 6, 64) +} + +func formatRadius(metres int) string { + if metres <= 0 { + return "desconhecido" + } + if metres < 1000 { + return strconv.Itoa(metres) + " m" + } + return strings.Replace(fmt.Sprintf("%.1f km", float64(metres)/1000), ".", ",", 1) +} + +// ErrorMessage translates a failure from the cep.is client into a sentence in +// Portuguese that says what happened and what the user can do about it. +// +// Every branch is spelled out. A user who gets no answer assumes the bot is +// broken and stops using it, and "não encontrado" and "a API está fora" call +// for completely different reactions. +func ErrorMessage(err error) string { + switch { + case errors.Is(err, cepis.ErrCEPNotFound): + return "🔍 CEP não encontrado na base dos Correios (DNE).\n" + + "Confira os oito dígitos — a base é atualizada a cada quinze dias." + case errors.Is(err, cepis.ErrInvalidCEP): + return "✋ CEP inválido. Preciso de oito dígitos: 01001000 " + + "ou 01001-000." + case errors.Is(err, cepis.ErrRateLimited): + return "🚦 A API está limitando as consultas neste momento. " + + "Tente de novo em alguns instantes." + case errors.Is(err, cepis.ErrServer): + return "💥 A API da cep.is respondeu com erro. " + + "Não é problema do seu CEP — tente de novo em alguns minutos." + case errors.Is(err, cepis.ErrNetwork): + return "🔌 Não consegui falar com a cep.is. " + + "Pode ser instabilidade de rede — tente de novo em alguns minutos." + case errors.Is(err, cepis.ErrInvalidResponse): + return "🤔 A cep.is respondeu algo que não consegui interpretar. " + + "Tente de novo em alguns minutos." + } + return "⚠️ Não consegui consultar este CEP agora. Tente de novo em alguns minutos." +} + +// RateLimitMessage is what a user held by the local token bucket is told, once +// per warning window. +func RateLimitMessage(retryAfter time.Duration) string { + seconds := int(retryAfter.Seconds()) + if seconds < 1 { + seconds = 1 + } + return "🐢 Calma — muitas consultas seguidas. Tente de novo em " + + strconv.Itoa(seconds) + "s." +} + +// HelpMessage explains the three ways to use the bot. The username comes from +// getMe, so the examples are copy-pasteable for whichever bot the operator +// created with @BotFather. +func HelpMessage(username string) string { + at := "@" + username + return "cep.is — consulta de CEP na base oficial dos Correios (DNE).\n\n" + + "Aqui no privado\n" + + "Mande o CEP e pronto: 01001000 ou 01001-000.\n\n" + + "Em grupos e canais\n" + + "Use /cep 01001000. Se houver outros bots, " + + "/cep" + esc(at) + " 01001000 também funciona.\n\n" + + "Em qualquer conversa\n" + + "Digite " + esc(at) + " 01001000 e escolha o resultado. " + + "Não precisa me adicionar ao grupo.\n\n" + + "Cerca de 57% dos CEPs têm coordenada. Quando o CEP é geral do " + + "município, eu aviso em vez de mandar um pino no mapa.\n\n" + + "API pública e documentação: https://cep.is" +} + +// StartMessage is the answer to /start: a greeting plus the whole of +// [HelpMessage], because a new user pressing Start has no other instructions. +func StartMessage(username string) string { + return "👋 Olá! Eu consulto CEP.\n\n" + HelpMessage(username) +} + +// UsageMessage is the answer to a bare /cep with no argument. +func UsageMessage(username string) string { + return "Use /cep 01001000.\n" + + "Em qualquer conversa também dá para usar @" + esc(username) + + " 01001000." +} + +// UnknownMessage is the answer to text in a private chat that is not a CEP and +// not a command. It is only ever sent in private: answering unrecognised text +// in a group would make the bot a nuisance. +const UnknownMessage = "Não entendi. Mande um CEP (01001000) " + + "ou use /help." + +func esc(text string) string { return html.EscapeString(text) } diff --git a/internal/bot/format_test.go b/internal/bot/format_test.go new file mode 100644 index 0000000..56b8dda --- /dev/null +++ b/internal/bot/format_test.go @@ -0,0 +1,126 @@ +package bot + +import ( + "encoding/json" + "errors" + "strings" + "testing" + + cepis "cep.is/go" +) + +func decodeAddress(t *testing.T, body string) *cepis.Address { + t.Helper() + var address cepis.Address + if err := json.Unmarshal([]byte(body), &address); err != nil { + t.Fatalf("decode: %v", err) + } + return &address +} + +const streetJSON = `{"cep":"01001000","logradouro":"Praça da Sé","complemento":"lado ímpar", + "unidade":"","bairro":"Sé","localidade":"São Paulo","uf":"SP","estado":"São Paulo", + "regiao":"Sudeste","ibge":"3550308","gia":"1004","ddd":"11","siafi":"7107", + "geo":{"lat":-23.567202,"lon":-46.649659,"raio_m":128,"precisao":"logradouro"}}` + +const municipalityJSON = `{"cep":"76200000","logradouro":"","complemento":"","unidade":"", + "bairro":"","localidade":"Iporá","uf":"GO","estado":"Goiás","regiao":"Centro-Oeste", + "ibge":"5210000","gia":"","ddd":"64","siafi":"9439", + "geo":{"lat":-16.441,"lon":-51.118,"raio_m":18400,"precisao":"localidade"}}` + +func TestFormatStreetLevelCEP(t *testing.T) { + text := FormatAddress(decodeAddress(t, streetJSON)) + + for _, needed := range []string{"01001-000", "Praça da Sé", "lado ímpar", "Sé — São Paulo/SP", "DDD 11", "128 m"} { + if !strings.Contains(text, needed) { + t.Fatalf("reply is missing %q:\n%s", needed, text) + } + } + if !strings.Contains(text, "openstreetmap.org/?mlat=-23.567202&mlon=-46.649659") { + t.Fatalf("a street-level CEP should link to the point:\n%s", text) + } +} + +func TestFormatMunicipalityCEP(t *testing.T) { + text := FormatAddress(decodeAddress(t, municipalityJSON)) + + if strings.Contains(text, "openstreetmap.org") { + t.Fatalf("a municipality CEP must not offer a pin:\n%s", text) + } + for _, needed := range []string{"CEP geral de município", "centroide de Iporá", "18,4 km", "não serve", "Iporá/GO"} { + if !strings.Contains(strings.ToLower(text), strings.ToLower(needed)) { + t.Fatalf("reply is missing %q:\n%s", needed, text) + } + } +} + +func TestFormatDiffersByPrecision(t *testing.T) { + street := FormatAddress(decodeAddress(t, streetJSON)) + municipality := FormatAddress(decodeAddress(t, municipalityJSON)) + if street == municipality { + t.Fatal("the two precisions produced the same reply") + } +} + +func TestFormatWithoutGeo(t *testing.T) { + text := FormatAddress(decodeAddress(t, `{"cep":"29902000","logradouro":"Rua X","localidade":"Linhares","uf":"ES"}`)) + if !strings.Contains(text, "Sem coordenada") { + t.Fatalf("reply should say there is no coordinate:\n%s", text) + } + if strings.Contains(text, "openstreetmap.org") { + t.Fatalf("no coordinate must mean no link:\n%s", text) + } +} + +func TestFormatEscapesHTML(t *testing.T) { + text := FormatAddress(decodeAddress(t, `{"cep":"12345678","logradouro":"Rua Fake & Cia","localidade":"X","uf":"SP"}`)) + if strings.Contains(text, "Fake") { + t.Fatalf("address data must be escaped:\n%s", text) + } + if !strings.Contains(text, "<b>Fake</b> & Cia") { + t.Fatalf("escaping is wrong:\n%s", text) + } +} + +func TestInlineDescriptionWarnsAboutMunicipality(t *testing.T) { + description := InlineDescription(decodeAddress(t, municipalityJSON)) + if !strings.Contains(description, "CEP geral do município") { + t.Fatalf("description = %q", description) + } + if street := InlineDescription(decodeAddress(t, streetJSON)); strings.Contains(street, "geral do município") { + t.Fatalf("street description = %q", street) + } +} + +func TestErrorMessagesAreDistinct(t *testing.T) { + cases := map[string]error{ + "not found": cepis.ErrCEPNotFound, + "invalid": cepis.ErrInvalidCEP, + "rate limited": cepis.ErrRateLimited, + "server": cepis.ErrServer, + "network": cepis.ErrNetwork, + "invalid response": cepis.ErrInvalidResponse, + "unknown": errors.New("something else"), + } + + seen := make(map[string]string, len(cases)) + for name, err := range cases { + message := ErrorMessage(err) + if message == "" { + t.Fatalf("%s produced no message", name) + } + if other, clash := seen[message]; clash { + t.Fatalf("%s and %s share a message: %q", name, other, message) + } + seen[message] = name + } +} + +func TestHelpMentionsEveryEntryPoint(t *testing.T) { + text := HelpMessage("cepisbot") + for _, needed := range []string{"01001-000", "/cep 01001000", "@cepisbot 01001000", "https://cep.is"} { + if !strings.Contains(text, needed) { + t.Fatalf("help is missing %q:\n%s", needed, text) + } + } +} diff --git a/internal/bot/ratelimit.go b/internal/bot/ratelimit.go new file mode 100644 index 0000000..b5333fd --- /dev/null +++ b/internal/bot/ratelimit.go @@ -0,0 +1,139 @@ +package bot + +import ( + "sync" + "time" +) + +// The public bot answers anyone, so one person with a script can spend the +// whole service's budget against the cep.is API. These constants bound what a +// single Telegram account may cost. +// +// Burst is what a human actually does: paste a handful of CEPs in a row, then +// think. Refill is the sustained rate afterwards — 20 lookups a minute, which +// no person reaches by hand and which a loop hits within a second. +// +// WarnInterval keeps the refusal from becoming its own flood: someone held at +// the limit is told once, then ignored until the window passes, so a script +// hammering the bot does not turn into the bot hammering Telegram. +const ( + Burst = 5 + Refill = 3 * time.Second + WarnInterval = 30 * time.Second +) + +// The bucket table is swept rather than kept forever: every distinct user gets +// an entry, and a public bot meets a lot of users. An entry is dropped once it +// has been idle long enough that it would have refilled to full anyway, at +// which point forgetting it and recreating it are the same thing. +const ( + bucketIdleTTL = 15 * time.Minute + sweepEvery = 5 * time.Minute +) + +// Decision is the answer of [Limiter.Allow]. +type Decision struct { + // Allowed reports whether the request may proceed. + Allowed bool + + // Warn reports whether the user should be told they are being limited. + // It is true at most once per [WarnInterval] per key. + Warn bool + + // RetryAfter is roughly how long until the next token, meant for the + // message shown to the user. + RetryAfter time.Duration +} + +type bucket struct { + tokens float64 + lastSeen time.Time + lastWarn time.Time +} + +// Limiter is a token bucket per key, where the key is the Telegram user id — or +// the chat id for channel posts, which carry no author. +// +// It is in-memory and per-process on purpose: the bot runs as a single +// container, and a shared store would add a dependency and a failure mode to +// protect against something a single process can already see. +type Limiter struct { + burst float64 + refill time.Duration + warnInterval time.Duration + now func() time.Time + + mu sync.Mutex + buckets map[int64]*bucket + lastSweep time.Time +} + +// NewLimiter builds a limiter with the package constants. +func NewLimiter() *Limiter { + return NewLimiterWith(Burst, Refill, WarnInterval, time.Now) +} + +// NewLimiterWith builds a limiter with explicit parameters and an injectable +// clock, which is how the test suite exercises refilling without waiting for it. +func NewLimiterWith(burst int, refill, warnInterval time.Duration, now func() time.Time) *Limiter { + if now == nil { + now = time.Now + } + return &Limiter{ + burst: float64(burst), + refill: refill, + warnInterval: warnInterval, + now: now, + buckets: make(map[int64]*bucket), + } +} + +// Allow takes one token for key and reports what the caller should do. +func (l *Limiter) Allow(key int64) Decision { + now := l.now() + + l.mu.Lock() + defer l.mu.Unlock() + + l.sweep(now) + + b, ok := l.buckets[key] + if !ok { + b = &bucket{tokens: l.burst} + l.buckets[key] = b + } else if l.refill > 0 { + b.tokens += float64(now.Sub(b.lastSeen)) / float64(l.refill) + if b.tokens > l.burst { + b.tokens = l.burst + } + } + b.lastSeen = now + + if b.tokens >= 1 { + b.tokens-- + return Decision{Allowed: true} + } + + decision := Decision{RetryAfter: time.Duration((1 - b.tokens) * float64(l.refill))} + if b.lastWarn.IsZero() || now.Sub(b.lastWarn) >= l.warnInterval { + b.lastWarn = now + decision.Warn = true + } + return decision +} + +func (l *Limiter) sweep(now time.Time) { + if l.lastSweep.IsZero() { + l.lastSweep = now + return + } + if now.Sub(l.lastSweep) < sweepEvery { + return + } + l.lastSweep = now + for key, b := range l.buckets { + if now.Sub(b.lastSeen) >= bucketIdleTTL { + delete(l.buckets, key) + } + } +} diff --git a/internal/bot/ratelimit_test.go b/internal/bot/ratelimit_test.go new file mode 100644 index 0000000..7f09787 --- /dev/null +++ b/internal/bot/ratelimit_test.go @@ -0,0 +1,128 @@ +package bot + +import ( + "sync" + "testing" + "time" +) + +type fakeClock struct { + mu sync.Mutex + now time.Time +} + +func (c *fakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +func (c *fakeClock) Advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.now = c.now.Add(d) +} + +func TestLimiterAllowsTheBurstThenStops(t *testing.T) { + clock := &fakeClock{now: time.Unix(1_700_000_000, 0)} + limiter := NewLimiterWith(3, time.Second, 30*time.Second, clock.Now) + + for i := 0; i < 3; i++ { + if decision := limiter.Allow(1); !decision.Allowed { + t.Fatalf("request %d was refused inside the burst", i+1) + } + } + decision := limiter.Allow(1) + if decision.Allowed { + t.Fatal("the fourth request should be refused") + } + if !decision.Warn { + t.Fatal("the first refusal should warn the user") + } + if decision.RetryAfter <= 0 { + t.Fatalf("retry after = %v, want a positive hint", decision.RetryAfter) + } +} + +func TestLimiterWarnsOncePerWindow(t *testing.T) { + clock := &fakeClock{now: time.Unix(1_700_000_000, 0)} + limiter := NewLimiterWith(1, time.Minute, 30*time.Second, clock.Now) + + limiter.Allow(1) + if !limiter.Allow(1).Warn { + t.Fatal("the first refusal should warn") + } + if limiter.Allow(1).Warn { + t.Fatal("a second refusal inside the window should stay silent") + } + + clock.Advance(31 * time.Second) + if !limiter.Allow(1).Warn { + t.Fatal("a refusal after the window should warn again") + } +} + +func TestLimiterRefills(t *testing.T) { + clock := &fakeClock{now: time.Unix(1_700_000_000, 0)} + limiter := NewLimiterWith(2, time.Second, time.Minute, clock.Now) + + limiter.Allow(1) + limiter.Allow(1) + if limiter.Allow(1).Allowed { + t.Fatal("the bucket should be empty") + } + + clock.Advance(time.Second) + if !limiter.Allow(1).Allowed { + t.Fatal("one second should have bought one token") + } + if limiter.Allow(1).Allowed { + t.Fatal("only one token should have been refilled") + } +} + +func TestLimiterNeverExceedsTheBurst(t *testing.T) { + clock := &fakeClock{now: time.Unix(1_700_000_000, 0)} + limiter := NewLimiterWith(2, time.Second, time.Minute, clock.Now) + + limiter.Allow(1) + clock.Advance(time.Hour) + + if !limiter.Allow(1).Allowed || !limiter.Allow(1).Allowed { + t.Fatal("the bucket should be full again") + } + if limiter.Allow(1).Allowed { + t.Fatal("an idle hour must not bank more than the burst") + } +} + +func TestLimiterKeysAreIndependent(t *testing.T) { + clock := &fakeClock{now: time.Unix(1_700_000_000, 0)} + limiter := NewLimiterWith(1, time.Minute, time.Minute, clock.Now) + + if !limiter.Allow(1).Allowed { + t.Fatal("first user refused") + } + if limiter.Allow(1).Allowed { + t.Fatal("first user should now be limited") + } + if !limiter.Allow(2).Allowed { + t.Fatal("one user's flood must not silence another") + } +} + +func TestLimiterIsConcurrencySafe(t *testing.T) { + limiter := NewLimiterWith(100, time.Millisecond, time.Second, nil) + + var wg sync.WaitGroup + for i := 0; i < 50; i++ { + wg.Add(1) + go func(key int64) { + defer wg.Done() + for j := 0; j < 20; j++ { + limiter.Allow(key % 5) + } + }(int64(i)) + } + wg.Wait() +} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..b173dbf --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,153 @@ +// Package config loads the bot's settings from the process environment. +// +// The environment is the only source: there is no configuration file and no +// command-line flag. That keeps the token out of the repository and out of the +// image, and it is what the container platform hands the process anyway. +package config + +import ( + "errors" + "fmt" + "log/slog" + "os" + "strings" + "time" +) + +// The environment variables read by [Load]. +const ( + // EnvToken holds the token issued by @BotFather. Required; [Load] fails + // when it is missing or blank. + EnvToken = "TELEGRAM_BOT_TOKEN" + + // EnvCepisBaseURL points the lookups at another instance of the API, + // typically a local one during development. + EnvCepisBaseURL = "CEPIS_BASE_URL" + + // EnvTelegramAPIBaseURL points the bot at another Bot API host. It + // exists for local Bot API servers and for the test suite, which stands + // up a fake Telegram in-process. + EnvTelegramAPIBaseURL = "TELEGRAM_API_BASE_URL" + + // EnvLogLevel selects the minimum level written to stdout: debug, info, + // warn or error. + EnvLogLevel = "LOG_LEVEL" +) + +// ErrMissingToken is returned by [Load] when [EnvToken] is absent or blank. +var ErrMissingToken = errors.New("config: " + EnvToken + " is not set") + +// Config is the fully resolved configuration of one bot process. +// +// Only Token, CepisBaseURL, TelegramAPIBaseURL and LogLevel come from the +// environment. The remaining fields are tuning that has no reason to differ +// between deployments; they are exported so the test suite can shorten the +// waits it would otherwise sit through. +type Config struct { + // Token is the @BotFather token. It is never logged: [Config.LogValue] + // redacts it, and the Telegram client scrubs it out of error messages. + Token string + + // CepisBaseURL is where address lookups go. + CepisBaseURL string + + // TelegramAPIBaseURL is the Bot API host, without a trailing slash and + // without the /bot prefix. + TelegramAPIBaseURL string + + // PollTimeout is the long-polling window asked of getUpdates. Telegram + // holds the request open for this long when there is nothing to send, + // which is what makes polling cheap. + PollTimeout time.Duration + + // RequestTimeout caps a single non-polling HTTP request. + RequestTimeout time.Duration + + // MaxAttempts is how many times a transient failure is retried, the + // first attempt included. + MaxAttempts int + + // BackoffBase and BackoffMax bound the jittered wait between attempts. + BackoffBase time.Duration + BackoffMax time.Duration + + // ShutdownGrace is how long an in-flight update may take to finish + // after a termination signal, and how long the final offset + // confirmation may take. + ShutdownGrace time.Duration + + // LogLevel is the minimum level written to stdout. + LogLevel slog.Level +} + +// Defaults returns the configuration with everything filled in except Token, +// which has no safe default. +func Defaults() Config { + return Config{ + CepisBaseURL: "https://cep.is", + TelegramAPIBaseURL: "https://api.telegram.org", + PollTimeout: 30 * time.Second, + RequestTimeout: 10 * time.Second, + MaxAttempts: 3, + BackoffBase: 500 * time.Millisecond, + BackoffMax: 8 * time.Second, + ShutdownGrace: 10 * time.Second, + LogLevel: slog.LevelInfo, + } +} + +// Load reads the environment on top of [Defaults]. +// +// It fails with [ErrMissingToken] when the token is absent, before any network +// is touched, so a misconfigured deployment dies at start-up rather than +// looking healthy while answering nobody. +func Load() (Config, error) { + cfg := Defaults() + + cfg.Token = strings.TrimSpace(os.Getenv(EnvToken)) + if cfg.Token == "" { + return Config{}, ErrMissingToken + } + + if value := strings.TrimSpace(os.Getenv(EnvCepisBaseURL)); value != "" { + cfg.CepisBaseURL = strings.TrimRight(value, "/") + } + if value := strings.TrimSpace(os.Getenv(EnvTelegramAPIBaseURL)); value != "" { + cfg.TelegramAPIBaseURL = strings.TrimRight(value, "/") + } + if value := strings.TrimSpace(os.Getenv(EnvLogLevel)); value != "" { + level, err := parseLevel(value) + if err != nil { + return Config{}, err + } + cfg.LogLevel = level + } + return cfg, nil +} + +// LogValue implements [log/slog.LogValuer] so that logging the whole +// configuration cannot leak the token, however carelessly it is done. +func (c Config) LogValue() slog.Value { + return slog.GroupValue( + slog.String("cepis_base_url", c.CepisBaseURL), + slog.String("telegram_api_base_url", c.TelegramAPIBaseURL), + slog.Duration("poll_timeout", c.PollTimeout), + slog.Int("max_attempts", c.MaxAttempts), + slog.String("log_level", c.LogLevel.String()), + slog.Bool("token_set", c.Token != ""), + ) +} + +func parseLevel(value string) (slog.Level, error) { + switch strings.ToLower(value) { + case "debug": + return slog.LevelDebug, nil + case "info": + return slog.LevelInfo, nil + case "warn", "warning": + return slog.LevelWarn, nil + case "error": + return slog.LevelError, nil + } + return 0, fmt.Errorf("config: %s must be debug, info, warn or error, got %q", EnvLogLevel, value) +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..7ae0d95 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,76 @@ +package config + +import ( + "errors" + "log/slog" + "strings" + "testing" +) + +func TestLoadRequiresTheToken(t *testing.T) { + t.Setenv(EnvToken, "") + if _, err := Load(); !errors.Is(err, ErrMissingToken) { + t.Fatalf("Load error = %v, want ErrMissingToken", err) + } +} + +func TestLoadDefaults(t *testing.T) { + t.Setenv(EnvToken, "424242:not-a-real-token") + + cfg, err := Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + if cfg.CepisBaseURL != "https://cep.is" { + t.Fatalf("base URL = %q", cfg.CepisBaseURL) + } + if cfg.TelegramAPIBaseURL != "https://api.telegram.org" { + t.Fatalf("telegram base URL = %q", cfg.TelegramAPIBaseURL) + } + if cfg.LogLevel != slog.LevelInfo { + t.Fatalf("log level = %v", cfg.LogLevel) + } +} + +func TestLoadOverrides(t *testing.T) { + t.Setenv(EnvToken, "424242:not-a-real-token") + t.Setenv(EnvCepisBaseURL, "http://localhost:8080/") + t.Setenv(EnvTelegramAPIBaseURL, "http://127.0.0.1:9000/") + t.Setenv(EnvLogLevel, "debug") + + cfg, err := Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + if cfg.CepisBaseURL != "http://localhost:8080" { + t.Fatalf("base URL = %q, the trailing slash should be gone", cfg.CepisBaseURL) + } + if cfg.TelegramAPIBaseURL != "http://127.0.0.1:9000" { + t.Fatalf("telegram base URL = %q", cfg.TelegramAPIBaseURL) + } + if cfg.LogLevel != slog.LevelDebug { + t.Fatalf("log level = %v", cfg.LogLevel) + } +} + +func TestLoadRejectsAnUnknownLogLevel(t *testing.T) { + t.Setenv(EnvToken, "424242:not-a-real-token") + t.Setenv(EnvLogLevel, "verbose") + + if _, err := Load(); err == nil { + t.Fatal("an unknown log level should be an error") + } +} + +func TestLogValueRedactsTheToken(t *testing.T) { + cfg := Defaults() + cfg.Token = "424242:not-a-real-token" + + rendered := cfg.LogValue().String() + if strings.Contains(rendered, cfg.Token) { + t.Fatalf("the token leaked into the log value: %s", rendered) + } + if !strings.Contains(rendered, "token_set=true") { + t.Fatalf("log value = %s", rendered) + } +} diff --git a/internal/telegram/client.go b/internal/telegram/client.go new file mode 100644 index 0000000..1b9be78 --- /dev/null +++ b/internal/telegram/client.go @@ -0,0 +1,359 @@ +// Package telegram is a small client for the Telegram Bot API, covering only +// the methods this bot calls: getMe, getUpdates, sendMessage and +// answerInlineQuery. +// +// The token appears in the request path, which makes every URL a secret. The +// client never puts a URL in an error or a log record, and [Client.Redact] +// scrubs the token out of anything that reaches it from the standard library, +// where a *url.Error carries the full URL by construction. +package telegram + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "math/rand" + "net/http" + "strings" + "time" +) + +// DefaultBaseURL is the public Bot API host. +const DefaultBaseURL = "https://api.telegram.org" + +const userAgent = "cepis-telegram/0.1.0 (+https://cep.is)" + +// Error is a failure reported by the Bot API itself, as opposed to a transport +// failure. Code is the Bot API error_code, which mirrors an HTTP status. +type Error struct { + Method string + Code int + Description string + RetryAfter time.Duration +} + +// Error implements the error interface. The token is never part of the message: +// nothing here is derived from the request URL. +func (e *Error) Error() string { + return fmt.Sprintf("telegram: %s failed with %d: %s", e.Method, e.Code, e.Description) +} + +// Permanent reports whether retrying the same call is pointless. A 400 means +// the request itself is wrong and will be wrong again; a 401 or 403 means the +// token or the chat permissions are, which no amount of waiting fixes. +func (e *Error) Permanent() bool { + switch e.Code { + case http.StatusBadRequest, http.StatusUnauthorized, http.StatusForbidden, http.StatusNotFound: + return true + } + return false +} + +// Client calls the Bot API over HTTP. It is safe for concurrent use. +type Client struct { + token string + baseURL string + httpClient *http.Client + timeout time.Duration + maxAttempts int + backoffBase time.Duration + backoffMax time.Duration + retryAfterCap time.Duration +} + +// Option configures a [Client]. +type Option func(*Client) error + +// WithBaseURL points the client at another Bot API host: a self-hosted Bot API +// server, or the fake one the test suite stands up. +func WithBaseURL(baseURL string) Option { + return func(c *Client) error { + trimmed := strings.TrimRight(strings.TrimSpace(baseURL), "/") + if trimmed == "" { + return errors.New("telegram: base URL must not be empty") + } + c.baseURL = trimmed + return nil + } +} + +// WithTimeout caps a single request, excluding long polling. [Client.GetUpdates] +// derives its own deadline from the polling window it asks for, because a +// 30-second long poll would otherwise trip a 10-second timeout every time. +func WithTimeout(timeout time.Duration) Option { + return func(c *Client) error { + if timeout <= 0 { + return errors.New("telegram: timeout must be > 0") + } + c.timeout = timeout + return nil + } +} + +// WithMaxAttempts sets how many attempts a call makes, the first included. +func WithMaxAttempts(attempts int) Option { + return func(c *Client) error { + if attempts < 1 { + return errors.New("telegram: max attempts must be >= 1") + } + c.maxAttempts = attempts + return nil + } +} + +// WithBackoff bounds the wait between attempts. Attempt n waits a random +// duration in [0, min(base*2^(n-1), max)] — full jitter, so a fleet of bots +// hitting the same rate limit does not come back in lockstep. +// +// It does not bound a retry_after sent by Telegram: that number is the server +// telling the client exactly how long it must wait, and shortening it only +// earns another 429. [WithRetryAfterCap] bounds that one. +func WithBackoff(base, max time.Duration) Option { + return func(c *Client) error { + if base < 0 || max < 0 { + return errors.New("telegram: backoff must not be negative") + } + c.backoffBase, c.backoffMax = base, max + return nil + } +} + +// WithRetryAfterCap bounds how long a Telegram-sent retry_after may park the +// caller. The value is honoured as received below the cap; the cap only exists +// so that an absurd number cannot stall the poll loop indefinitely. +func WithRetryAfterCap(limit time.Duration) Option { + return func(c *Client) error { + if limit <= 0 { + return errors.New("telegram: retry-after cap must be > 0") + } + c.retryAfterCap = limit + return nil + } +} + +// WithHTTPClient supplies the underlying HTTP client. +func WithHTTPClient(httpClient *http.Client) Option { + return func(c *Client) error { + if httpClient == nil { + return errors.New("telegram: http client must not be nil") + } + c.httpClient = httpClient + return nil + } +} + +// New builds a client for the given @BotFather token. +func New(token string, opts ...Option) (*Client, error) { + if strings.TrimSpace(token) == "" { + return nil, errors.New("telegram: token must not be empty") + } + c := &Client{ + token: strings.TrimSpace(token), + baseURL: DefaultBaseURL, + httpClient: &http.Client{}, + timeout: 10 * time.Second, + maxAttempts: 3, + backoffBase: 500 * time.Millisecond, + backoffMax: 8 * time.Second, + retryAfterCap: 60 * time.Second, + } + for _, opt := range opts { + if opt == nil { + continue + } + if err := opt(c); err != nil { + return nil, err + } + } + return c, nil +} + +// Redact replaces every occurrence of the token in a string with a placeholder. +// +// The Bot API puts the token in the path, so any error produced by net/http +// quotes it in full. Every log line and every error this package emits goes +// through here. +func (c *Client) Redact(text string) string { + if c.token == "" { + return text + } + return strings.ReplaceAll(text, c.token, "") +} + +// GetMe identifies the bot. It is called once at start-up: it validates the +// token before the first poll, and it yields the username needed to recognise +// the "/cep@username" form of a command. +func (c *Client) GetMe(ctx context.Context) (*User, error) { + var user User + if err := c.call(ctx, "getMe", nil, c.timeout, &user); err != nil { + return nil, err + } + return &user, nil +} + +// GetUpdates fetches the next batch of updates, confirming everything below +// req.Offset. +// +// The request deadline is the polling window plus a margin, so an idle long poll +// is not mistaken for a stalled request. Cancelling ctx aborts the poll; no +// update is lost by doing so, because updates are only confirmed by the offset +// of a later call. +func (c *Client) GetUpdates(ctx context.Context, req GetUpdatesRequest) ([]Update, error) { + deadline := c.timeout + time.Duration(req.Timeout)*time.Second + var updates []Update + if err := c.call(ctx, "getUpdates", req, deadline, &updates); err != nil { + return nil, err + } + return updates, nil +} + +// SendMessage posts a message to a chat. +func (c *Client) SendMessage(ctx context.Context, req SendMessageRequest) error { + return c.call(ctx, "sendMessage", req, c.timeout, nil) +} + +// AnswerInlineQuery answers an inline query. Telegram accepts exactly one +// answer per query and only for a few seconds after it arrives. +func (c *Client) AnswerInlineQuery(ctx context.Context, req AnswerInlineQueryRequest) error { + return c.call(ctx, "answerInlineQuery", req, c.timeout, nil) +} + +func (c *Client) call(ctx context.Context, method string, payload any, deadline time.Duration, out any) error { + var body []byte + if payload != nil { + encoded, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("telegram: cannot encode %s request: %w", method, err) + } + body = encoded + } + + var lastErr error + for attempt := 1; attempt <= c.maxAttempts; attempt++ { + if err := ctx.Err(); err != nil { + if lastErr == nil { + lastErr = err + } + return lastErr + } + + result, err := c.attempt(ctx, method, body, deadline) + if err == nil { + if out == nil || len(result) == 0 { + return nil + } + if err := json.Unmarshal(result, out); err != nil { + return fmt.Errorf("telegram: %s returned an undecodable result: %w", method, err) + } + return nil + } + + lastErr = err + var apiErr *Error + if errors.As(err, &apiErr) && apiErr.Permanent() { + return err + } + if attempt == c.maxAttempts { + break + } + if !sleep(ctx, c.wait(attempt, err)) { + return lastErr + } + } + return lastErr +} + +func (c *Client) attempt(ctx context.Context, method string, body []byte, deadline time.Duration) (json.RawMessage, error) { + ctx, cancel := context.WithTimeout(ctx, deadline) + defer cancel() + + target := c.baseURL + "/bot" + c.token + "/" + method + var reader io.Reader + if body != nil { + reader = bytes.NewReader(body) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, target, reader) + if err != nil { + return nil, c.transportError(method, err) + } + req.Header.Set("User-Agent", userAgent) + req.Header.Set("Accept", "application/json") + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, c.transportError(method, err) + } + defer resp.Body.Close() + + raw, err := io.ReadAll(resp.Body) + if err != nil { + return nil, c.transportError(method, err) + } + + var envelope Response + if err := json.Unmarshal(raw, &envelope); err != nil { + if resp.StatusCode >= 500 { + return nil, &Error{ + Method: method, + Code: resp.StatusCode, + Description: "unparseable body from an upstream error", + } + } + return nil, fmt.Errorf("telegram: %s returned an undecodable envelope: %w", method, err) + } + if envelope.OK { + return envelope.Result, nil + } + + apiErr := &Error{Method: method, Code: envelope.ErrorCode, Description: envelope.Description} + if apiErr.Code == 0 { + apiErr.Code = resp.StatusCode + } + if envelope.Parameters != nil && envelope.Parameters.RetryAfter > 0 { + apiErr.RetryAfter = time.Duration(envelope.Parameters.RetryAfter) * time.Second + } + return nil, apiErr +} + +func (c *Client) transportError(method string, err error) error { + return fmt.Errorf("telegram: %s could not be delivered: %s", method, c.Redact(err.Error())) +} + +func (c *Client) wait(attempt int, err error) time.Duration { + var apiErr *Error + if errors.As(err, &apiErr) && apiErr.RetryAfter > 0 { + if apiErr.RetryAfter > c.retryAfterCap { + return c.retryAfterCap + } + return apiErr.RetryAfter + } + ceiling := c.backoffBase << (attempt - 1) + if ceiling > c.backoffMax || ceiling <= 0 { + ceiling = c.backoffMax + } + if ceiling <= 0 { + return 0 + } + return time.Duration(rand.Int63n(int64(ceiling) + 1)) +} + +func sleep(ctx context.Context, wait time.Duration) bool { + if wait <= 0 { + return ctx.Err() == nil + } + timer := time.NewTimer(wait) + defer timer.Stop() + select { + case <-timer.C: + return true + case <-ctx.Done(): + return false + } +} diff --git a/internal/telegram/client_test.go b/internal/telegram/client_test.go new file mode 100644 index 0000000..60d6cd8 --- /dev/null +++ b/internal/telegram/client_test.go @@ -0,0 +1,209 @@ +package telegram + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" +) + +const testToken = "424242:not-a-real-token-generated-for-tests" + +func newTestClient(t *testing.T, handler http.HandlerFunc, opts ...Option) *Client { + t.Helper() + server := httptest.NewServer(handler) + t.Cleanup(server.Close) + + base := []Option{ + WithBaseURL(server.URL), + WithTimeout(2 * time.Second), + WithBackoff(time.Millisecond, 5*time.Millisecond), + } + client, err := New(testToken, append(base, opts...)...) + if err != nil { + t.Fatalf("New: %v", err) + } + return client +} + +func writeJSON(w http.ResponseWriter, status int, body string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + w.Write([]byte(body)) +} + +func TestGetMe(t *testing.T) { + client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + if !strings.HasSuffix(r.URL.Path, "/getMe") { + t.Errorf("path = %q", r.URL.Path) + } + writeJSON(w, 200, `{"ok":true,"result":{"id":1,"is_bot":true,"first_name":"cep.is","username":"cepisbot"}}`) + }) + + me, err := client.GetMe(context.Background()) + if err != nil { + t.Fatalf("GetMe: %v", err) + } + if me.Username != "cepisbot" { + t.Fatalf("username = %q", me.Username) + } +} + +func TestRetryAfterIsHonoured(t *testing.T) { + var calls int32 + client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + if atomic.AddInt32(&calls, 1) == 1 { + writeJSON(w, 429, `{"ok":false,"error_code":429,"description":"Too Many Requests","parameters":{"retry_after":1}}`) + return + } + writeJSON(w, 200, `{"ok":true,"result":{"message_id":1}}`) + }) + + started := time.Now() + if err := client.SendMessage(context.Background(), SendMessageRequest{ChatID: 1, Text: "oi"}); err != nil { + t.Fatalf("SendMessage: %v", err) + } + elapsed := time.Since(started) + if elapsed < time.Second { + t.Fatalf("waited %v, want at least the 1s asked for by retry_after", elapsed) + } + if atomic.LoadInt32(&calls) != 2 { + t.Fatalf("made %d calls, want 2", calls) + } +} + +func TestRetryAfterIsCapped(t *testing.T) { + var calls int32 + client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + if atomic.AddInt32(&calls, 1) == 1 { + writeJSON(w, 429, `{"ok":false,"error_code":429,"description":"slow down","parameters":{"retry_after":3600}}`) + return + } + writeJSON(w, 200, `{"ok":true,"result":true}`) + }, WithRetryAfterCap(20*time.Millisecond)) + + started := time.Now() + if err := client.SendMessage(context.Background(), SendMessageRequest{ChatID: 1, Text: "oi"}); err != nil { + t.Fatalf("SendMessage: %v", err) + } + if elapsed := time.Since(started); elapsed > time.Second { + t.Fatalf("an hour-long retry_after parked the caller for %v", elapsed) + } +} + +func TestBadRequestIsNotRetried(t *testing.T) { + var calls int32 + client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + writeJSON(w, 400, `{"ok":false,"error_code":400,"description":"Bad Request: chat not found"}`) + }) + + err := client.SendMessage(context.Background(), SendMessageRequest{ChatID: 1, Text: "oi"}) + if err == nil { + t.Fatal("a 400 should be an error") + } + if atomic.LoadInt32(&calls) != 1 { + t.Fatalf("made %d calls, want 1: a 400 is definitive", calls) + } +} + +func TestServerErrorIsRetried(t *testing.T) { + var calls int32 + client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + if atomic.AddInt32(&calls, 1) < 3 { + writeJSON(w, 502, `bad gateway`) + return + } + writeJSON(w, 200, `{"ok":true,"result":true}`) + }) + + if err := client.SendMessage(context.Background(), SendMessageRequest{ChatID: 1, Text: "oi"}); err != nil { + t.Fatalf("SendMessage: %v", err) + } + if atomic.LoadInt32(&calls) != 3 { + t.Fatalf("made %d calls, want 3", calls) + } +} + +func TestTokenNeverLeaksIntoErrors(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + url := server.URL + server.Close() + + client, err := New(testToken, + WithBaseURL(url), + WithTimeout(200*time.Millisecond), + WithMaxAttempts(1), + ) + if err != nil { + t.Fatalf("New: %v", err) + } + + err = client.SendMessage(context.Background(), SendMessageRequest{ChatID: 1, Text: "oi"}) + if err == nil { + t.Fatal("a closed server should fail") + } + if strings.Contains(err.Error(), testToken) { + t.Fatalf("the token leaked into the error: %q", err.Error()) + } + if !strings.Contains(err.Error(), "") { + t.Fatalf("the error should show the redaction: %q", err.Error()) + } +} + +func TestGetUpdatesSendsOffsetAndAllowedUpdates(t *testing.T) { + var seen GetUpdatesRequest + client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + json.NewDecoder(r.Body).Decode(&seen) + writeJSON(w, 200, `{"ok":true,"result":[{"update_id":7,"message":{"message_id":1,"chat":{"id":5,"type":"private"},"text":"01001000"}}]}`) + }) + + updates, err := client.GetUpdates(context.Background(), GetUpdatesRequest{ + Offset: 12, + Timeout: 1, + AllowedUpdates: AllowedUpdates, + }) + if err != nil { + t.Fatalf("GetUpdates: %v", err) + } + if seen.Offset != 12 { + t.Fatalf("offset sent = %d, want 12", seen.Offset) + } + if len(seen.AllowedUpdates) != 3 { + t.Fatalf("allowed_updates = %v", seen.AllowedUpdates) + } + if len(updates) != 1 || updates[0].UpdateID != 7 { + t.Fatalf("updates = %+v", updates) + } +} + +func TestCancelledContextStopsTheCall(t *testing.T) { + client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + io.Copy(io.Discard, r.Body) + select { + case <-r.Context().Done(): + case <-time.After(5 * time.Second): + } + }) + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(20 * time.Millisecond) + cancel() + }() + + if _, err := client.GetUpdates(ctx, GetUpdatesRequest{Timeout: 1}); err == nil { + t.Fatal("a cancelled poll should return an error") + } +} + +func TestEmptyTokenIsRejected(t *testing.T) { + if _, err := New(" "); err == nil { + t.Fatal("an empty token should be rejected") + } +} diff --git a/internal/telegram/poller.go b/internal/telegram/poller.go new file mode 100644 index 0000000..46219dc --- /dev/null +++ b/internal/telegram/poller.go @@ -0,0 +1,198 @@ +package telegram + +import ( + "context" + "log/slog" + "math/rand" + "time" +) + +// Handler consumes one update. +// +// It is the seam between transport and behaviour: long polling calls it in a +// loop, and a webhook transport added later would call the same method for each +// decoded update, with no change to the bot itself. A handler is expected to +// deal with its own failures; a returned error is logged and the update is +// still confirmed, because an update that reliably fails would otherwise be +// redelivered forever. +type Handler interface { + Handle(ctx context.Context, update Update) error +} + +// Poller drives getUpdates in a loop and owns the offset. +// +// The offset is the only piece of state the bot keeps between calls, and it is +// not persisted anywhere: Telegram holds it. An update is confirmed by a later +// getUpdates carrying a higher offset, never by handling it, which is why +// [Poller.Run] sends one last confirming call on the way out. +type Poller struct { + client *Client + handler Handler + logger *slog.Logger + + pollTimeout time.Duration + grace time.Duration + errorBase time.Duration + errorMax time.Duration + + offset int64 +} + +// PollerOption configures a [Poller]. +type PollerOption func(*Poller) + +// WithPollTimeout sets the long-polling window asked of getUpdates. +func WithPollTimeout(timeout time.Duration) PollerOption { + return func(p *Poller) { p.pollTimeout = timeout } +} + +// WithShutdownGrace sets how long an update already being handled may run after +// the context is cancelled, and how long the final offset confirmation may +// take. +func WithShutdownGrace(grace time.Duration) PollerOption { + return func(p *Poller) { p.grace = grace } +} + +// WithErrorBackoff bounds the jittered wait after a failed poll. +func WithErrorBackoff(base, max time.Duration) PollerOption { + return func(p *Poller) { p.errorBase, p.errorMax = base, max } +} + +// WithLogger sets the logger. The default discards nothing but writes through +// the package default. +func WithLogger(logger *slog.Logger) PollerOption { + return func(p *Poller) { + if logger != nil { + p.logger = logger + } + } +} + +// NewPoller builds a poller over a client and a handler. +func NewPoller(client *Client, handler Handler, opts ...PollerOption) *Poller { + p := &Poller{ + client: client, + handler: handler, + logger: slog.Default(), + pollTimeout: 30 * time.Second, + grace: 10 * time.Second, + errorBase: time.Second, + errorMax: 30 * time.Second, + } + for _, opt := range opts { + opt(p) + } + return p +} + +// Offset reports the identifier of the next update expected, which is one past +// the last update handled. It is meant for tests and diagnostics. +func (p *Poller) Offset() int64 { return p.offset } + +// Run polls until ctx is cancelled, then shuts down without losing the offset. +// +// Cancelling ctx aborts an idle long poll immediately. An update already being +// handled is finished first, under a fresh context bounded by the shutdown +// grace, so a cancelled poll cannot leave a user with a half-sent answer. Once +// the loop stops, one final getUpdates confirms everything handled: without it +// the last batch would be redelivered on the next start and every one of those +// users would be answered twice. +func (p *Poller) Run(ctx context.Context) error { + failures := 0 + for ctx.Err() == nil { + updates, err := p.client.GetUpdates(ctx, GetUpdatesRequest{ + Offset: p.offset, + Timeout: int(p.pollTimeout / time.Second), + AllowedUpdates: AllowedUpdates, + }) + if err != nil { + if ctx.Err() != nil { + break + } + failures++ + p.logger.Warn("poll failed", + slog.String("error", p.client.Redact(err.Error())), + slog.Int("consecutive_failures", failures)) + if !sleep(ctx, jitter(p.errorBase, p.errorMax, failures)) { + break + } + continue + } + failures = 0 + + for _, update := range updates { + p.dispatch(ctx, update) + p.offset = update.UpdateID + 1 + if ctx.Err() != nil { + break + } + } + } + return p.confirm() +} + +func (p *Poller) dispatch(ctx context.Context, update Update) { + workCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), p.grace) + defer cancel() + + started := time.Now() + err := p.handler.Handle(workCtx, update) + attrs := []any{ + slog.Int64("update_id", update.UpdateID), + slog.String("kind", kindOf(update)), + slog.Duration("took", time.Since(started)), + } + if err != nil { + p.logger.Error("update failed", append(attrs, slog.String("error", p.client.Redact(err.Error())))...) + return + } + p.logger.Info("update handled", attrs...) +} + +func (p *Poller) confirm() error { + if p.offset == 0 { + return nil + } + ctx, cancel := context.WithTimeout(context.Background(), p.grace) + defer cancel() + + if _, err := p.client.GetUpdates(ctx, GetUpdatesRequest{ + Offset: p.offset, + Limit: 1, + Timeout: 0, + AllowedUpdates: AllowedUpdates, + }); err != nil { + p.logger.Error("could not confirm the offset on shutdown, the last batch will be redelivered", + slog.Int64("offset", p.offset), + slog.String("error", p.client.Redact(err.Error()))) + return err + } + p.logger.Info("offset confirmed on shutdown", slog.Int64("offset", p.offset)) + return nil +} + +func kindOf(update Update) string { + switch { + case update.InlineQuery != nil: + return "inline_query" + case update.ChannelPost != nil: + return "channel_post" + case update.Message != nil: + return "message" + } + return "other" +} + +func jitter(base, max time.Duration, attempt int) time.Duration { + if base <= 0 { + return 0 + } + ceiling := base << (attempt - 1) + if ceiling > max || ceiling <= 0 { + ceiling = max + } + if ceiling <= 0 { + return 0 + } + return time.Duration(rand.Int63n(int64(ceiling) + 1)) +} diff --git a/internal/telegram/poller_test.go b/internal/telegram/poller_test.go new file mode 100644 index 0000000..4d4ccb4 --- /dev/null +++ b/internal/telegram/poller_test.go @@ -0,0 +1,247 @@ +package telegram + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +type recordingHandler struct { + mu sync.Mutex + handled []int64 + block chan struct{} + started chan struct{} + finished chan struct{} +} + +func (h *recordingHandler) Handle(ctx context.Context, update Update) error { + if h.started != nil { + select { + case h.started <- struct{}{}: + default: + } + } + if h.block != nil { + <-h.block + } + h.mu.Lock() + h.handled = append(h.handled, update.UpdateID) + h.mu.Unlock() + if h.finished != nil { + select { + case h.finished <- struct{}{}: + default: + } + } + return nil +} + +func (h *recordingHandler) seen() []int64 { + h.mu.Lock() + defer h.mu.Unlock() + return append([]int64(nil), h.handled...) +} + +type updateServer struct { + mu sync.Mutex + batches [][]Update + offsets []int64 +} + +func (s *updateServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if !strings.HasSuffix(r.URL.Path, "/getUpdates") { + writeJSON(w, 200, `{"ok":true,"result":true}`) + return + } + + var req GetUpdatesRequest + json.NewDecoder(r.Body).Decode(&req) + + s.mu.Lock() + s.offsets = append(s.offsets, req.Offset) + var batch []Update + if len(s.batches) > 0 { + batch = s.batches[0] + s.batches = s.batches[1:] + } + s.mu.Unlock() + + if len(batch) == 0 { + select { + case <-r.Context().Done(): + case <-time.After(50 * time.Millisecond): + } + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"ok": true, "result": batch}) +} + +func (s *updateServer) seenOffsets() []int64 { + s.mu.Lock() + defer s.mu.Unlock() + return append([]int64(nil), s.offsets...) +} + +func message(id int64) Update { + return Update{ + UpdateID: id, + Message: &Message{MessageID: id, Chat: Chat{ID: 1, Type: "private"}, Text: "01001000"}, + } +} + +func newPoller(t *testing.T, source *updateServer, handler Handler, opts ...PollerOption) *Poller { + t.Helper() + server := httptest.NewServer(source) + t.Cleanup(server.Close) + + client, err := New(testToken, + WithBaseURL(server.URL), + WithTimeout(2*time.Second), + WithBackoff(time.Millisecond, 5*time.Millisecond), + ) + if err != nil { + t.Fatalf("New: %v", err) + } + + base := []PollerOption{ + WithLogger(slog.New(slog.NewJSONHandler(io.Discard, nil))), + WithPollTimeout(0), + WithShutdownGrace(2 * time.Second), + WithErrorBackoff(time.Millisecond, 5*time.Millisecond), + } + return NewPoller(client, handler, append(base, opts...)...) +} + +func TestPollerAdvancesTheOffset(t *testing.T) { + source := &updateServer{batches: [][]Update{ + {message(4), message(5)}, + {message(9)}, + }} + handler := &recordingHandler{finished: make(chan struct{}, 8)} + poller := newPoller(t, source, handler) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- poller.Run(ctx) }() + + deadline := time.After(5 * time.Second) + for len(handler.seen()) < 3 { + select { + case <-handler.finished: + case <-deadline: + t.Fatalf("only handled %v", handler.seen()) + } + } + cancel() + <-done + + offsets := source.seenOffsets() + if offsets[0] != 0 { + t.Fatalf("first poll asked for %d, want 0", offsets[0]) + } + if offsets[1] != 6 { + t.Fatalf("second poll asked for %d, want 6", offsets[1]) + } + if last := offsets[len(offsets)-1]; last != 10 { + t.Fatalf("last poll asked for %d, want 10", last) + } + if poller.Offset() != 10 { + t.Fatalf("offset = %d, want 10", poller.Offset()) + } +} + +func TestPollerFinishesTheUpdateInFlightAndConfirmsIt(t *testing.T) { + source := &updateServer{batches: [][]Update{{message(3)}}} + handler := &recordingHandler{ + block: make(chan struct{}), + started: make(chan struct{}, 1), + finished: make(chan struct{}, 1), + } + poller := newPoller(t, source, handler) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- poller.Run(ctx) }() + + select { + case <-handler.started: + case <-time.After(5 * time.Second): + t.Fatal("the update was never dispatched") + } + cancel() + + select { + case <-done: + t.Fatal("Run returned while an update was still being handled") + case <-time.After(100 * time.Millisecond): + } + + close(handler.block) + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Run did not return after the update finished") + } + + if seen := handler.seen(); len(seen) != 1 || seen[0] != 3 { + t.Fatalf("handled %v, want [3]", seen) + } + offsets := source.seenOffsets() + if last := offsets[len(offsets)-1]; last != 4 { + t.Fatalf("shutdown confirmed offset %d, want 4", last) + } +} + +func TestPollerSurvivesAFailedPoll(t *testing.T) { + var calls int + var mu sync.Mutex + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + calls++ + current := calls + mu.Unlock() + if current <= 2 { + writeJSON(w, 500, `{"ok":false,"error_code":500,"description":"boom"}`) + return + } + writeJSON(w, 200, `{"ok":true,"result":[{"update_id":1,"message":{"message_id":1,"chat":{"id":1,"type":"private"},"text":"01001000"}}]}`) + })) + t.Cleanup(server.Close) + + client, err := New(testToken, + WithBaseURL(server.URL), + WithTimeout(time.Second), + WithMaxAttempts(1), + WithBackoff(time.Millisecond, 2*time.Millisecond), + ) + if err != nil { + t.Fatalf("New: %v", err) + } + + handler := &recordingHandler{finished: make(chan struct{}, 4)} + poller := NewPoller(client, handler, + WithLogger(slog.New(slog.NewJSONHandler(io.Discard, nil))), + WithPollTimeout(0), + WithShutdownGrace(time.Second), + WithErrorBackoff(time.Millisecond, 5*time.Millisecond), + ) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- poller.Run(ctx) }() + + select { + case <-handler.finished: + case <-time.After(5 * time.Second): + t.Fatal("the poller gave up after transient failures") + } + cancel() + <-done +} diff --git a/internal/telegram/types.go b/internal/telegram/types.go new file mode 100644 index 0000000..76f314c --- /dev/null +++ b/internal/telegram/types.go @@ -0,0 +1,152 @@ +package telegram + +import "encoding/json" + +// Update is one event delivered by getUpdates. +// +// Exactly one of the payload fields is set on any given update. Only the kinds +// this bot asks for in [AllowedUpdates] are ever populated; the rest of the Bot +// API's update kinds are decoded as nil and ignored. +type Update struct { + UpdateID int64 `json:"update_id"` + Message *Message `json:"message,omitempty"` + ChannelPost *Message `json:"channel_post,omitempty"` + InlineQuery *InlineQuery `json:"inline_query,omitempty"` +} + +// AllowedUpdates lists the update kinds the bot subscribes to. +// +// Narrowing the subscription is not only about bandwidth: it also means an +// update kind added to the Bot API later cannot arrive unannounced and consume +// an offset the bot never handled. +var AllowedUpdates = []string{"message", "channel_post", "inline_query"} + +// Message is a message or a channel post. +// +// From is nil for channel posts: a channel speaks as itself and there is no +// author to attribute the post to. Anything keyed by user must fall back to the +// chat in that case. +type Message struct { + MessageID int64 `json:"message_id"` + From *User `json:"from,omitempty"` + Chat Chat `json:"chat"` + Date int64 `json:"date"` + Text string `json:"text,omitempty"` + Entities []MessageEntity `json:"entities,omitempty"` +} + +// User is a Telegram account, human or bot. +type User struct { + ID int64 `json:"id"` + IsBot bool `json:"is_bot"` + FirstName string `json:"first_name"` + Username string `json:"username,omitempty"` +} + +// Chat is where a message was sent. Type is "private", "group", "supergroup" or +// "channel". +type Chat struct { + ID int64 `json:"id"` + Type string `json:"type"` + Title string `json:"title,omitempty"` + Username string `json:"username,omitempty"` +} + +// IsPrivate reports whether the chat is a one-to-one conversation with the bot. +// It is the only place where a bare CEP with no command is accepted. +func (c Chat) IsPrivate() bool { return c.Type == "private" } + +// MessageEntity is a span of special formatting inside a message's text. +type MessageEntity struct { + Type string `json:"type"` + Offset int `json:"offset"` + Length int `json:"length"` +} + +// InlineQuery is what arrives when someone types "@botname something" in any +// chat, including chats the bot is not a member of. +type InlineQuery struct { + ID string `json:"id"` + From User `json:"from"` + Query string `json:"query"` + Offset string `json:"offset"` +} + +// GetUpdatesRequest is the argument of getUpdates. +// +// Offset is the load-bearing field: it both selects the next update to return +// and confirms every update with a lower identifier, which is what stops +// Telegram from redelivering them. Sending the wrong value replays handled +// updates or silently drops unhandled ones. +type GetUpdatesRequest struct { + Offset int64 `json:"offset,omitempty"` + Limit int `json:"limit,omitempty"` + Timeout int `json:"timeout,omitempty"` + AllowedUpdates []string `json:"allowed_updates,omitempty"` +} + +// SendMessageRequest is the argument of sendMessage. +type SendMessageRequest struct { + ChatID int64 `json:"chat_id"` + Text string `json:"text"` + ParseMode string `json:"parse_mode,omitempty"` + LinkPreviewOptions *LinkPreviewOptions `json:"link_preview_options,omitempty"` + ReplyParameters *ReplyParameters `json:"reply_parameters,omitempty"` +} + +// LinkPreviewOptions controls the preview card Telegram appends to a message +// containing a link. +type LinkPreviewOptions struct { + IsDisabled bool `json:"is_disabled,omitempty"` +} + +// ReplyParameters attaches the answer to the message that asked for it, which +// is what keeps a busy group readable. +type ReplyParameters struct { + MessageID int64 `json:"message_id"` + AllowSendingWithoutReply bool `json:"allow_sending_without_reply,omitempty"` +} + +// AnswerInlineQueryRequest is the argument of answerInlineQuery. +// +// An inline query must be answered within seconds or Telegram discards it, and +// each query can only be answered once. +type AnswerInlineQueryRequest struct { + InlineQueryID string `json:"inline_query_id"` + Results []InlineQueryResult `json:"results"` + CacheTime int `json:"cache_time"` + IsPersonal bool `json:"is_personal,omitempty"` +} + +// InlineQueryResult is one entry of the list shown above the compose box. Only +// the "article" kind is produced by this bot. +type InlineQueryResult struct { + Type string `json:"type"` + ID string `json:"id"` + Title string `json:"title"` + Description string `json:"description,omitempty"` + InputMessageContent InputMessageContent `json:"input_message_content"` +} + +// InputMessageContent is the message that gets sent when a result is picked. +type InputMessageContent struct { + MessageText string `json:"message_text"` + ParseMode string `json:"parse_mode,omitempty"` + LinkPreviewOptions *LinkPreviewOptions `json:"link_preview_options,omitempty"` +} + +// Response is the envelope every Bot API method answers with. +type Response struct { + OK bool `json:"ok"` + Result json.RawMessage `json:"result,omitempty"` + ErrorCode int `json:"error_code,omitempty"` + Description string `json:"description,omitempty"` + Parameters *ResponseParameters `json:"parameters,omitempty"` +} + +// ResponseParameters carries the machine-readable part of a failure. RetryAfter +// is how many seconds Telegram wants to be left alone after a 429. +type ResponseParameters struct { + RetryAfter int `json:"retry_after,omitempty"` + MigrateToChatID int64 `json:"migrate_to_chat_id,omitempty"` +} diff --git a/main.go b/main.go new file mode 100644 index 0000000..36aecab --- /dev/null +++ b/main.go @@ -0,0 +1,50 @@ +// Command cepis-telegram is the official Telegram bot for https://cep.is. +// +// It answers a bare CEP in a private chat, "/cep " in groups and channels, +// and inline queries anywhere. Updates arrive by long polling, so the process +// needs no inbound port, no public hostname and no webhook secret. +// +// Configuration is environment-only; see the config package. +package main + +import ( + "context" + "errors" + "log/slog" + "os" + "os/signal" + "syscall" + + "cep.is/telegrambot/internal/app" + "cep.is/telegrambot/internal/config" +) + +func main() { + if err := run(); err != nil { + slog.New(slog.NewJSONHandler(os.Stdout, nil)).Error("fatal", slog.String("error", err.Error())) + os.Exit(1) + } +} + +func run() error { + cfg, err := config.Load() + if err != nil { + if errors.Is(err, config.ErrMissingToken) { + return errors.New(config.EnvToken + " is required: create a bot with @BotFather and pass its token in the environment") + } + return err + } + + logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: cfg.LogLevel})) + slog.SetDefault(logger) + logger.Info("configuration loaded", slog.Any("config", cfg)) + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + application, err := app.New(ctx, cfg, logger) + if err != nil { + return err + } + return application.Run(ctx) +} -- cgit v1.3