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

// telegram: a channel of the operator's conversation, kind `telegram` of the `channel` and `intake` benches
// (novox/hq ADR 0234, ADR 0259). A Go bundle the node's runtime launches as this module's own account, never
// root and never the operator's: the bot token and this channel's state are that account's alone. stdout is
// the MCP channel; what this module says, it says on stderr.
package main

import (
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"time"

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

func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }

// Worker is the work queue of this kind, as the controller names it.
const Worker = "SEAT_CHANNEL_TELEGRAM_worker"

// busStore keeps the update offset and where each message is in this module's state on the bus.
type busStore struct{}

func (busStore) Offset() (int64, error) {
	e, err := stdio.State("offset").Get("updates")
	if err != nil || e == nil {
		return 0, err
	}
	var o struct {
		Next int64 `json:"next"`
	}
	return o.Next, json.Unmarshal(e.Value, &o)
}
func (busStore) SetOffset(next int64) error {
	_, err := stdio.State("offset").Put("updates", map[string]int64{"next": next})
	return err
}
func (busStore) Placed(handle string) (*Placed, error) {
	e, err := stdio.State("messages").Get(handle)
	if err != nil || e == nil {
		return nil, err
	}
	var p Placed
	return &p, json.Unmarshal(e.Value, &p)
}
func (busStore) Place(handle string, p Placed) error {
	_, err := stdio.State("messages").Put(handle, p)
	return err
}

// shownKey is where what a message shows is kept, beside the router's handles in the same state.
func shownKey(chat, id int64) string { return fmt.Sprintf("shown-words.%d.%d", chat, id) }

func (busStore) Shown(chat, id int64) (string, error) {
	e, err := stdio.State("messages").Get(shownKey(chat, id))
	if err != nil || e == nil {
		return "", err
	}
	var w struct {
		Digest string `json:"digest"`
	}
	return w.Digest, json.Unmarshal(e.Value, &w)
}
func (busStore) Show(chat, id int64, digest string) error {
	_, err := stdio.State("messages").Put(shownKey(chat, id), map[string]string{"digest": digest})
	return err
}

func main() {
	c := &Channel{
		Bot:     &Bot{TokenFile: os.Getenv("MESH_TELEGRAM_TOKEN_FILE"), Client: &http.Client{Timeout: PollWait + 15*time.Second}},
		Store:   busStore{},
		Publish: func(subject string, body any, id string) error { return stdio.SeatPublish(subject, body, id) },
		Prove:   stdio.SeatProve,
		Logf:    logf,
	}
	c.init()
	go run(c)
	tools := []stdio.Tool{{Name: "telegram_status",
		Description: "Whether this channel can reach the operator on Telegram, and why not: whether the bot token works " +
			"(never the token), the bot's name, when updates last arrived, the last delivery and the last error, how " +
			"many it delivered and failed since it started, and the rules it keeps. Linking an account as the " +
			"operator is the router's (messenger_identities).",
		Run: func(map[string]any) (any, error) { return c.Status(), nil }}}
	if err := stdio.Serve("", tools); err != nil {
		logf("%v", err)
		os.Exit(1)
	}
}

// run takes this kind's work, long-polls the updates and says its standing — each retried, each failure said.
func run(c *Channel) {
	time.Sleep(500 * time.Millisecond) // Serve first: the bus is reached through it
	go func() {
		for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
			err := stdio.SeatTake(Worker, func(w stdio.Work) error { return c.Take(w.Subject, w.Body) })
			if err == nil {
				logf("[telegram] taking the router's messages for this kind")
				return
			}
			logf("[telegram] not taking messages yet (%v); asking again in %s", err, wait)
			time.Sleep(wait)
		}
	}()
	go func() {
		for {
			c.SayStanding()
			time.Sleep(time.Minute)
		}
	}()
	said := ""
	for {
		if _, err := c.Bot.token(); err != nil {
			if err.Error() != said {
				logf("[telegram] %v", err)
				said = err.Error()
			}
			time.Sleep(time.Minute)
			continue
		}
		if err := c.Poll(PollWait); err != nil {
			if err.Error() != said {
				logf("[telegram] the updates do not arrive: %v", err)
				said = err.Error()
			}
			c.SayStanding()
			time.Sleep(10 * time.Second)
			continue
		}
		said = ""
	}
}