Maven/internal/delivery/telegramsink/intake.go

// 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"
	"fmt"
	"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
}

// ValidateIntakeChatID refuses a chat id the intake half cannot use. The push
// half accepts @channelusername as a destination. The intake half cannot: an
// inbound update names its chat by numeric id, so an @-name would match nothing
// and the poller would read the chat and answer none of it. Config validation
// calls this, so the box refuses to boot rather than running a dead reach —
// NewPoller returning an error is too late, because the daemon is already up.
func ValidateIntakeChatID(chatID string) error {
	id := strings.TrimSpace(chatID)
	if id == "" {
		return errors.New("telegramsink: intake needs a chat id")
	}
	digits := strings.TrimPrefix(id, "-")
	if digits == "" || strings.TrimLeft(digits, "0123456789") != "" {
		return fmt.Errorf("telegramsink: intake needs the numeric chat id, not %s", chatID)
	}
	return nil
}

// 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")
	}
	if err := ValidateIntakeChatID(s.cfg.ChatID); err != nil {
		return nil, err
	}
	// 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) {
	// getUpdates returns at most 100 per call, so one call is not the queue. The
	// loop is bounded rather than "until empty": the timeout is 0, so an instance
	// that keeps handing back a full batch would spin, and a thousand skipped
	// messages is already a box that was down for a long time.
	skipped := 0
	for range 10 {
		updates, err := p.getUpdates(ctx, 0)
		if err != nil {
			// Not fatal. The offset stays where it was, so the first real poll sees
			// what is left and answers it late. Say so rather than hide it.
			log.Printf("telegram intake: could not skip the backlog, old messages may be answered: %v", err)
			return
		}
		skipped += len(updates)
		if len(updates) == 0 {
			break
		}
	}
	if skipped > 0 {
		log.Printf("telegram intake: skipped %d message(s) queued while the daemon was down", skipped)
	}
}

// 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 from the owner
// 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. A tap
// from anyone else gets silence, the same as a message from a stranger.
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) }