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) }