From a4123897258dfbbb4b1986f44cca9b1e6ff7e78f Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 8 Oct 2026 16:57:05 +0200 Subject: [PATCH] Add telegram and the desk as channels of their own, each judging nothing and holding only its own secret (hq ADR 0259) --- modules/desk-channel/README.md | 30 ++ modules/desk-channel/cmd/desk-channel/desk.go | 367 ++++++++++++++ .../cmd/desk-channel/desk_test.go | 161 +++++++ modules/desk-channel/cmd/desk-channel/main.go | 136 ++++++ modules/desk-channel/go.mod | 5 + modules/desk-channel/go.sum | 2 + modules/desk-channel/module.json | 76 +++ modules/telegram/README.md | 69 +++ modules/telegram/cmd/telegram/bot.go | 290 +++++++++++ modules/telegram/cmd/telegram/channel.go | 453 ++++++++++++++++++ modules/telegram/cmd/telegram/channel_test.go | 365 ++++++++++++++ modules/telegram/cmd/telegram/main.go | 115 +++++ .../telegram/cmd/telegram/manifest_test.go | 67 +++ modules/telegram/go.mod | 5 + modules/telegram/go.sum | 2 + modules/telegram/module.json | 77 +++ 16 files changed, 2220 insertions(+) create mode 100644 modules/desk-channel/README.md create mode 100644 modules/desk-channel/cmd/desk-channel/desk.go create mode 100644 modules/desk-channel/cmd/desk-channel/desk_test.go create mode 100644 modules/desk-channel/cmd/desk-channel/main.go create mode 100644 modules/desk-channel/go.mod create mode 100644 modules/desk-channel/go.sum create mode 100644 modules/desk-channel/module.json create mode 100644 modules/telegram/README.md create mode 100644 modules/telegram/cmd/telegram/bot.go create mode 100644 modules/telegram/cmd/telegram/channel.go create mode 100644 modules/telegram/cmd/telegram/channel_test.go create mode 100644 modules/telegram/cmd/telegram/main.go create mode 100644 modules/telegram/cmd/telegram/manifest_test.go create mode 100644 modules/telegram/go.mod create mode 100644 modules/telegram/go.sum create mode 100644 modules/telegram/module.json diff --git a/modules/desk-channel/README.md b/modules/desk-channel/README.md new file mode 100644 index 00000000..43d86698 --- /dev/null +++ b/modules/desk-channel/README.md @@ -0,0 +1,30 @@ +# desk-channel + +The desk as a channel of the operator's conversation: kind `desktop` of the `channel` and `intake` benches the +router (`messenger`) declares (novox/hq ADR 0258, ADR 0259). It shows the router's messages on the notifier of +each machine the operator sits at (the `node-notifier` seat), and says which button was chosen there. + +- **It proves nothing about who clicked.** Any process of the operator's account can choose a notification's + action (`dunstctl action`), so its claim declares no `verified-sender`: the router offers here only + acknowledgements (Silence for a week, Details) and says in the words where an approval can be given. +- **An answer is taken only on a machine its message was shown on**, and once; a refused click is said where it + was made. A message with buttons stays until it is dismissed. +- **A notifier that started again** (`dunst.started`) is shown again every open message with buttons it showed. +- A message no machine took is said to the router as failed; the router tries it again. + +## What the operator gives + +The machines to show on, as a setting: `settings` for `desk-channel` with +`{"desktop-machines": ["", …]}`. Until then it says it cannot send, and the +router sends warnings to the channel that reaches the operator away. + +## Where it runs + +Held once for the mesh, on the control node beside the router. It asks each machine's notifier through the +mesh: nothing new runs on the machines. + +## Tools + +| tool | does | +|---|---| +| `desk_channel_status` | whether it can show messages, the machines and each one's last answer, and its rules | diff --git a/modules/desk-channel/cmd/desk-channel/desk.go b/modules/desk-channel/cmd/desk-channel/desk.go new file mode 100644 index 00000000..dcbedb60 --- /dev/null +++ b/modules/desk-channel/cmd/desk-channel/desk.go @@ -0,0 +1,367 @@ +package main + +// The desk: kind `desktop` of the `channel` and `intake` benches (novox/hq ADR 0258, ADR 0259). It shows the +// router's messages on the notifier of each machine the operator sits at (the `node-notifier` seat, dunst) +// and says which button was chosen there. It is a channel like any other: the router judges every answer. +// +// - **It proves nothing about who clicked.** Any process of the operator's account can choose a +// notification's action, so its claim declares no `verified-sender`: the router offers here only +// acknowledgements, and says where an approval can be given. +// - **An answer is bound to the machines it was shown on**: a click on another machine is refused here, and +// said there. A message with buttons stays until it is dismissed. +// - **A notifier that started again lost the answers it waited for**: every open message with buttons shown +// on that machine is shown there again. +// - **Standing**: it can send when machines are named in its settings; which of them answered last is in its +// status. A message no machine took is said to the router as failed, and tried again by it. + +import ( + "encoding/json" + "errors" + "fmt" + "html" + "sort" + "strconv" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" +) + +// Kind is this channel's kind on both benches. +const Kind = "desktop" + +// StandingEvery is how often the channel says whether it can send, at least. +const StandingEvery = 5 * time.Minute + +// Shown is one of the router's messages as the desk shows it: the message, and its notification on each +// machine. +type Shown struct { + Message asks.Message `json:"message"` + On map[string]int `json:"on"` + Sent time.Time `json:"sent"` +} + +// Store keeps what is shown, by the router's handle. +type Store interface { + Get(handle string) (*Shown, error) + Put(handle string, s Shown) error + Handles() ([]string, error) +} + +// Desk is the channel's whole working. +type Desk struct { + Machines func() []string + // Notify calls one machine's notifier: `seat:node-notifier.send@`. + Notify func(machine string, args map[string]any) (json.RawMessage, error) + Store Store + Publish func(subject string, body any, id string) error + Now func() time.Time + Logf func(string, ...any) + + mu sync.Mutex + last map[string]string // machine -> its last error, "" after a success + clickedOn map[string]string // a choice's id -> the machine it was made on, for the router's reply + said asks.Standing + saidAt time.Time + acted map[string]bool +} + +func (d *Desk) init() { + d.mu.Lock() + defer d.mu.Unlock() + if d.last == nil { + d.last, d.clickedOn, d.acted = map[string]string{}, map[string]string{}, map[string]bool{} + } + if d.Now == nil { + d.Now = time.Now + } + if d.Logf == nil { + d.Logf = func(string, ...any) {} + } +} + +func (d *Desk) note(machine string, err error) { + d.mu.Lock() + defer d.mu.Unlock() + if err != nil { + d.last[machine] = err.Error() + } else { + d.last[machine] = "" + } +} + +// Take does one piece of the desk's work. It returns an error only to have the work offered again. +func (d *Desk) Take(subject string, body json.RawMessage) error { + d.init() + parts := strings.Split(subject, ".") + if len(parts) != 6 || parts[5] != Kind { + d.Logf("[desk-channel] work on %s is not this kind's; taken and dropped", subject) + return nil + } + var m asks.Message + if err := json.Unmarshal(body, &m); err != nil || m.Handle == "" { + d.Logf("[desk-channel] work on %s could not be read; taken and dropped", subject) + return nil + } + machines := d.Machines() + if m.Reply != "" { + // An answer to a click is shown where the click was made. + d.mu.Lock() + on, ok := d.clickedOn[m.Reply] + d.mu.Unlock() + if ok { + machines = []string{on} + } + } + var before *Shown + if parts[4] == asks.Edit { + var err error + if before, err = d.Store.Get(m.Handle); err != nil { + return err + } + } + s := Shown{Message: m, On: map[string]int{}, Sent: d.Now()} + var failed []string + for _, machine := range machines { + replace := 0 + if before != nil { + replace = before.On[machine] + if replace == 0 { + continue // an edit goes where the message was shown, and nowhere new + } + } + id, err := d.show(machine, m, replace) + if err != nil { + failed = append(failed, machine+": "+err.Error()) + continue + } + s.On[machine] = id + } + if len(s.On) == 0 { + why := "no machine showed it" + if len(failed) > 0 { + why += " — " + strings.Join(failed, "; ") + } + if len(machines) == 0 { + why = "no machine to show it on: the setting desktop-machines names none" + } + f := asks.Failed{Handle: m.Handle, Why: why, At: d.Now()} + if err := d.Publish(asks.IntakeSubject(asks.FailedWhat, Kind), f, "failed."+m.Handle+"."+strconv.FormatInt(d.Now().UnixNano(), 10)); err != nil { + return err + } + return nil + } + if err := d.Store.Put(m.Handle, s); err != nil { + d.Logf("[desk-channel] shown, and what was shown could not be kept: an edit of it will not reach it: %v", err) + } + return nil +} + +// show puts one message on one machine's notifier, in place of `replace` when it names one. +func (d *Desk) show(machine string, m asks.Message, replace int) (int, error) { + urgency := "normal" + switch { + case m.Urgent: + urgency = "critical" + case m.Silent: + urgency = "low" + } + args := map[string]any{"summary": m.Title, "body": html.EscapeString(m.Body), "urgency": urgency, "app_name": "mesh"} + if replace != 0 { + args["replace_id"] = replace + } + if len(m.Buttons) > 0 { + var actions []map[string]string + for _, b := range m.Buttons { + actions = append(actions, map[string]string{"token": b.Ticket, "label": b.Label}) + } + // Answers stay until the operator dismisses them: a warning's answers must not die in ten seconds. + args["expire_ms"] = 0 + args["actions"] = actions + } + raw, err := d.Notify(machine, args) + if err == nil { + var id int + if id, err = notificationID(raw); err == nil { + d.note(machine, nil) + return id, nil + } + } + d.note(machine, err) + return 0, err +} + +// Chosen takes a button chosen on a machine's notifier (dunst's `action-chosen`): said to the router as this +// kind's choice when the message it belongs to was shown on that machine; refused there otherwise. The +// machine is the event's own node, said by the runtime; the same event is acted on once. +func (d *Desk) Chosen(event, machine, ticket string) { + d.init() + d.mu.Lock() + done := d.acted[event] + if event != "" { + d.acted[event] = true + } + d.mu.Unlock() + if done { + return + } + handle, shown := d.find(ticket) + if shown == nil { + _, _ = d.show(machine, asks.Message{Title: "That answer was not taken", + Body: "It belongs to no message shown here any more. Nothing was done.", Silent: true}, 0) + return + } + if _, here := shown.On[machine]; !here { + _, _ = d.show(machine, asks.Message{Title: "That answer was not taken", + Body: "It was not shown on " + machine + ". Nothing was done.", Silent: true}, 0) + return + } + id := "desk-" + event + if event == "" { + id = "desk-" + ticket + "-" + strconv.FormatInt(d.Now().UnixNano(), 10) + } + d.mu.Lock() + if len(d.clickedOn) > 500 { + d.clickedOn = map[string]string{} + } + d.clickedOn[id] = machine + d.mu.Unlock() + c := asks.Choice{ID: id, Ticket: ticket, Handle: handle, At: d.Now(), + Sender: asks.Sender{Identity: machine, Display: machine, Authenticated: false}} + if err := d.Publish(asks.IntakeSubject(asks.Chosen, Kind), c, id); err != nil { + d.Logf("[desk-channel] a choice could not be said to the router: %v", err) + _, _ = d.show(machine, asks.Message{Title: "That answer could not be taken now", + Body: "The mesh did not take it. Try again in a minute.", Silent: true}, 0) + } +} + +func (d *Desk) find(ticket string) (string, *Shown) { + handles, err := d.Store.Handles() + if err != nil { + return "", nil + } + for _, h := range handles { + s, err := d.Store.Get(h) + if err != nil || s == nil { + continue + } + for _, b := range s.Message.Buttons { + if b.Ticket == ticket { + return h, s + } + } + } + return "", nil +} + +// Started shows again, on a machine whose notifier started again, every message with buttons shown there. +func (d *Desk) Started(machine string) int { + d.init() + handles, err := d.Store.Handles() + if err != nil { + return 0 + } + sort.Strings(handles) + n := 0 + for _, h := range handles { + s, err := d.Store.Get(h) + if err != nil || s == nil || len(s.Message.Buttons) == 0 { + continue + } + if _, here := s.On[machine]; !here { + continue + } + id, err := d.show(machine, s.Message, 0) + if err != nil { + continue + } + s.On[machine] = id + _ = d.Store.Put(h, *s) + n++ + } + if n > 0 { + d.Logf("[desk-channel] the notifier on %s started again: %d message(s) with answers shown there again", machine, n) + } + return n +} + +// Standing is whether the desk can send: machines are named to show on. +func (d *Desk) Standing() asks.Standing { + d.init() + s := asks.Standing{At: d.Now()} + if len(d.Machines()) == 0 { + s.Why = "no machine to show it on: the setting desktop-machines names none" + return s + } + s.Ready = true + return s +} + +// SayStanding says the standing when it changed, or every StandingEvery. +func (d *Desk) SayStanding() { + s := d.Standing() + d.mu.Lock() + same := s.Ready == d.said.Ready && s.Why == d.said.Why && s.At.Sub(d.saidAt) < StandingEvery + d.mu.Unlock() + if same { + return + } + if err := d.Publish(asks.IntakeSubject(asks.StandingWhat, Kind), s, "standing."+strconv.FormatInt(s.At.UnixNano(), 10)); err != nil { + d.Logf("[desk-channel] could not say whether it can send: %v", err) + return + } + d.mu.Lock() + d.said, d.saidAt = s, s.At + d.mu.Unlock() +} + +// Status is the desk's account of itself. +func (d *Desk) Status() map[string]any { + s := d.Standing() + d.mu.Lock() + defer d.mu.Unlock() + last := map[string]string{} + for k, v := range d.last { + if v == "" { + v = "took the last message" + } + last[k] = v + } + out := map[string]any{"kind": Kind, "can_send": s.Ready, "machines": d.Machines(), "machines_last": last, + "rules": []string{ + "it shows the router's messages on the notifier of each machine named in desktop-machines, and says which button was chosen there; the router judges every answer (novox/hq ADR 0259)", + "it proves nothing about who clicked: any process of the operator's account can choose an action, so it answers acknowledgements only", + "an answer is taken only on a machine its message was shown on; a message with buttons stays until dismissed", + "a notifier that started again is shown again every open message with buttons it showed", + }} + if s.Why != "" { + out["why_not"] = s.Why + } + return out +} + +// notificationID reads the notifier's answer, {"id": N}, bare or in a tool reply's text. +func notificationID(raw json.RawMessage) (int, error) { + var direct struct { + ID *int `json:"id"` + } + if json.Unmarshal(raw, &direct) == nil && direct.ID != nil { + return *direct.ID, nil + } + var wrapped struct { + Content []struct { + Text string `json:"text"` + } `json:"content"` + IsError bool `json:"isError"` + } + if json.Unmarshal(raw, &wrapped) == nil && len(wrapped.Content) > 0 { + if wrapped.IsError { + return 0, errors.New(wrapped.Content[0].Text) + } + if json.Unmarshal([]byte(wrapped.Content[0].Text), &direct) == nil && direct.ID != nil { + return *direct.ID, nil + } + } + return 0, fmt.Errorf("the notifier answered no id: %.80s", string(raw)) +} diff --git a/modules/desk-channel/cmd/desk-channel/desk_test.go b/modules/desk-channel/cmd/desk-channel/desk_test.go new file mode 100644 index 00000000..7c767b70 --- /dev/null +++ b/modules/desk-channel/cmd/desk-channel/desk_test.go @@ -0,0 +1,161 @@ +package main + +import ( + "encoding/json" + "errors" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" +) + +type memShown map[string]Shown + +func (m memShown) Get(h string) (*Shown, error) { + s, ok := m[h] + if !ok { + return nil, nil + } + return &s, nil +} +func (m memShown) Put(h string, s Shown) error { m[h] = s; return nil } +func (m memShown) Handles() ([]string, error) { + var out []string + for h := range m { + out = append(out, h) + } + return out, nil +} + +type notified struct { + machine string + args map[string]any +} + +func newDesk(t *testing.T, machines ...string) (*Desk, *[]notified, map[string][]json.RawMessage, memShown) { + t.Helper() + var shown []notified + said := map[string][]json.RawMessage{} + store := memShown{} + n := 0 + d := &Desk{Machines: func() []string { return machines }, Store: store, Logf: t.Logf, + Now: func() time.Time { return time.Date(2026, 10, 8, 14, 0, 0, 0, time.UTC) }, + Notify: func(machine string, args map[string]any) (json.RawMessage, error) { + if machine == "away-machine" { + return nil, errors.New("no session answers") + } + n++ + shown = append(shown, notified{machine, args}) + raw, _ := json.Marshal(map[string]int{"id": n}) + return raw, nil + }, + Publish: func(subject string, body any, id string) error { + raw, _ := json.Marshal(body) + said[subject] = append(said[subject], raw) + return nil + }} + d.init() + return d, &shown, said, store +} + +func take(t *testing.T, d *Desk, verb string, m asks.Message) { + t.Helper() + raw, _ := json.Marshal(m) + if err := d.Take(asks.ChannelSubject(verb, Kind), raw); err != nil { + t.Fatal(err) + } +} + +// novox/hq ADR 0259: the desk shows the router's message with its tickets as the notifier's actions, and +// says a click as a choice the router judges — only on a machine the message was shown on. +func TestAClickOnAMachineTheMessageWasShownOnIsSaidToTheRouter(t *testing.T) { + d, shown, said, _ := newDesk(t, "laptop", "away-machine") + take(t, d, asks.Show, asks.Message{Handle: "m1", Title: "3 failed services", Body: "Needs you: them.", + Buttons: []asks.Button{{Label: "Silence for a week", Ticket: "t1"}}}) + if len(*shown) != 1 || (*shown)[0].machine != "laptop" || (*shown)[0].args["expire_ms"] != 0 || + (*shown)[0].args["body"] != "Needs you: <mend> them." { + t.Fatalf("shown %+v", *shown) + } + d.Chosen("e1", "laptop", "t1") + d.Chosen("e1", "laptop", "t1") // a redelivery + got := said[asks.IntakeSubject(asks.Chosen, Kind)] + var c asks.Choice + if len(got) != 1 || json.Unmarshal(got[0], &c) != nil || c.Ticket != "t1" || c.Handle != "m1" || + c.Sender.Identity != "laptop" || c.Sender.Authenticated { + t.Fatalf("said %s", got) + } + // Elsewhere it is refused, there, and nothing is said to the router. + d.Chosen("e2", "workstation", "t1") + d.Chosen("e3", "laptop", "forged") + if len(said[asks.IntakeSubject(asks.Chosen, Kind)]) != 1 { + t.Error("a click not shown there was said") + } + if last := (*shown)[len(*shown)-1]; last.machine != "laptop" || last.args["summary"] != "That answer was not taken" { + t.Errorf("the refusal was not shown where it was chosen: %+v", last) + } + // The router's reply to the click is shown on the machine it was made on. + take(t, d, asks.Send, asks.Message{Handle: "r1", Title: "Details: 3 failed services", Reply: c.ID}) + if last := (*shown)[len(*shown)-1]; last.machine != "laptop" || last.args["summary"] != "Details: 3 failed services" { + t.Errorf("%+v", last) + } +} + +func TestAnEditReplacesTheNotificationAndANotifierThatStartedAgainIsShownItsQuestions(t *testing.T) { + d, shown, _, store := newDesk(t, "laptop") + take(t, d, asks.Show, asks.Message{Handle: "m1", Title: "q", Buttons: []asks.Button{{Label: "Silence for a week", Ticket: "t1"}}}) + take(t, d, asks.Edit, asks.Message{Handle: "m1", Title: "Silence for a week: chosen"}) + if last := (*shown)[len(*shown)-1]; last.args["replace_id"] != 1 || last.args["actions"] != nil { + t.Errorf("the edit: %+v", last.args) + } + take(t, d, asks.Show, asks.Message{Handle: "m2", Title: "q2", Buttons: []asks.Button{{Label: "Silence for a week", Ticket: "t2"}}}) + before := len(*shown) + if n := d.Started("laptop"); n != 1 || len(*shown) != before+1 { + t.Errorf("shown again: %d", n) + } + if store["m2"].On["laptop"] != len(*shown) { + t.Errorf("the new notification is not kept: %+v", store["m2"]) + } +} + +func TestNoMachineThatShowsItIsAFailureSaidToTheRouter(t *testing.T) { + d, _, said, _ := newDesk(t, "away-machine") + take(t, d, asks.Send, asks.Message{Handle: "m1", Title: "x"}) + if f := said[asks.IntakeSubject(asks.FailedWhat, Kind)]; len(f) != 1 || !strings.Contains(string(f[0]), "no session answers") || + strings.Contains(string(f[0]), `"permanent":true`) { + t.Errorf("%s", f) + } + none, _, _, _ := newDesk(t) + if s := none.Standing(); s.Ready || !strings.Contains(s.Why, "desktop-machines names none") { + t.Errorf("%+v", s) + } +} + +// It claims the desk's kind, and promises nothing that would let it approve. +func TestItPromisesNoVerifiedSender(t *testing.T) { + raw, err := os.ReadFile(filepath.Join("..", "..", "module.json")) + if err != nil { + t.Fatal(err) + } + var m struct { + Claims []struct { + Name, Kind string + Capabilities []string + } + Consumes []string + } + _ = json.Unmarshal(raw, &m) + if len(m.Claims) != 2 || m.Claims[0].Kind != Kind || m.Claims[1].Kind != Kind { + t.Fatalf("%+v", m.Claims) + } + for _, c := range m.Claims[0].Capabilities { + if c == "verified-sender" || c == "reaches-away" { + t.Errorf("the desk promises %s", c) + } + } + if strings.Join(m.Consumes, ",") != AnswerEvent+","+StartedEvent { + t.Errorf("consumes %v", m.Consumes) + } +} diff --git a/modules/desk-channel/cmd/desk-channel/main.go b/modules/desk-channel/cmd/desk-channel/main.go new file mode 100644 index 00000000..bba17dbf --- /dev/null +++ b/modules/desk-channel/cmd/desk-channel/main.go @@ -0,0 +1,136 @@ +// desk-channel: the desk as a channel of the operator's conversation, kind `desktop` of the `channel` and +// `intake` benches (novox/hq ADR 0258, ADR 0259). A Go bundle the node's runtime launches. stdout is the MCP +// channel; what this module says, it says on stderr. +package main + +import ( + "encoding/json" + "fmt" + "os" + "strings" + "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_DESKTOP_worker" + +// The notifier's events, which the dunst holder of node-notifier emits as its module's own. +const ( + AnswerEvent = "dunst.action-chosen" + StartedEvent = "dunst.started" +) + +type busStore struct{} + +func (busStore) Get(handle string) (*Shown, error) { + e, err := stdio.State("shown").Get(handle) + if err != nil || e == nil { + return nil, err + } + var s Shown + return &s, json.Unmarshal(e.Value, &s) +} +func (busStore) Put(handle string, s Shown) error { + _, err := stdio.State("shown").Put(handle, s) + return err +} +func (busStore) Handles() ([]string, error) { return stdio.State("shown").Keys() } + +// machines reads the setting desktop-machines at each use, so a change takes effect without a restart. +func machines(path string) []string { + raw, err := os.ReadFile(path) + if err != nil { + return nil + } + var s struct { + Machines []string `json:"desktop-machines"` + } + if json.Unmarshal(raw, &s) != nil { + return nil + } + var out []string + for _, m := range s.Machines { + if m = strings.TrimSpace(m); m != "" { + out = append(out, m) + } + } + return out +} + +func main() { + settings := os.Getenv("MESH_DESK_SETTINGS") + d := &Desk{ + Machines: func() []string { return machines(settings) }, + Notify: func(machine string, args map[string]any) (json.RawMessage, error) { + return stdio.Ask("seat:node-notifier.send@"+machine, args) + }, + Store: busStore{}, + Publish: func(subject string, body any, id string) error { return stdio.SeatPublish(subject, body, id) }, + Logf: logf, + } + d.init() + go run(d) + tools := []stdio.Tool{{Name: "desk_channel_status", + Description: "Whether the desk can show the operator's messages, and on which machines: the setting " + + "desktop-machines, each machine's last answer, and the rules it keeps.", + Run: func(map[string]any) (any, error) { return d.Status(), nil }}} + if err := stdio.Serve("", tools); err != nil { + logf("%v", err) + os.Exit(1) + } +} + +func run(d *Desk) { + 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) { + if err := stdio.SeatTake(Worker, func(w stdio.Work) error { return d.Take(w.Subject, w.Body) }); err == nil { + logf("[desk-channel] taking the router's messages for the desk") + return + } else { + logf("[desk-channel] not taking messages yet (%v); asking again in %s", err, wait) + } + time.Sleep(wait) + } + }() + go func() { + for { + d.SayStanding() + time.Sleep(time.Minute) + } + }() + handle := func(e stdio.Envelope) error { + var body struct { + Token string `json:"token"` + Machine string `json:"machine"` + } + _ = json.Unmarshal(e.Body, &body) + // The machine is the event's own node, said by the runtime: the body's word is only a hint. + machine := e.Node + if machine == "" { + machine = body.Machine + } + switch e.Key { + case StartedEvent: + d.Started(machine) + case AnswerEvent: + if body.Token != "" { + d.Chosen(e.Headers["x-event-id"], machine, body.Token) + } + } + return nil + } + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + if err := stdio.Subscribe("dunst.*", handle); err == nil { + logf("[desk-channel] listening for the answers chosen on the desk") + return + } else { + logf("[desk-channel] not hearing the desk's answers yet (%v); asking again in %s", err, wait) + } + time.Sleep(wait) + } +} diff --git a/modules/desk-channel/go.mod b/modules/desk-channel/go.mod new file mode 100644 index 00000000..44601787 --- /dev/null +++ b/modules/desk-channel/go.mod @@ -0,0 +1,5 @@ +module deskchannel + +go 1.22 + +require git.novox.be/novox/mesh-sdk/go v0.1.8-0.20261008145004-62367ce15ad6 diff --git a/modules/desk-channel/go.sum b/modules/desk-channel/go.sum new file mode 100644 index 00000000..1d160856 --- /dev/null +++ b/modules/desk-channel/go.sum @@ -0,0 +1,2 @@ +git.novox.be/novox/mesh-sdk/go v0.1.8-0.20261008145004-62367ce15ad6 h1:JT7xM1bnLNInW7/oImV2OlXTrcQ4/GSM0Y8tAb+AhmY= +git.novox.be/novox/mesh-sdk/go v0.1.8-0.20261008145004-62367ce15ad6/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= diff --git a/modules/desk-channel/module.json b/modules/desk-channel/module.json new file mode 100644 index 00000000..04e1e21f --- /dev/null +++ b/modules/desk-channel/module.json @@ -0,0 +1,76 @@ +{ + "module": "desk-channel", + "version": "1", + "slug": "desk", + "claims": [ + { + "name": "channel", + "scope": "mesh", + "kind": "desktop", + "capabilities": [ + "deliver", + "silent", + "loud", + "edit", + "private", + "choice", + "exact-render" + ] + }, + { + "name": "intake", + "scope": "mesh", + "kind": "desktop" + } + ], + "consumes": [ + "dunst.action-chosen", + "dunst.started" + ], + "invokes": [ + "seat:node-notifier.send" + ], + "state": [ + { + "name": "shown", + "ttl-seconds": 2592000 + } + ], + "tools": [ + "desk_channel_status" + ], + "resources": [ + { + "id": "state", + "type": "directory", + "mode": "0700", + "place": "." + }, + { + "id": "settings", + "type": "file", + "path": "${dir:state}/settings.json", + "mode": "0600", + "merge": "json", + "content": "{\n \"desktop-machines\": []\n}\n" + } + ], + "build": { + "artifacts": [ + { + "name": "tools", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/desk-channel", + "binary": "desk-channel", + "loads": [ + "desk-channel" + ], + "env": { + "MESH_DESK_SETTINGS": "${dir:state}/settings.json" + } + } + ] + } +} diff --git a/modules/telegram/README.md b/modules/telegram/README.md new file mode 100644 index 00000000..17056b21 --- /dev/null +++ b/modules/telegram/README.md @@ -0,0 +1,69 @@ +# telegram + +A channel of the operator's conversation: kind `telegram` of the `channel` and `intake` benches the router +(`messenger`) declares (novox/hq ADR 0234, ADR 0259). It reaches the operator away from the machines, and it +proves who answered: Telegram authenticates the sender, and the router compares it with the account linked +as the operator. + +- **Shows what the router sends** (its kind's worker on the `channel` queue): a message, a question with + buttons — each button carries only the router's one-time ticket — or an edit of a message shown before. + It sends to the account the router names as linked, or, before one is, to the chat a `/start` came from. +- **Says what the operator chose** on `intake` under its own kind: a tapped button (`choice`), with the + sender as Telegram authenticated it; `/start` (`link`); a message it could not deliver (`failed`); and + whether it can send (`standing`). Only a private chat with the account itself counts. +- **A code** (six digits, the one shown on the desk when linking) is asked of the router as a proof, by + request and reply: never in an event, never kept, and deleted from the chat at once. +- **Judges nothing.** Whether the sender is the operator, and what an answer may do, is the router's. + +## The defects of the client it replaces (novox/hq to-be 46 §12) + +| | fixed here | +|---|---| +| D1 | up to 4096 characters arrive whole; longer is cut and the cut said | +| D4 | a warning or a clearing arrives without sound; an urgent message or a question rings | +| D5 | a rate limit is waited out as told | +| D6 | a refusal is a permanent failure, said to the router and not retried; an edit to the same text is a success | +| D7 | the deprecated preview flag is gone | +| D8 | a request that cannot be written is said | +| D10 | `standing` checks the token with Telegram (`getMe`), not only that a value is present | + +D2 and D3 (a reopening said by an edit; a clearing edits the first message) are the router's, and D9 the +watcher's. + +## How it keeps the token from the agents (novox/hq ADR 0259 §8) + +- **It runs as the account `telegram`**, which the module makes (no shell, no login). The tool runner, which + runs as root, starts this bundle as that account (`MESH_RUN_AS`) and refuses root and the operator's + account. +- **Its state directory is the account's, 0700, and its secret `telegram-token` the account's, 0600.** The + operator's account — which every agent runs as — reads neither, and cannot attach to its process. +- **The bus credential stays the runner's**, root's only. The bundle reaches the bus through its own + standard input and output, and the runner publishes for it only its own kind's subjects. +- **An agent reaches the runner only through the mesh MCP server's tools**, which make tool calls; no tool + says a choice, a link or a proof. +- **If an agent holds root on the machine, none of this holds**: root reads the token and the runner's bus + credential and can publish anything. + +## Where it runs + +Held once for the mesh: assign it to the control node, the stably reachable one. Long polling is outbound +only, so no route into the mesh is needed, and a stolen token can read updates (a conflict says so) but +never inject one. + +## What the operator gives + +1. **A bot**: in Telegram, talk to @BotFather, `/newbot`, a name and a username. BotFather answers with the + bot's token. Keep it on the phone; never paste it into a chat with an agent. +2. **The token, to the vault**, as this module's own secret `telegram-token` (novox/hq ADR 0259 §10, the + vault's verb for a secret given at the desk; until it is built, `secret accept telegram + telegram-token` at a terminal on the control node, which reads it without echoing it). Then push the + control node. +3. **The link**: send the bot `/start`. A six-digit code appears on the desk; send it to the bot within ten + minutes. The account answers acknowledgements at once and approves an hour later; every channel says so. + `messenger_identities` shows it, `messenger_unlink` undoes it. + +## Tools + +| tool | does | +|---|---| +| `telegram_status` | whether it can send and why not (never the token), the bot's name, when updates last arrived, the last delivery and error, counts since start, and its rules | diff --git a/modules/telegram/cmd/telegram/bot.go b/modules/telegram/cmd/telegram/bot.go new file mode 100644 index 00000000..9a05a56a --- /dev/null +++ b/modules/telegram/cmd/telegram/bot.go @@ -0,0 +1,290 @@ +package main + +// The Bot API, as this channel uses it. The bot token is this module's own secret, read from its file at +// each use, owned by the account this module runs as and by no other (novox/hq ADR 0259 §8). It is never +// said: an error from the HTTP client carries the URL, and the URL carries the token, so every error is +// rebuilt here from its kind before it leaves this file, and a description the API gives back has the +// token taken out. + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "net" + "net/http" + "net/url" + "os" + "regexp" + "strings" + "time" +) + +// API is where the Bot API answers; a test points it elsewhere. +var API = "https://api.telegram.org" + +var tokenShape = regexp.MustCompile(`^\d{5,}:[A-Za-z0-9_-]{30,}$`) + +// Bot calls the Bot API with the token in its file. +type Bot struct { + TokenFile string + Client *http.Client +} + +// apiError is the API's refusal: its code, its words, and how long it asks to be left alone. +type apiError struct { + Method string + Code int + Words string + RetryAfter time.Duration +} + +func (e *apiError) Error() string { + return fmt.Sprintf("telegram %s refused: %d %s", e.Method, e.Code, e.Words) +} + +// permanent says whether trying again cannot help: a refusal of the request itself (400, 403) or of the +// token (401, 404), never a rate limit or the service failing (D6). +func permanent(err error) bool { + var a *apiError + return errors.As(err, &a) && (a.Code == 400 || a.Code == 401 || a.Code == 403 || a.Code == 404) +} + +func retryAfter(err error) time.Duration { + var a *apiError + if errors.As(err, &a) && a.Code == 429 { + if a.RetryAfter > 0 { + return a.RetryAfter + } + return time.Second + } + return 0 +} + +// notModified is an edit to the text it already has: a success, never a reason to send again (D6). +func notModified(err error) bool { + var a *apiError + return errors.As(err, &a) && a.Code == 400 && strings.Contains(a.Words, "message is not modified") +} + +func gone(err error) bool { + var a *apiError + return errors.As(err, &a) && a.Code == 400 && (strings.Contains(a.Words, "message to edit not found") || + strings.Contains(a.Words, "message to delete not found")) +} + +// token reads the bot token, and says in words what is missing when it cannot: never the value. +func (b *Bot) token() (string, error) { + if b.TokenFile == "" { + return "", errors.New("no bot token: this module was started without a token file") + } + raw, err := os.ReadFile(b.TokenFile) + if errors.Is(err, os.ErrNotExist) { + return "", errors.New("no bot token: the own secret telegram-token is not on this machine yet; the operator " + + "gives it to the vault (the module's README says how)") + } + if err != nil { + return "", errors.New("the own secret telegram-token cannot be read: " + plainError(err)) + } + token := strings.TrimSpace(string(raw)) + if token == "" { + return "", errors.New("no bot token: the own secret telegram-token is empty") + } + if !tokenShape.MatchString(token) { + // The mesh makes a random value for an own secret nobody gave; that is not a bot token. + return "", errors.New("the own secret telegram-token is not a bot token (digits, a colon, then the key " + + "BotFather gave): most likely the mesh made it because none was given; the operator gives it to the vault") + } + return token, nil +} + +// call asks the API one method, decoding its result into `into`. +func (b *Bot) call(method string, body map[string]any, into any) error { + token, err := b.token() + if err != nil { + return err + } + raw, err := json.Marshal(body) + if err != nil { + // Said, never discarded (D8). + return fmt.Errorf("telegram %s: the request could not be written: %v", method, err) + } + req, err := http.NewRequest(http.MethodPost, API+"/bot"+token+"/"+method, bytes.NewReader(raw)) + if err != nil { + return errors.New("telegram " + method + ": the request could not be made") + } + req.Header.Set("Content-Type", "application/json") + client := b.Client + if client == nil { + client = &http.Client{Timeout: 20 * time.Second} + } + resp, err := client.Do(req) + if err != nil { + return fmt.Errorf("telegram %s: %s", method, plainError(err)) + } + defer resp.Body.Close() + var answer struct { + OK bool `json:"ok"` + ErrorCode int `json:"error_code"` + Description string `json:"description"` + Result json.RawMessage `json:"result"` + Parameters struct { + RetryAfter int `json:"retry_after"` + } `json:"parameters"` + } + if err := json.NewDecoder(resp.Body).Decode(&answer); err != nil { + return fmt.Errorf("telegram %s: HTTP %d with an answer that is not the Bot API's", method, resp.StatusCode) + } + if !answer.OK { + return &apiError{Method: method, Code: answer.ErrorCode, Words: strings.ReplaceAll(answer.Description, token, ""), + RetryAfter: time.Duration(answer.Parameters.RetryAfter) * time.Second} + } + if into != nil && len(answer.Result) > 0 { + if err := json.Unmarshal(answer.Result, into); err != nil { + return fmt.Errorf("telegram %s: its answer could not be read: %v", method, err) + } + } + return nil +} + +// plainError is an error in words, without the URL a transport error carries. +func plainError(err error) string { + var ue *url.Error + if errors.As(err, &ue) { + err = ue.Err + } + var ne net.Error + switch { + case errors.As(err, &ne) && ne.Timeout(): + return "no answer in time" + case errors.Is(err, os.ErrPermission): + return "permission denied" + } + var dns *net.DNSError + if errors.As(err, &dns) { + return "the Bot API's name does not resolve" + } + var op *net.OpError + if errors.As(err, &op) { + return "cannot connect (" + op.Op + ")" + } + s := err.Error() + if strings.Contains(s, "/bot") || strings.Contains(s, "://") { + return "the request failed" + } + return s +} + +// User is a Telegram account as the API says it. +type User struct { + ID int64 `json:"id"` + IsBot bool `json:"is_bot"` + FirstName string `json:"first_name"` + LastName string `json:"last_name"` + Username string `json:"username"` +} + +// Display is the name the service shows for an account. +func (u User) Display() string { + name := strings.TrimSpace(u.FirstName + " " + u.LastName) + if u.Username != "" { + if name != "" { + return name + " (@" + u.Username + ")" + } + return "@" + u.Username + } + return name +} + +// Chat is where a message was written. +type Chat struct { + ID int64 `json:"id"` + Type string `json:"type"` +} + +// Msg is a message as the API says it. +type Msg struct { + MessageID int64 `json:"message_id"` + From *User `json:"from"` + Chat Chat `json:"chat"` + Text string `json:"text"` +} + +// Callback is a button tapped. +type Callback struct { + ID string `json:"id"` + From User `json:"from"` + Message *Msg `json:"message"` + Data string `json:"data"` +} + +// Update is one thing that happened to the bot. +type Update struct { + UpdateID int64 `json:"update_id"` + Message *Msg `json:"message"` + Callback *Callback `json:"callback_query"` +} + +// Updates long-polls for what happened since offset: outbound only, so no route into the mesh is needed, and +// a stolen token can read updates (a conflict says so) but never inject one (novox/hq ADR 0234). +func (b *Bot) Updates(offset int64, wait time.Duration) ([]Update, error) { + var out []Update + err := b.call("getUpdates", map[string]any{"offset": offset, "timeout": int(wait.Seconds()), + "allowed_updates": []string{"message", "callback_query"}}, &out) + return out, err +} + +// Me is the bot's own account: the check that the token works, with nothing sent. +func (b *Bot) Me() (User, error) { + var me User + return me, b.call("getMe", map[string]any{}, &me) +} + +// keyboard is buttons one to a row, each carrying only its ticket. +func keyboard(buttons [][2]string) map[string]any { + rows := [][]map[string]string{} + for _, b := range buttons { + rows = append(rows, []map[string]string{{"text": b[0], "callback_data": b[1]}}) + } + return map[string]any{"inline_keyboard": rows} +} + +// Send posts a message and answers its id. silent: it arrives without sound (D4). +func (b *Bot) Send(chat int64, text string, buttons [][2]string, silent bool) (int64, error) { + body := map[string]any{"chat_id": chat, "text": text, "disable_notification": silent, + "link_preview_options": map[string]any{"is_disabled": true}} + if len(buttons) > 0 { + body["reply_markup"] = keyboard(buttons) + } + var out Msg + err := b.call("sendMessage", body, &out) + return out.MessageID, err +} + +// Edit replaces a message's text and buttons; none given, the buttons go. +func (b *Bot) Edit(chat, id int64, text string, buttons [][2]string) error { + err := b.call("editMessageText", map[string]any{"chat_id": chat, "message_id": id, "text": text, + "reply_markup": keyboard(buttons), "link_preview_options": map[string]any{"is_disabled": true}}, nil) + if notModified(err) { + return nil + } + return err +} + +// AnswerTap ends the phone's spinner for a tapped button, at once. +func (b *Bot) AnswerTap(id, text string) error { + body := map[string]any{"callback_query_id": id} + if text != "" { + body["text"] = text + } + return b.call("answerCallbackQuery", body, nil) +} + +// Delete removes a message from the chat: a code once read. +func (b *Bot) Delete(chat, id int64) error { + err := b.call("deleteMessage", map[string]any{"chat_id": chat, "message_id": id}, nil) + if gone(err) { + return nil + } + return err +} diff --git a/modules/telegram/cmd/telegram/channel.go b/modules/telegram/cmd/telegram/channel.go new file mode 100644 index 00000000..1819c155 --- /dev/null +++ b/modules/telegram/cmd/telegram/channel.go @@ -0,0 +1,453 @@ +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. `/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 ( + "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 +} + +// 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 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 + 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 + // At once, whatever comes of it: the phone shows a spinner until then. + if err := c.Bot.AnswerTap(cb.ID, ""); err != nil { + c.Logf("[telegram] a tap could not be acknowledged: %v", err) + } + if cb.Message == nil || !private(&cb.From, cb.Message.Chat) { + c.Logf("[telegram] a tap outside a private chat was dropped") + return + } + 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) 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 + } + } + if len(updates) > 0 { + 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, which alone reads the bot token", + } + return out +} diff --git a/modules/telegram/cmd/telegram/channel_test.go b/modules/telegram/cmd/telegram/channel_test.go new file mode 100644 index 00000000..818f42a8 --- /dev/null +++ b/modules/telegram/cmd/telegram/channel_test.go @@ -0,0 +1,365 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "reflect" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" +) + +const goodToken = "123456789:AAEhBP0av28P4XFQnIuR-o-7Xnz1kkUzW3g" + +// fakeAPI is the Bot API, answering each method as the test says and keeping what it was asked. +type fakeAPI struct { + mu sync.Mutex + calls []call + answer map[string]func(body map[string]any) (int, string) + next int64 +} + +type call struct { + Method string + Body map[string]any +} + +func (f *fakeAPI) of(method string) []call { + f.mu.Lock() + defer f.mu.Unlock() + var out []call + for _, c := range f.calls { + if c.Method == method { + out = append(out, c) + } + } + return out +} + +func newAPI(t *testing.T) *fakeAPI { + t.Helper() + f := &fakeAPI{answer: map[string]func(map[string]any) (int, string){}} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + parts := strings.Split(r.URL.Path, "/") + method := parts[len(parts)-1] + if parts[1] != "bot"+goodToken { + w.WriteHeader(401) + _, _ = w.Write([]byte(`{"ok":false,"error_code":401,"description":"Unauthorized"}`)) + return + } + var body map[string]any + _ = json.NewDecoder(r.Body).Decode(&body) + f.mu.Lock() + f.calls = append(f.calls, call{method, body}) + answer := f.answer[method] + f.next++ + id := f.next + f.mu.Unlock() + if answer != nil { + code, out := answer(body) + w.WriteHeader(code) + _, _ = w.Write([]byte(out)) + return + } + switch method { + case "sendMessage": + _, _ = w.Write([]byte(`{"ok":true,"result":{"message_id":` + jsonInt(id) + `,"chat":{"id":42,"type":"private"}}}`)) + case "getMe": + _, _ = w.Write([]byte(`{"ok":true,"result":{"id":1,"is_bot":true,"username":"mesh_bot"}}`)) + default: + _, _ = w.Write([]byte(`{"ok":true,"result":true}`)) + } + })) + t.Cleanup(srv.Close) + was := API + API = srv.URL + t.Cleanup(func() { API = was }) + return f +} + +func jsonInt(n int64) string { b, _ := json.Marshal(n); return string(b) } + +type memStore struct { + offset int64 + placed map[string]Placed +} + +func (m *memStore) Offset() (int64, error) { return m.offset, nil } +func (m *memStore) SetOffset(n int64) error { m.offset = n; return nil } +func (m *memStore) Place(h string, p Placed) error { + if m.placed == nil { + m.placed = map[string]Placed{} + } + m.placed[h] = p + return nil +} +func (m *memStore) Placed(h string) (*Placed, error) { + p, ok := m.placed[h] + if !ok { + return nil, nil + } + return &p, nil +} + +type said struct { + mu sync.Mutex + by map[string][]json.RawMessage + proved []json.RawMessage +} + +func (s *said) of(subject string) []json.RawMessage { + s.mu.Lock() + defer s.mu.Unlock() + return s.by[subject] +} + +func newChannel(t *testing.T, token string) (*Channel, *fakeAPI, *said, *memStore) { + t.Helper() + f := newAPI(t) + file := filepath.Join(t.TempDir(), "telegram-token") + if token != "" { + if err := os.WriteFile(file, []byte(token+"\n"), 0o600); err != nil { + t.Fatal(err) + } + } + s := &said{by: map[string][]json.RawMessage{}} + st := &memStore{} + now := time.Date(2026, 10, 8, 14, 0, 0, 0, time.UTC) + c := &Channel{Bot: &Bot{TokenFile: file}, Store: st, Now: func() time.Time { return now }, Logf: t.Logf, + Sleep: func(time.Duration) {}, + Publish: func(subject string, body any, id string) error { + raw, _ := json.Marshal(body) + s.mu.Lock() + s.by[subject] = append(s.by[subject], raw) + s.mu.Unlock() + return nil + }, + Prove: func(subject string, body any) (json.RawMessage, error) { + raw, _ := json.Marshal(body) + s.mu.Lock() + s.proved = append(s.proved, raw) + s.mu.Unlock() + if strings.Contains(string(raw), `"code":"123456"`) { + return json.RawMessage(`{"accepted":true,"words":"Linked."}`), nil + } + return json.RawMessage(`{"accepted":false,"words":"That code is not right."}`), nil + }} + c.init() + return c, f, s, st +} + +func work(verb string, m asks.Message) (string, json.RawMessage) { + raw, _ := json.Marshal(m) + return asks.ChannelSubject(verb, Kind), raw +} + +func TestAQuestionIsShownWithItsTicketsAndRingsAndAWarningIsSilent(t *testing.T) { + c, f, _, st := newChannel(t, goodToken) + subject, body := work(asks.Show, asks.Message{Handle: "m1", Title: "Delivery of hq held", Body: "Needs you: release it.", + To: "42", Buttons: []asks.Button{{Label: "Release", Ticket: "0123456789abcdef0123456789abcdef"}}}) + if err := c.Take(subject, body); err != nil { + t.Fatal(err) + } + sent := f.of("sendMessage") + if len(sent) != 1 { + t.Fatalf("sent %d", len(sent)) + } + b := sent[0].Body + kb, _ := json.Marshal(b["reply_markup"]) + if b["chat_id"] != float64(42) || b["text"] != "Delivery of hq held\nNeeds you: release it." || b["disable_notification"] != false || + string(kb) != `{"inline_keyboard":[[{"callback_data":"0123456789abcdef0123456789abcdef","text":"Release"}]]}` { + t.Errorf("sent %v %s", b, kb) + } + if _, deprecated := b["disable_web_page_preview"]; deprecated { + t.Error("the deprecated preview flag is still sent (D7)") + } + if st.placed["m1"].ID == 0 { + t.Error("where it is was not kept") + } + subject, body = work(asks.Send, asks.Message{Handle: "m2", Title: "Resolved: hq works again", To: "42"}) + _ = c.Take(subject, body) + if f.of("sendMessage")[1].Body["disable_notification"] != true { + t.Error("a clearing rang (D4)") + } + subject, body = work(asks.Send, asks.Message{Handle: "m3", Title: "Urgent: ace silent", Urgent: true, To: "42"}) + _ = c.Take(subject, body) + if f.of("sendMessage")[2].Body["disable_notification"] != false { + t.Error("an urgent message was silent (D4)") + } +} + +func TestALongMessageIsCutAndTheCutSaid(t *testing.T) { + c, f, _, _ := newChannel(t, goodToken) + subject, body := work(asks.Send, asks.Message{Handle: "m1", Title: "t", Body: strings.Repeat("x", 5000), To: "42"}) + if err := c.Take(subject, body); err != nil { + t.Fatal(err) + } + got := f.of("sendMessage")[0].Body["text"].(string) + if n := len([]rune(got)); n != MaxLength || !strings.HasSuffix(got, cutNote) { + t.Errorf("%d characters, ends %q (D1)", n, got[len(got)-40:]) + } +} + +func TestAnEditToTheSameTextIsASuccessAndAGoneMessageIsSentAgain(t *testing.T) { + c, f, s, st := newChannel(t, goodToken) + st.Place("m1", Placed{Chat: 42, ID: 7}) + f.answer["editMessageText"] = func(map[string]any) (int, string) { + return 400, `{"ok":false,"error_code":400,"description":"Bad Request: message is not modified"}` + } + subject, body := work(asks.Edit, asks.Message{Handle: "m1", Title: "same", To: "42"}) + if err := c.Take(subject, body); err != nil || len(f.of("sendMessage")) != 0 || len(s.of(asks.IntakeSubject(asks.FailedWhat, Kind))) != 0 { + t.Fatalf("an identical edit was not a success (D6): %v", err) + } + f.answer["editMessageText"] = func(map[string]any) (int, string) { + return 400, `{"ok":false,"error_code":400,"description":"Bad Request: message to edit not found"}` + } + _ = c.Take(subject, body) + if len(f.of("sendMessage")) != 1 { + t.Error("an edit of a message gone was not sent again") + } + if kb, _ := json.Marshal(f.of("editMessageText")[0].Body["reply_markup"]); string(kb) != `{"inline_keyboard":[]}` { + t.Errorf("an edit without buttons left them: %s", kb) + } +} + +func TestARefusalIsPermanentARateLimitIsWaitedOutAndNeitherCarriesTheToken(t *testing.T) { + c, f, s, _ := newChannel(t, goodToken) + tries := 0 + var slept []time.Duration + c.Sleep = func(d time.Duration) { slept = append(slept, d) } + f.answer["sendMessage"] = func(map[string]any) (int, string) { + tries++ + if tries == 1 { + return 429, `{"ok":false,"error_code":429,"description":"Too Many Requests","parameters":{"retry_after":7}}` + } + return 403, `{"ok":false,"error_code":403,"description":"Forbidden: bot was blocked by the user ` + goodToken + `"}` + } + subject, body := work(asks.Send, asks.Message{Handle: "m1", Title: "x", To: "42"}) + if err := c.Take(subject, body); err != nil { + t.Fatal(err) + } + if len(slept) != 1 || slept[0] != 7*time.Second || tries != 2 { + t.Errorf("the rate limit was not waited out as told (D5): slept %v, tries %d", slept, tries) + } + failed := s.of(asks.IntakeSubject(asks.FailedWhat, Kind)) + if len(failed) != 1 || !strings.Contains(string(failed[0]), `"permanent":true`) || strings.Contains(string(failed[0]), goodToken) { + t.Fatalf("the refusal: %s (D6)", failed) + } + // No linked account and nothing to reply to: said as failed for good, never retried. + subject, body = work(asks.Send, asks.Message{Handle: "m2", Title: "x"}) + _ = c.Take(subject, body) + if f := s.of(asks.IntakeSubject(asks.FailedWhat, Kind)); len(f) != 2 || !strings.Contains(string(f[1]), "no account is linked") { + t.Errorf("%s", f) + } +} + +func TestATapIsAcknowledgedAtOnceAndSaidWithItsAuthenticatedSender(t *testing.T) { + c, f, s, _ := newChannel(t, goodToken) + c.Handle(Update{UpdateID: 1, Callback: &Callback{ID: "cb1", Data: "tkt", From: User{ID: 42, FirstName: "Jo"}, + Message: &Msg{MessageID: 7, Chat: Chat{ID: 42, Type: "private"}}}}) + if len(f.of("answerCallbackQuery")) != 1 { + t.Error("the tap was not acknowledged") + } + got := s.of(asks.IntakeSubject(asks.Chosen, Kind)) + var ch asks.Choice + if len(got) != 1 || json.Unmarshal(got[0], &ch) != nil || ch.Ticket != "tkt" || + !reflect.DeepEqual(ch.Sender, asks.Sender{Identity: "42", Display: "Jo", Authenticated: true}) { + t.Fatalf("said %s", got) + } + // In a group, or from another bot, nothing is said: it is nobody's own chat with the bot. + c.Handle(Update{UpdateID: 2, Callback: &Callback{ID: "cb2", Data: "tkt", From: User{ID: 42}, + Message: &Msg{Chat: Chat{ID: -100, Type: "group"}}}}) + c.Handle(Update{UpdateID: 3, Callback: &Callback{ID: "cb3", Data: "tkt", From: User{ID: 9, IsBot: true}, + Message: &Msg{Chat: Chat{ID: 9, Type: "private"}}}}) + if len(s.of(asks.IntakeSubject(asks.Chosen, Kind))) != 1 { + t.Error("a tap outside the account's own chat was said") + } +} + +func TestStartAsksALinkAndACodeIsAProofDeletedFromTheChat(t *testing.T) { + c, f, s, _ := newChannel(t, goodToken) + jo := &User{ID: 42, FirstName: "Jo", Username: "jo"} + c.Handle(Update{UpdateID: 10, Message: &Msg{MessageID: 5, From: jo, Chat: Chat{ID: 42, Type: "private"}, Text: "/start"}}) + var l asks.LinkAsked + got := s.of(asks.IntakeSubject(asks.Link, Kind)) + if len(got) != 1 || json.Unmarshal(got[0], &l) != nil || l.Sender.Identity != "42" || l.Sender.Display != "Jo (@jo)" { + t.Fatalf("link %s", got) + } + // The router's answer to the link goes to the chat it came from, though nobody is linked yet. + subject, body := work(asks.Send, asks.Message{Handle: "r1", Title: "Send the code shown on your desk", Reply: l.ID}) + if err := c.Take(subject, body); err != nil || len(f.of("sendMessage")) != 1 || f.of("sendMessage")[0].Body["chat_id"] != float64(42) { + t.Fatalf("the reply did not reach the chat: %v %v", err, f.of("sendMessage")) + } + c.Handle(Update{UpdateID: 11, Message: &Msg{MessageID: 6, From: jo, Chat: Chat{ID: 42, Type: "private"}, Text: "123 456"}}) + if len(s.proved) != 1 || !strings.Contains(string(s.proved[0]), `"code":"123456"`) { + t.Fatalf("the code was not asked as a proof: %s", s.proved) + } + for subject, events := range s.by { + for _, e := range events { + if strings.Contains(string(e), "123456") { + t.Errorf("the code was said on %s", subject) + } + } + } + if d := f.of("deleteMessage"); len(d) != 1 || d[0].Body["message_id"] != float64(6) { + t.Error("the code was not deleted from the chat") + } + if last := f.of("sendMessage"); last[len(last)-1].Body["text"] != "Linked." { + t.Errorf("the router's answer was not said: %v", last[len(last)-1].Body) + } + c.Handle(Update{UpdateID: 12, Message: &Msg{MessageID: 7, From: jo, Chat: Chat{ID: 42, Type: "private"}, Text: "release it"}}) + if last := f.of("sendMessage"); last[len(last)-1].Body["text"] != FixedLine { + t.Error("other words were not answered with the fixed line") + } + if c.Store.(*memStore).offset != 0 { + t.Error("the offset moved without a poll") + } +} + +func TestStandingSaysWhatIsMissingAndNeverTheToken(t *testing.T) { + for token, want := range map[string]string{ + "": "no bot token", + "a-random-value-x": "not a bot token", + "987654321:" + strings.Repeat("Z", 35): "does not work", + } { + c, _, _, _ := newChannel(t, token) + s := c.Standing() + if s.Ready || !strings.Contains(s.Why, want) || (token != "" && strings.Contains(s.Why, token)) { + t.Errorf("%q: %+v", token, s) + } + } + c, _, said, _ := newChannel(t, goodToken) + if s := c.Standing(); !s.Ready || !s.EditsSilently { + t.Errorf("%+v", s) + } + c.SayStanding() + c.SayStanding() + if n := len(said.of(asks.IntakeSubject(asks.StandingWhat, Kind))); n != 1 { + t.Errorf("standing said %d times without a change", n) + } +} + +func TestPollTakesTheUpdatesAndKeepsTheOffsetAndSaysAConflict(t *testing.T) { + c, f, s, st := newChannel(t, goodToken) + f.answer["getUpdates"] = func(map[string]any) (int, string) { + return 200, `{"ok":true,"result":[{"update_id":41,"message":{"message_id":1,"from":{"id":42},"chat":{"id":42,"type":"private"},"text":"/start"}}]}` + } + if err := c.Poll(time.Second); err != nil || st.offset != 42 || len(s.of(asks.IntakeSubject(asks.Link, Kind))) != 1 { + t.Fatalf("%v offset %d", err, st.offset) + } + f.answer["getUpdates"] = func(map[string]any) (int, string) { + return 409, `{"ok":false,"error_code":409,"description":"Conflict: terminated by other getUpdates request"}` + } + if err := c.Poll(time.Second); err == nil || !strings.Contains(err.Error(), "another program takes this bot's updates") { + t.Errorf("%v", err) + } + if s := c.Standing(); s.Ready { + t.Error("ready while the updates do not arrive") + } +} diff --git a/modules/telegram/cmd/telegram/main.go b/modules/telegram/cmd/telegram/main.go new file mode 100644 index 00000000..fc94243c --- /dev/null +++ b/modules/telegram/cmd/telegram/main.go @@ -0,0 +1,115 @@ +// 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 +} + +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 = "" + } +} diff --git a/modules/telegram/cmd/telegram/manifest_test.go b/modules/telegram/cmd/telegram/manifest_test.go new file mode 100644 index 00000000..19da5ee7 --- /dev/null +++ b/modules/telegram/cmd/telegram/manifest_test.go @@ -0,0 +1,67 @@ +package main + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" +) + +// The manifest keeps the bot token and this channel's state from every agent (novox/hq ADR 0259 §8): the +// bundle runs as the module's own account, which owns its state directory and its secret; it claims one kind +// on both benches, with what it promises; and it names nothing of one installation. +func TestItRunsAsItsOwnAccountWhichAloneHoldsTheToken(t *testing.T) { + raw, err := os.ReadFile(filepath.Join("..", "..", "module.json")) + if err != nil { + t.Fatal(err) + } + var m struct { + Claims []struct { + Name, Scope, Kind string + Capabilities []string + } + Own map[string]string `json:"own-secrets"` + Owner string `json:"secrets-owner"` + Resources []map[string]any + Build struct { + Artifacts []struct { + Env map[string]string + } + } + } + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + if len(m.Claims) != 2 || m.Claims[0].Name != "channel" || m.Claims[0].Kind != Kind || m.Claims[1].Name != "intake" || + m.Claims[1].Kind != Kind { + t.Fatalf("claims %+v", m.Claims) + } + promises := strings.Join(m.Claims[0].Capabilities, " ") + for _, c := range []string{"reaches-away", "choice", "verified-sender", "edit", "max-length:4096"} { + if !strings.Contains(promises, c) { + t.Errorf("it does not promise %s", c) + } + } + env := m.Build.Artifacts[0].Env + if env["MESH_RUN_AS"] != "telegram" || env["MESH_TELEGRAM_TOKEN_FILE"] != m.Own["telegram-token"] || m.Owner != "telegram" { + t.Fatalf("env %v, secret %v, owner %q", env, m.Own, m.Owner) + } + var account, state bool + for _, r := range m.Resources { + switch r["type"] { + case "user": + account = r["name"] == "telegram" && r["shell"] == "/usr/bin/nologin" + case "directory": + state = r["id"] == "state" && r["owner"] == "telegram" && r["mode"] == "0700" + } + } + if !account || !state { + t.Errorf("the account %v, its state %v", account, state) + } + for _, never := range []string{"/home/", "jochen", "g14", "shanks", "novox", "api.telegram"} { + if strings.Contains(strings.ToLower(string(raw)), never) { + t.Errorf("module.json names %q", never) + } + } +} diff --git a/modules/telegram/go.mod b/modules/telegram/go.mod new file mode 100644 index 00000000..633ef03b --- /dev/null +++ b/modules/telegram/go.mod @@ -0,0 +1,5 @@ +module telegram + +go 1.22 + +require git.novox.be/novox/mesh-sdk/go v0.1.8-0.20261008145004-62367ce15ad6 diff --git a/modules/telegram/go.sum b/modules/telegram/go.sum new file mode 100644 index 00000000..1d160856 --- /dev/null +++ b/modules/telegram/go.sum @@ -0,0 +1,2 @@ +git.novox.be/novox/mesh-sdk/go v0.1.8-0.20261008145004-62367ce15ad6 h1:JT7xM1bnLNInW7/oImV2OlXTrcQ4/GSM0Y8tAb+AhmY= +git.novox.be/novox/mesh-sdk/go v0.1.8-0.20261008145004-62367ce15ad6/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= diff --git a/modules/telegram/module.json b/modules/telegram/module.json new file mode 100644 index 00000000..38097f9c --- /dev/null +++ b/modules/telegram/module.json @@ -0,0 +1,77 @@ +{ + "module": "telegram", + "version": "1", + "slug": "tgrm", + "claims": [ + { + "name": "channel", + "scope": "mesh", + "kind": "telegram", + "capabilities": [ + "deliver", + "reaches-away", + "silent", + "edit", + "max-length:4096", + "choice", + "verified-sender", + "exact-render", + "code-factor" + ] + }, + { + "name": "intake", + "scope": "mesh", + "kind": "telegram" + } + ], + "state": [ + "offset", + { + "name": "messages", + "ttl-seconds": 2592000 + } + ], + "own-secrets": { + "telegram-token": "${dir:state}/telegram-token" + }, + "secrets-owner": "telegram", + "tools": [ + "telegram_status" + ], + "resources": [ + { + "id": "account", + "type": "user", + "name": "telegram", + "shell": "/usr/bin/nologin", + "home": "/var/lib/telegram" + }, + { + "id": "state", + "type": "directory", + "mode": "0700", + "owner": "telegram", + "place": "." + } + ], + "build": { + "artifacts": [ + { + "name": "tools", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/telegram", + "binary": "telegram", + "loads": [ + "telegram" + ], + "env": { + "MESH_RUN_AS": "telegram", + "MESH_TELEGRAM_TOKEN_FILE": "${dir:state}/telegram-token" + } + } + ] + } +}