From d42372e99627b2cc13d0d2a17f43a5b7e371c71f Mon Sep 17 00:00:00 2001 From: claude Date: Thu, 6 Aug 2026 20:52:34 +0400 Subject: [PATCH] the poller reads one chat and answers in it (V-637) Long-poll getUpdates rather than a webhook: the box takes no inbound connections and reaches telegram through a relay, so the direction has to stay outbound. A failed poll waits and retries, because the relay going down is the normal cause and it comes back on its own. The backlog is discarded on start. Telegram holds undelivered updates for 24 hours, so a daemon that was down overnight would otherwise answer every question in order, and a reminder set from an eight-hour-old message lands at the wrong time. Missing it is the safe direction. ChatID is the only accepted sender and anything else is dropped without a reply, because a reply confirms the bot exists and whose it is. Chat ids are not guessable but they are not secret either, so that is the whole authorisation and it is an allowlist of one. --- internal/delivery/telegramsink/intake.go | 191 +++++++++++++++++++++++ 1 file changed, 191 insertions(+) create mode 100644 internal/delivery/telegramsink/intake.go diff --git a/internal/delivery/telegramsink/intake.go b/internal/delivery/telegramsink/intake.go new file mode 100644 index 0000000..8fbc21f --- /dev/null +++ b/internal/delivery/telegramsink/intake.go @@ -0,0 +1,191 @@ +// intake.go — the inbound half of the telegram channel (V-637). +// +// Until this file, telegram was a reach and nothing else: the sink pushes an +// away message and the chat has no way to answer. That made the correction +// gesture (V-630) reachable from the web and from voice only, and the sample of +// labels skews to wherever the owner happens to be standing. +// +// Long-poll getUpdates, not a webhook. The box takes no inbound connections and +// it reaches api.telegram.org through a relay, so the direction of the +// connection has to stay outbound. The poller is off unless the telegram block +// says intake, and it accepts messages from exactly one chat. +package telegramsink + +import ( + "context" + "errors" + "log" + "net/http" + "strings" + "time" +) + +// longPollSeconds — how long telegram holds an empty getUpdates open. The HTTP +// client's own timeout has to sit above it or every poll ends as a transport +// error, which is why the poller does not reuse the sink's client. +const longPollSeconds = 25 + +// pollBackoff — the wait after a failed poll. The relay going down is the +// normal cause and it comes back on its own, so this is a quiet retry rather +// than an escalation. +const pollBackoff = 15 * time.Second + +// Turn runs one utterance as a turn and reports the reply and the persisted +// trace id. traceID 0 means nothing persisted, and then the reply carries no +// correction buttons — there is no row for them to point at. +type Turn func(ctx context.Context, conversation, text string) (reply string, traceID int64, err error) + +// Correct records the owner's correction of one turn. shouldBe empty is the +// cheap half of the gesture: wrong, target unstated. +type Correct func(ctx context.Context, traceID int64, shouldBe string) error + +// Poller reads the configured chat and answers in it. One per daemon. +type Poller struct { + sink *Sink + turn Turn + correct Correct + hc *http.Client + offset int64 +} + +// NewPoller builds the intake half around an already-validated sink, so the +// token, the base URL and the relay are resolved in one place. turn is +// required; correct may be nil, and then the reply carries no buttons. +func NewPoller(s *Sink, turn Turn, correct Correct) (*Poller, error) { + if s == nil { + return nil, errors.New("telegramsink: intake needs a sink") + } + if turn == nil { + return nil, errors.New("telegramsink: intake needs a turn handler") + } + // The sink's transport already carries the relay. Only the timeout differs, + // and it has to clear the long poll. + hc := &http.Client{ + Timeout: (longPollSeconds + 10) * time.Second, + Transport: s.hc.Transport, + } + return &Poller{sink: s, turn: turn, correct: correct, hc: hc}, nil +} + +// Run polls until the context ends. It never returns an error: a chat that +// cannot be read is a degraded reach, not a reason to stop the daemon. +func (p *Poller) Run(ctx context.Context) { + p.discardBacklog(ctx) + log.Printf("telegram intake: reading chat %s", p.sink.cfg.ChatID) + for ctx.Err() == nil { + updates, err := p.getUpdates(ctx, longPollSeconds) + if err != nil { + if ctx.Err() != nil { + return + } + log.Printf("telegram intake: poll: %v", err) + select { + case <-ctx.Done(): + return + case <-time.After(pollBackoff): + } + continue + } + for _, u := range updates { + p.handle(ctx, u) + } + } +} + +// discardBacklog moves the offset past whatever is already queued, without +// acting on any of it. +// +// Telegram holds undelivered updates for 24 hours, so a daemon that was down +// overnight would otherwise wake up and answer every question in order. A +// question asked eight hours ago has been answered by the owner himself or has +// stopped mattering, and a reminder set from it would land at the wrong time. +// Missing it is the safe direction. +func (p *Poller) discardBacklog(ctx context.Context) { + updates, err := p.getUpdates(ctx, 0) + if err != nil { + // Not fatal. The offset stays 0, so the first real poll sees the backlog + // and the messages below get answered late. Say so rather than hide it. + log.Printf("telegram intake: could not skip the backlog, old messages may be answered: %v", err) + return + } + if len(updates) > 0 { + log.Printf("telegram intake: skipped %d message(s) queued while the daemon was down", len(updates)) + } +} + +// handle dispatches one update. Anything that is neither a message from the +// owner's chat nor a callback on one of Maven's own keyboards is dropped in +// silence: a reply to a stranger confirms the bot exists and who it belongs to. +func (p *Poller) handle(ctx context.Context, u update) { + switch { + case u.CallbackQuery != nil: + p.onCallback(ctx, u.CallbackQuery) + case u.Message != nil: + p.onMessage(ctx, u.Message) + } +} + +func (p *Poller) onMessage(ctx context.Context, m *message) { + text := strings.TrimSpace(m.Text) + if text == "" || !p.fromOwner(m.Chat.idString()) { + return + } + // The conversation id keys the dialogue, so a clarify question asked in the + // chat is not answered by an utterance typed on the web. + reply, traceID, err := p.turn(ctx, "telegram:"+m.Chat.idString(), text) + if err != nil { + log.Printf("telegram intake: turn: %v", err) + return + } + if strings.TrimSpace(reply) == "" { + return + } + if err := p.send(ctx, reply, p.correctionKeyboard(traceID)); err != nil { + log.Printf("telegram intake: reply: %v", err) + } +} + +// onCallback handles a tap on a correction button. Every path answers the +// callback: telegram spins a clock on the button until it is answered, and an +// unanswered tap reads as a gesture that was dropped. +func (p *Poller) onCallback(ctx context.Context, cb *callbackQuery) { + if !p.fromOwner(cb.Message.Chat.idString()) { + return + } + traceID, target, kind := parseCallback(cb.Data) + if kind == callbackUnknown || p.correct == nil { + p.answerCallback(ctx, cb.ID, "") + return + } + // A tap on "не то" only opens the second row. Nothing is written yet: the + // target is worth much more than the negative, so he gets the chance to name + // it before the gesture is spent. + if kind == callbackAskTarget { + p.answerCallback(ctx, cb.ID, "") + if err := p.editKeyboard(ctx, cb.Message.Chat.idString(), cb.Message.MessageID, targetKeyboard(traceID)); err != nil { + log.Printf("telegram intake: open the target row: %v", err) + } + return + } + if err := p.correct(ctx, traceID, target); err != nil { + log.Printf("telegram intake: correct turn %d: %v", traceID, err) + p.answerCallback(ctx, cb.ID, "не записалось") + return + } + p.answerCallback(ctx, cb.ID, "записала") + // The buttons come off, because the correction is given and a live keyboard + // on an answered turn invites correcting it twice. + if err := p.editKeyboard(ctx, cb.Message.Chat.idString(), cb.Message.MessageID, nil); err != nil { + log.Printf("telegram intake: clear the keyboard: %v", err) + } +} + +// fromOwner — one chat, and it is the one the sink already sends to. Telegram +// chat ids are not guessable, but they are also not secret: they travel in +// every forwarded message. So this is the whole authorisation and it is an +// allowlist of one. +func (p *Poller) fromOwner(chatID string) bool { + return chatID != "" && chatID == p.cfgChatID() +} + +func (p *Poller) cfgChatID() string { return strings.TrimSpace(p.sink.cfg.ChatID) }