mesh-catalog/modules/telegram/cmd/telegram/channel.go

package main

// The Telegram channel: kind `telegram` of the `channel` and `intake` benches (novox/hq ADR 0259). It shows
// what the router sends and says what the operator chose, and nothing else: it never judges whether the
// sender is the operator (the router does, from its list), never calls another module's verb, and knows
// nothing of what an answer does.
//
//   - **Work** comes from its kind's worker on the `channel` queue: `show` (a message with buttons, each
//     carrying only the router's ticket), `send` (words), `edit` (a message shown before, by the router's
//     handle). Up to 4096 characters arrive whole; longer is cut and the cut said (D1). A warning or a
//     clearing arrives without sound, an urgent message or a question rings (D4). A rate limit is waited out
//     as told (D5); a refusal is a permanent failure, said to the router, never retried; an edit to the same
//     text is a success (D6). A message that cannot be delivered after three tries is said as failed.
//   - **What arrives** is long-polled: outbound only. A tapped button is acknowledged at once (the phone's
//     spinner stops) and said as a `choice` with the sender the service authenticated; only a private chat
//     with the account itself counts, and only on a message showing the words this channel put there (another
//     holder of the token may have changed them). `/start` is said as a `link`; six digits are asked as a `code` proof by
//     request and reply, never in an event, and deleted from the chat. Anything else is answered with one
//     fixed line.
//   - **Standing** is said every five minutes and whenever it changes: whether the token works, whether
//     the updates arrive, and that an edit here notifies nobody.

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"
	"unicode/utf8"

	"git.novox.be/novox/mesh-sdk/go/asks"
)

// Kind is this channel's kind on both benches.
const Kind = "telegram"

const (
	// MaxLength is what a message carries whole (Telegram's own bound).
	MaxLength = 4096
	// cutNote is said where a message was cut.
	cutNote = "\n… (cut here: the rest is in the mesh's history)"
	// Tries is how often a message is tried before it is said as failed.
	Tries = 3
	// PollWait is how long one long poll waits for something to happen.
	PollWait = 50 * time.Second
	// StandingEvery is how often the channel says whether it can send, at least.
	StandingEvery = 5 * time.Minute
	// FixedLine is what anything the bot does not take is answered with.
	FixedLine = "This bot carries the mesh's messages and takes their buttons. To link this account as the " +
		"operator, send /start."
)

var codeShape = regexp.MustCompile(`^\s*(\d{3})\s?(\d{3})\s*$`)

// Placed is where one of the router's messages is in the chat.
type Placed struct {
	Chat int64 `json:"chat"`
	ID   int64 `json:"id"`
}

// Store is this module's state on the bus: the update offset and where each message is.
type Store interface {
	Offset() (int64, error)
	SetOffset(int64) error
	Placed(handle string) (*Placed, error)
	Place(handle string, p Placed) error
	// Shown is the digest of the words a message in a chat shows as this channel last put them there (wordsDigest),
	// empty for one it never showed; Show keeps it.
	Shown(chat, id int64) (string, error)
	Show(chat, id int64, digest string) error
}

// wordsDigest is what a tap's message is compared by: the words as Telegram gives them back — its line
// endings and outer spaces aside — so a message whose words another holder of the token changed is told
// from the one this channel showed (the review of 2026-10-09).
func wordsDigest(words string) string {
	w := strings.TrimSpace(strings.ReplaceAll(words, "\r\n", "\n"))
	sum := sha256.Sum256([]byte(w))
	return hex.EncodeToString(sum[:])
}

// keepShown keeps what a message now shows; a message whose words could not be kept answers no tap.
func (c *Channel) keepShown(chat, id int64, words string) {
	if err := c.Store.Show(chat, id, wordsDigest(words)); err != nil {
		c.Logf("[telegram] what a message shows could not be kept, so its buttons will be refused: %v", err)
	}
}

// Channel is the channel's whole working.
type Channel struct {
	Bot     *Bot
	Store   Store
	Publish func(subject string, body any, id string) error
	Prove   func(subject string, body any) (json.RawMessage, error)
	Now     func() time.Time
	Logf    func(string, ...any)
	Sleep   func(time.Duration)

	mu        sync.Mutex
	replyTo   map[string]int64 // an arrival's id to the chat it came from, for the router's reply
	lastPoll  time.Time
	pollErr   string
	meAt      time.Time
	me        string
	meErr     string
	said      asks.Standing
	saidAt    time.Time
	lastSent  time.Time
	lastErr   string
	delivered int
	failed    int
}

func (c *Channel) init() {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.replyTo == nil {
		c.replyTo = map[string]int64{}
	}
	if c.Now == nil {
		c.Now = time.Now
	}
	if c.Logf == nil {
		c.Logf = func(string, ...any) {}
	}
	if c.Sleep == nil {
		c.Sleep = time.Sleep
	}
}

// text is a message as the chat shows it, cut at MaxLength and the cut said (D1).
func text(m asks.Message) string {
	t := m.Title
	if m.Body != "" {
		t += "\n" + m.Body
	}
	if utf8.RuneCountInString(t) <= MaxLength {
		return t
	}
	runes := []rune(t)
	keep := MaxLength - utf8.RuneCountInString(cutNote)
	return string(runes[:keep]) + cutNote
}

// buttons are a message's buttons as the API takes them: a label and the ticket, which must fit the 64
// bytes Telegram allows; one that does not is left out and said.
func (c *Channel) buttons(m asks.Message) [][2]string {
	var out [][2]string
	for _, b := range m.Buttons {
		if len(b.Ticket) == 0 || len(b.Ticket) > 64 {
			c.Logf("[telegram] a button %q carries a ticket Telegram cannot hold; left out", b.Label)
			continue
		}
		out = append(out, [2]string{b.Label, b.Ticket})
	}
	return out
}

// chatFor is where a message goes: the linked account the router names, else the chat of the arrival it
// replies to. Neither is a permanent failure: nobody is linked, and nobody asked.
func (c *Channel) chatFor(m asks.Message) (int64, error) {
	if m.To != "" {
		id, err := strconv.ParseInt(m.To, 10, 64)
		if err != nil {
			return 0, fmt.Errorf("the router names %q as the account, which is not a Telegram account's id", m.To)
		}
		return id, nil
	}
	c.mu.Lock()
	chat, ok := c.replyTo[m.Reply]
	c.mu.Unlock()
	if m.Reply != "" && ok {
		return chat, nil
	}
	return 0, errors.New("no chat to send to: no account is linked as the operator on Telegram")
}

// Take does one piece of the channel's work. It returns an error only to have the work offered again:
// a message delivered, or said as failed, is taken.
func (c *Channel) Take(subject string, body json.RawMessage) error {
	c.init()
	parts := strings.Split(subject, ".")
	if len(parts) != 6 || parts[5] != Kind {
		c.Logf("[telegram] work on %s is not this kind's; taken and dropped", subject)
		return nil
	}
	verb := parts[4]
	var m asks.Message
	if err := json.Unmarshal(body, &m); err != nil || m.Handle == "" {
		c.Logf("[telegram] work on %s could not be read; taken and dropped", subject)
		return nil
	}
	chat, err := c.chatFor(m)
	if err != nil {
		c.fail(m.Handle, err, true)
		return nil
	}
	words, buttons := text(m), c.buttons(m)
	silent := m.Silent || (!m.Urgent && len(buttons) == 0)
	var place *Placed
	if verb == asks.Edit {
		if place, err = c.Store.Placed(m.Handle); err != nil {
			return err // the state is away: offered again
		}
	}
	for try := 1; ; try++ {
		var id int64
		switch {
		case place != nil:
			err = c.Bot.Edit(place.Chat, place.ID, words, buttons)
			if err == nil || notModified(err) {
				c.keepShown(place.Chat, place.ID, words)
			}
			if gone(err) {
				place, err = nil, c.newMessage(m.Handle, chat, words, buttons, silent, &id)
			}
		default:
			err = c.newMessage(m.Handle, chat, words, buttons, silent, &id)
		}
		if err == nil {
			c.mu.Lock()
			c.lastSent, c.lastErr = c.Now(), ""
			c.delivered++
			c.mu.Unlock()
			return nil
		}
		if wait := retryAfter(err); wait > 0 && try < Tries {
			c.Sleep(min(wait, 30*time.Second))
			continue
		}
		if permanent(err) {
			c.fail(m.Handle, err, true)
			return nil
		}
		if try >= Tries {
			c.fail(m.Handle, err, false)
			return nil
		}
		c.Sleep(time.Duration(try*try) * time.Second)
	}
}

func (c *Channel) newMessage(handle string, chat int64, words string, buttons [][2]string, silent bool, id *int64) error {
	got, err := c.Bot.Send(chat, words, buttons, silent)
	if err != nil {
		return err
	}
	*id = got
	c.keepShown(chat, got, words)
	if err := c.Store.Place(handle, Placed{Chat: chat, ID: got}); err != nil {
		c.Logf("[telegram] a message was sent and where it is could not be kept: an edit of it will be a new message: %v", err)
	}
	return nil
}

// fail says to the router that a message did not arrive, and why.
func (c *Channel) fail(handle string, err error, forGood bool) {
	c.mu.Lock()
	c.lastErr = err.Error()
	c.failed++
	c.mu.Unlock()
	c.Logf("[telegram] a message was not delivered: %v", err)
	f := asks.Failed{Handle: handle, Why: err.Error(), Permanent: forGood, At: c.Now()}
	if perr := c.Publish(asks.IntakeSubject(asks.FailedWhat, Kind), f, "failed."+handle+"."+strconv.FormatInt(c.Now().UnixNano(), 10)); perr != nil {
		c.Logf("[telegram] and the router could not be told: %v", perr)
	}
}

// private says whether an arrival is from an account in its own private chat with the bot: anything in a
// group, or from another bot, is not the operator's own, whoever the router would say they are.
func private(from *User, chat Chat) bool {
	return from != nil && !from.IsBot && chat.Type == "private" && chat.ID == from.ID
}

func senderOf(u User) asks.Sender {
	return asks.Sender{Identity: strconv.FormatInt(u.ID, 10), Display: u.Display(), Authenticated: true}
}

// Handle takes one update.
func (c *Channel) Handle(u Update) {
	c.init()
	now := c.Now()
	switch {
	case u.Callback != nil:
		cb := u.Callback
		if cb.Message == nil || !private(&cb.From, cb.Message.Chat) {
			c.answerTap(cb.ID, "")
			c.Logf("[telegram] a tap outside a private chat was dropped")
			return
		}
		// Only on a message showing exactly what this channel put there: one whose words another holder of
		// the token changed, or one it never showed, answers nothing (the review of 2026-10-09).
		if want, err := c.Store.Shown(cb.Message.Chat.ID, cb.Message.MessageID); err != nil || want == "" ||
			want != wordsDigest(cb.Message.Text) {
			c.answerTap(cb.ID, "That answer was not taken: this message does not show what the mesh put there. Nothing was done.")
			c.Logf("[telegram] a tap on a message not showing what this channel put there was refused")
			return
		}
		// At once, whatever comes of it: the phone shows a spinner until then.
		c.answerTap(cb.ID, "")
		id := "tg-tap-" + cb.ID
		c.remember(id, cb.Message.Chat.ID)
		ch := asks.Choice{ID: id, Ticket: cb.Data, Sender: senderOf(cb.From), At: now}
		if err := c.Publish(asks.IntakeSubject(asks.Chosen, Kind), ch, id); err != nil {
			c.Logf("[telegram] a tap could not be said to the router: %v", err)
			_ = c.Bot.AnswerTap(cb.ID, "The mesh could not take this answer now. Try again in a minute.")
		}
	case u.Message != nil:
		m := u.Message
		if !private(m.From, m.Chat) {
			return // a group, a channel post, another bot: nothing here is the operator's own
		}
		id := fmt.Sprintf("tg-msg-%d", u.UpdateID)
		c.remember(id, m.Chat.ID)
		switch t := strings.TrimSpace(m.Text); {
		case t == "/start" || strings.HasPrefix(t, "/start "):
			l := asks.LinkAsked{ID: id, Sender: senderOf(*m.From), At: now}
			if err := c.Publish(asks.IntakeSubject(asks.Link, Kind), l, id); err != nil {
				c.reply(m.Chat.ID, "The mesh could not take this now. Try again in a minute.")
			}
		case codeShape.MatchString(t):
			code := codeShape.ReplaceAllString(t, "$1$2")
			// Out of the chat as soon as it is read (ADR 0234: deleted where the service allows).
			if err := c.Bot.Delete(m.Chat.ID, m.MessageID); err != nil {
				c.Logf("[telegram] a code could not be deleted from the chat: %v", err)
			}
			raw, err := c.Prove(asks.ProofSubject(asks.CodeProof, Kind), asks.Code{Sender: senderOf(*m.From), Code: code, Purpose: "link"})
			if err != nil {
				c.reply(m.Chat.ID, "The mesh could not check the code now. Try again in a minute.")
				return
			}
			var a asks.ProofAnswer
			if json.Unmarshal(raw, &a) != nil || a.Words == "" {
				a.Words = "The mesh's answer could not be read."
			}
			c.reply(m.Chat.ID, a.Words)
		default:
			c.reply(m.Chat.ID, FixedLine)
		}
	}
}

func (c *Channel) answerTap(id, words string) {
	if err := c.Bot.AnswerTap(id, words); err != nil {
		c.Logf("[telegram] a tap could not be answered: %v", err)
	}
}

func (c *Channel) remember(id string, chat int64) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if len(c.replyTo) > 500 {
		c.replyTo = map[string]int64{}
	}
	c.replyTo[id] = chat
}

func (c *Channel) reply(chat int64, words string) {
	if _, err := c.Bot.Send(chat, words, nil, true); err != nil {
		c.Logf("[telegram] could not answer in the chat: %v", err)
	}
}

// Poll asks for what happened once, and takes it. A conflict is another program taking this bot's
// updates: a webhook set, or the token in other hands.
func (c *Channel) Poll(wait time.Duration) error {
	c.init()
	offset, err := c.Store.Offset()
	if err != nil {
		return err
	}
	updates, err := c.Bot.Updates(offset, wait)
	c.mu.Lock()
	if err != nil {
		var a *apiError
		if errors.As(err, &a) && a.Code == 409 {
			err = errors.New("another program takes this bot's updates (a webhook, or the token in other hands): " + a.Words)
		}
		c.pollErr = err.Error()
	} else {
		c.pollErr, c.lastPoll = "", c.Now()
	}
	c.mu.Unlock()
	if err != nil {
		return err
	}
	for _, u := range updates {
		c.Handle(u)
		if u.UpdateID >= offset {
			offset = u.UpdateID + 1
		}
		// Kept after each update, not after the batch: a restart in the middle takes again only the one it
		// was handling (the review of 2026-10-08), and the router acts on an answer once by its id anyway.
		if err := c.Store.SetOffset(offset); err != nil {
			c.Logf("[telegram] the update offset could not be kept; a restart may take some updates again: %v", err)
		}
	}
	return nil
}

// Standing is whether the channel can send now: the token works and the updates arrive.
func (c *Channel) Standing() asks.Standing {
	c.init()
	now := c.Now()
	s := asks.Standing{EditsSilently: true, At: now}
	if _, err := c.Bot.token(); err != nil {
		s.Why = err.Error()
		return s
	}
	c.mu.Lock()
	stale := now.Sub(c.meAt) >= StandingEvery
	c.mu.Unlock()
	if stale {
		me, err := c.Bot.Me()
		c.mu.Lock()
		c.meAt = now
		c.me, c.meErr = me.Username, ""
		if err != nil {
			c.meErr = err.Error()
		}
		c.mu.Unlock()
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	switch {
	case c.meErr != "":
		s.Why = "the bot token does not work: " + c.meErr
	case c.pollErr != "":
		s.Why = "the updates do not arrive: " + c.pollErr
	default:
		s.Ready = true
	}
	return s
}

// SayStanding says the standing when it changed, or every StandingEvery.
func (c *Channel) SayStanding() {
	s := c.Standing()
	c.mu.Lock()
	same := s.Ready == c.said.Ready && s.Why == c.said.Why && s.At.Sub(c.saidAt) < StandingEvery
	c.mu.Unlock()
	if same {
		return
	}
	if err := c.Publish(asks.IntakeSubject(asks.StandingWhat, Kind), s, "standing."+strconv.FormatInt(s.At.UnixNano(), 10)); err != nil {
		c.Logf("[telegram] could not say whether it can send: %v", err)
		return
	}
	c.mu.Lock()
	c.said, c.saidAt = s, s.At
	c.mu.Unlock()
}

// Status is the channel's account of itself, for the tool.
func (c *Channel) Status() map[string]any {
	s := c.Standing()
	c.mu.Lock()
	defer c.mu.Unlock()
	out := map[string]any{"kind": Kind, "can_send": s.Ready, "delivered_since_start": c.delivered,
		"failed_since_start": c.failed, "edits_notify_nobody": true}
	if s.Why != "" {
		out["why_not"] = s.Why
	}
	if c.me != "" {
		out["bot"] = "@" + c.me
	}
	if !c.lastPoll.IsZero() {
		out["updates_last_arrived"] = c.lastPoll.UTC().Format(time.RFC3339)
	}
	if !c.lastSent.IsZero() {
		out["last_delivered"] = c.lastSent.UTC().Format(time.RFC3339)
	}
	if c.lastErr != "" {
		out["last_error"] = c.lastErr
	}
	out["rules"] = []string{
		"it shows what the router sends and says what was chosen; whether the sender is the operator is the router's to judge, from the account linked there (novox/hq ADR 0259)",
		"a button carries only the router's one-time ticket",
		"only a private chat with the account itself counts; a tap is acknowledged at once",
		"/start asks to link this account; the code shown on the desk is checked by request and reply, never in an event, and deleted from the chat",
		"updates are long-polled: outbound only, no route into the mesh",
		"up to 4096 characters arrive whole; longer is cut and the cut said",
		"warnings and clearings arrive without sound; urgent messages and questions ring",
		"a rate limit is waited out; a refusal is said to the router as permanent; an edit to the same text is a success",
		"it runs as its own account, in a runtime of its own on a bus account of its own, never the machine's runtime; that account alone reads the bot token",
	}
	return out
}