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