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. A
//     `send` or `show` under a handle already shown replaces that message: the new one sent, then the earlier
//     one deleted.
//   - **What arrives** is long-polled: outbound only. A tapped button is said as a `choice` with the sender
//     the service authenticated, and acknowledged on the phone with the router's word on it — what was
//     recorded, or why it was not — which comes as a reply to the tap; without one in TapWait, with words that
//     claim nothing, and the word when it comes is said in the chat (issue 372); 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
	// TapWait is how long a tap's acknowledgement waits for the router's word on it (issue 372): the phone
	// shows its spinner meanwhile, and Telegram takes an acknowledgement only for a short while.
	// It is longer than the router may wait for the controller's word on root before it judges an answer
	// (messenger's RootFreeWithin, 10 seconds), so a slow controller still leaves the word to the toast.
	TapWait = 13 * time.Second
	// TapPending is a tap's acknowledgement when the router said nothing within TapWait: it claims nothing.
	TapPending = "Passed to the mesh; nothing is confirmed yet."
	// ToastLength is what a tap's acknowledgement shows (Telegram's own bound).
	ToastLength = 200
	// 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)
	// After calls f once d has passed (time.AfterFunc); a test calls it when it says.
	After func(d time.Duration, f func())

	mu        sync.Mutex
	replyTo   map[string]int64  // an arrival's id to the chat it came from, for the router's reply
	taps      map[string]string // a tap's arrival id to its callback, until it is acknowledged
	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
	}
	if c.taps == nil {
		c.taps = map[string]string{}
	}
	if c.After == nil {
		c.After = func(d time.Duration, f func()) { time.AfterFunc(d, f) }
	}
}

// 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
	}
	// The router's word on a tap, said where the tap was made: the tap's own acknowledgement (issue 372). A
	// word of one line is that alone; one with more to it is also said in the chat, where it stays.
	if verb == asks.Send && m.Reply != "" && !m.Secret && c.settleTap(m.Reply, toast(m)) && strings.TrimSpace(m.Body) == "" {
		c.mu.Lock()
		c.lastSent, c.lastErr = c.Now(), ""
		c.delivered++
		c.mu.Unlock()
		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)
	// **A message said again under a handle already shown replaces it** (issue 373): a reminder or an
	// escalation is one message per problem, not one more beside it. An edit here notifies nobody, so this one
	// is sent, ringing as its kind rings, and only then is the earlier one deleted — one the service will not
	// delete loses its buttons and says a newer message follows. A send that fails leaves the earlier one.
	var place, earlier *Placed
	switch {
	case verb == asks.Edit:
		if place, err = c.Store.Placed(m.Handle); err != nil {
			return err // the state is away: offered again
		}
	case !m.Secret:
		if earlier, err = c.Store.Placed(m.Handle); err != nil {
			c.Logf("[telegram] where an earlier message under this handle is could not be read (%v): sent as a new one", err)
			earlier = nil
		}
	}
	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 {
			if earlier != nil {
				c.replaced(*earlier)
			}
			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)
	}
}

// replaced takes an earlier message out of the chat: deleted, or where the service refuses that (a message
// older than two days), its buttons taken away and the newer message named.
func (c *Channel) replaced(p Placed) {
	err := c.Bot.Delete(p.Chat, p.ID)
	if err == nil {
		return
	}
	c.Logf("[telegram] an earlier message could not be deleted (%v); its buttons are taken away instead", err)
	if err := c.Bot.Edit(p.Chat, p.ID, "Replaced by a newer message below.", nil); err != nil && !gone(err) {
		c.Logf("[telegram] an earlier message could not be marked replaced either: %v", err)
	}
}

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) {
			words := "That answer was not taken: this message does not show what the mesh put there. Nothing was done."
			c.answerTap(cb.ID, words)
			// Said on the phone alone, never in the chat: the tapper may be somebody the buttons reached, who is told
			// nothing more (the lab's proof).
			c.Logf("[telegram] a tap on a message not showing what this channel put there was refused")
			return
		}
		// Acknowledged with the router's word on it, which comes as a reply to this tap (Take): what was recorded,
		// or why it was refused. Never before: an acknowledgement claims nothing the router has not said. If the
		// router says nothing within TapWait, the tap is acknowledged with words that claim nothing, and its word
		// when it comes is said in the chat (issue 372).
		id := "tg-tap-" + cb.ID
		c.remember(id, cb.Message.Chat.ID)
		c.mu.Lock()
		c.taps[id] = cb.ID
		c.mu.Unlock()
		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)
			words := "The mesh could not take this answer now. Nothing was done. Try again in a minute."
			c.settleTap(id, words)
			c.reply(cb.Message.Chat.ID, words)
			return
		}
		c.After(TapWait, func() { c.settleTap(id, TapPending) })
	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)
		}
	}
}

// settleTap acknowledges a tap still waiting, once, with words: true when it was waiting and Telegram took
// the acknowledgement. A tap already acknowledged, or one Telegram no longer takes, is false: the words are
// then for the chat.
func (c *Channel) settleTap(arrival, words string) bool {
	c.mu.Lock()
	cb, waiting := c.taps[arrival]
	delete(c.taps, arrival)
	c.mu.Unlock()
	if !waiting {
		return false
	}
	if err := c.Bot.AnswerTap(cb, words); err != nil {
		c.Logf("[telegram] a tap could not be answered: %v", err)
		return false
	}
	return true
}

// toast is a message as a tap's acknowledgement shows it: its title and the first line of its body, cut at
// ToastLength.
func toast(m asks.Message) string {
	t := strings.TrimSpace(m.Title)
	if first, _, _ := strings.Cut(strings.TrimSpace(m.Body), "\n"); first != "" {
		t += ": " + first
	}
	if utf8.RuneCountInString(t) > ToastLength {
		t = string([]rune(t)[:ToastLength-1]) + "…"
	}
	return t
}

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 with the router's word on it — what was recorded, or why not — or, without one in 13 seconds, with words that claim nothing",
		"/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
}