diff options
| -rw-r--r-- | .dockerignore | 6 | ||||
| -rw-r--r-- | .gitignore | 6 | ||||
| -rw-r--r-- | Dockerfile | 16 | ||||
| -rw-r--r-- | LICENSE | 15 | ||||
| -rw-r--r-- | Makefile | 56 | ||||
| -rw-r--r-- | README.md | 99 | ||||
| -rw-r--r-- | go.mod | 5 | ||||
| -rw-r--r-- | go.sum | 2 | ||||
| -rw-r--r-- | internal/app/app.go | 96 | ||||
| -rw-r--r-- | internal/app/app_test.go | 510 | ||||
| -rw-r--r-- | internal/app/fakes_test.go | 285 | ||||
| -rw-r--r-- | internal/bot/bot.go | 288 | ||||
| -rw-r--r-- | internal/bot/command.go | 54 | ||||
| -rw-r--r-- | internal/bot/command_test.go | 59 | ||||
| -rw-r--r-- | internal/bot/format.go | 225 | ||||
| -rw-r--r-- | internal/bot/format_test.go | 126 | ||||
| -rw-r--r-- | internal/bot/ratelimit.go | 139 | ||||
| -rw-r--r-- | internal/bot/ratelimit_test.go | 128 | ||||
| -rw-r--r-- | internal/config/config.go | 153 | ||||
| -rw-r--r-- | internal/config/config_test.go | 76 | ||||
| -rw-r--r-- | internal/telegram/client.go | 359 | ||||
| -rw-r--r-- | internal/telegram/client_test.go | 209 | ||||
| -rw-r--r-- | internal/telegram/poller.go | 198 | ||||
| -rw-r--r-- | internal/telegram/poller_test.go | 247 | ||||
| -rw-r--r-- | internal/telegram/types.go | 152 | ||||
| -rw-r--r-- | main.go | 50 |
26 files changed, 3559 insertions, 0 deletions
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"] @@ -0,0 +1,15 @@ +ISC License + +Copyright (c) 2026 Murilo Ijanc <[email protected]> + +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). @@ -0,0 +1,5 @@ +module cep.is/telegrambot + +go 1.23 + +require cep.is/go v0.0.0-20260726172041-323d9a8422f0 @@ -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("📮 <b>" + esc(address.FormattedCEP()) + "</b>\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("<i>" + esc(line) + "</i>\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 "\n<i>Sem coordenada para este CEP na base do IBGE.</i>\n" + } + + if geo.IsPrecise() { + return "\n🗺 <a href=\"" + esc(mapURL(geo.Lat, geo.Lon)) + "\">Ver no mapa</a>\n" + + "<i>Precisão de logradouro — o ponto é o trecho de rua (raio aprox. " + + esc(formatRadius(geo.RaioM)) + ").</i>\n" + } + + return "\n⚠️ <b>CEP geral de município</b> — 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" + + "<i>" + esc(coordText(geo.Lat, geo.Lon)) + "</i>\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: <code>01001000</code> " + + "ou <code>01001-000</code>." + 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 "<b>cep.is</b> — consulta de CEP na base oficial dos Correios (DNE).\n\n" + + "<b>Aqui no privado</b>\n" + + "Mande o CEP e pronto: <code>01001000</code> ou <code>01001-000</code>.\n\n" + + "<b>Em grupos e canais</b>\n" + + "Use <code>/cep 01001000</code>. Se houver outros bots, " + + "<code>/cep" + esc(at) + " 01001000</code> também funciona.\n\n" + + "<b>Em qualquer conversa</b>\n" + + "Digite <code>" + esc(at) + " 01001000</code> 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 <code>/cep 01001000</code>.\n" + + "Em qualquer conversa também dá para usar <code>@" + esc(username) + + " 01001000</code>." +} + +// 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 (<code>01001000</code>) " + + "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 <b>Fake</b> & Cia","localidade":"X","uf":"SP"}`)) + if strings.Contains(text, "<b>Fake</b>") { + 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<token> 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, "<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, `<html>bad gateway</html>`) + 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(), "<token>") { + 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"` +} @@ -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 <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) +} |