feat(telegram): bot support relay — per-user forum topics

Users who DM the bot anything but /start are relayed into a private forum supergroup, one topic per user. Operators (the chat's admins) reply in the topic and the bot copies it back to the user; an info card opening each topic carries a Block/Unblock toggle and a Clear button. State is a small JSON file on a new /data volume — the bot host has no database. Off by default (TELEGRAM_SUPPORT_CHAT_ID=0): the prod bot is unchanged until the operator sets the chat id and adds the bot as a forum admin. - internal/support: concurrency-safe JSON store (topic map, block list, relayed message ids) with field-targeted mutators and atomic save - bot/support.go: relay both ways via copyMessage, short-TTL admin cache, callback buttons, per-user topic-create lock, loop guard (skip the bot's own posts), reopen a deleted topic on the next message - config + compose + CI/prod-deploy: TELEGRAM_SUPPORT_CHAT_ID per contour + TELEGRAM_SUPPORT_STATE_DIR; bot-state named volume; /data pre-owned by UID 65532 so a fresh volume is writable under distroless - docs: ARCHITECTURE §15 + decision record, FUNCTIONAL (+ru), README
2026-06-23 17:47:00 +02:00
parent e6277dcd43
commit 6a602aefae
18 changed files with 1465 additions and 7 deletions
@@ -0,0 +1,227 @@
+// Package support persists the Telegram bot's support-relay state. For each user
+// who direct-messages the bot it records the dedicated forum topic the bot opened
+// in the operators' support chat, whether the user is blocked, and the ids of the
+// messages relayed into that topic — so an operator's "clear" can delete them while
+// keeping the topic's info card. The state is a small JSON file on a writable
+// volume: the bot host has no database.
+//
+// The store is concurrency-safe. The go-telegram/bot library dispatches each update
+// in its own goroutine, so every exported method locks. Mutators update only their
+// own fields (SetTopic never touches Blocked, SetBlocked never touches the relayed
+// ids) so a topic recreate cannot clobber a concurrent block toggle.
+package support
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"sync"
+	"time"
+)
+
+// User is one user's support-relay state.
+type User struct {
+	// UserID is the user's Telegram user id (the private chat id the bot relays to).
+	UserID int64 `json:"user_id"`
+	// TopicID is the forum topic's message_thread_id in the support chat; 0 means no
+	// topic has been created yet.
+	TopicID int `json:"topic_id"`
+	// HeaderMsgID is the info-card message that opens the topic and carries the
+	// block/clear buttons; it is never deleted by a clear.
+	HeaderMsgID int `json:"header_msg_id"`
+	// Blocked reports whether the bot drops this user's incoming messages.
+	Blocked bool `json:"blocked"`
+	// FirstName, LastName and Username snapshot the user's Telegram profile at the
+	// time the topic was created or last refreshed (for the topic name and info card).
+	FirstName string `json:"first_name,omitempty"`
+	LastName  string `json:"last_name,omitempty"`
+	Username  string `json:"username,omitempty"`
+	// RelayedMsgIDs are the ids of the messages posted into the topic after the header
+	// (the user's relayed messages and the operators' replies); a clear deletes them.
+	RelayedMsgIDs []int `json:"relayed_msg_ids,omitempty"`
+	// CreatedAt is when the topic was first opened for this user.
+	CreatedAt time.Time `json:"created_at"`
+}
+
+// Store is the concurrency-safe, JSON-backed support-relay state. The zero value is
+// not usable; build one with Open or New.
+type Store struct {
+	mu      sync.Mutex
+	path    string
+	users   map[int64]*User
+	byTopic map[int]int64 // TopicID -> UserID, for resolving operator replies
+}
+
+// file is the on-disk shape: a flat list of users. A JSON object keyed by user id
+// would force the int64 keys through strings; a list keeps the ids typed.
+type file struct {
+	Users []*User `json:"users"`
+}
+
+// New returns an empty store that persists to path, creating the parent directory
+// when missing. Use it as the fallback when Open reports a corrupt file.
+func New(path string) *Store {
+	_ = os.MkdirAll(filepath.Dir(path), 0o755)
+	return &Store{path: path, users: map[int64]*User{}, byTopic: map[int]int64{}}
+}
+
+// Open loads the store from path. A missing file yields an empty store and no
+// error; an unreadable or malformed file returns an error so the caller can decide
+// whether to start empty.
+func Open(path string) (*Store, error) {
+	s := New(path)
+	b, err := os.ReadFile(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return s, nil
+		}
+		return nil, fmt.Errorf("support: read state %s: %w", path, err)
+	}
+	var f file
+	if err := json.Unmarshal(b, &f); err != nil {
+		return nil, fmt.Errorf("support: parse state %s: %w", path, err)
+	}
+	for _, u := range f.Users {
+		if u == nil || u.UserID == 0 {
+			continue
+		}
+		s.users[u.UserID] = u
+		if u.TopicID != 0 {
+			s.byTopic[u.TopicID] = u.UserID
+		}
+	}
+	return s, nil
+}
+
+// Get returns a detached copy of the user's state and whether it exists.
+func (s *Store) Get(userID int64) (User, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	u, ok := s.users[userID]
+	if !ok {
+		return User{}, false
+	}
+	cp := *u
+	cp.RelayedMsgIDs = append([]int(nil), u.RelayedMsgIDs...)
+	return cp, true
+}
+
+// ByTopic returns the user id owning the given forum topic, and whether one is known.
+func (s *Store) ByTopic(topicID int) (int64, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	uid, ok := s.byTopic[topicID]
+	return uid, ok
+}
+
+// Blocked reports whether the user's incoming messages are dropped.
+func (s *Store) Blocked(userID int64) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if u, ok := s.users[userID]; ok {
+		return u.Blocked
+	}
+	return false
+}
+
+// SetTopic records the forum topic and info-card message opened for the user and
+// refreshes the profile snapshot, creating the record on first contact. It rebuilds
+// the topic index when the topic changes (a recreate) and preserves the block state
+// and the relayed-message ids. It persists the result.
+func (s *Store) SetTopic(userID int64, topicID, headerMsgID int, firstName, lastName, username string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	u, ok := s.users[userID]
+	if !ok {
+		u = &User{UserID: userID, CreatedAt: time.Now().UTC()}
+		s.users[userID] = u
+	} else if u.TopicID != 0 && u.TopicID != topicID {
+		delete(s.byTopic, u.TopicID)
+	}
+	u.TopicID = topicID
+	u.HeaderMsgID = headerMsgID
+	u.FirstName, u.LastName, u.Username = firstName, lastName, username
+	if topicID != 0 {
+		s.byTopic[topicID] = userID
+	}
+	return s.save()
+}
+
+// SetBlocked sets the user's blocked flag, creating a minimal record when none
+// exists, and persists the result.
+func (s *Store) SetBlocked(userID int64, blocked bool) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	u, ok := s.users[userID]
+	if !ok {
+		u = &User{UserID: userID, CreatedAt: time.Now().UTC()}
+		s.users[userID] = u
+	}
+	u.Blocked = blocked
+	return s.save()
+}
+
+// AppendMsg records a message posted into the user's topic (for a later clear) and
+// persists the result. It is a no-op when the user has no record yet.
+func (s *Store) AppendMsg(userID int64, msgID int) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	u, ok := s.users[userID]
+	if !ok {
+		return nil
+	}
+	u.RelayedMsgIDs = append(u.RelayedMsgIDs, msgID)
+	return s.save()
+}
+
+// ClearMsgs empties and returns the user's relayed-message ids (the info card is not
+// among them) and persists the result, so the caller can delete them from the chat.
+func (s *Store) ClearMsgs(userID int64) ([]int, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	u, ok := s.users[userID]
+	if !ok || len(u.RelayedMsgIDs) == 0 {
+		return nil, nil
+	}
+	ids := u.RelayedMsgIDs
+	u.RelayedMsgIDs = nil
+	if err := s.save(); err != nil {
+		// Roll back so the ids are not lost if the write fails.
+		u.RelayedMsgIDs = ids
+		return nil, err
+	}
+	return ids, nil
+}
+
+// save writes the state atomically (temp file in the same directory, then rename).
+// The caller must hold s.mu.
+func (s *Store) save() error {
+	f := file{Users: make([]*User, 0, len(s.users))}
+	for _, u := range s.users {
+		f.Users = append(f.Users, u)
+	}
+	sort.Slice(f.Users, func(i, j int) bool { return f.Users[i].UserID < f.Users[j].UserID })
+	b, err := json.MarshalIndent(f, "", "  ")
+	if err != nil {
+		return fmt.Errorf("support: marshal state: %w", err)
+	}
+	tmp, err := os.CreateTemp(filepath.Dir(s.path), ".support-*.json")
+	if err != nil {
+		return fmt.Errorf("support: create temp state: %w", err)
+	}
+	tmpName := tmp.Name()
+	defer func() { _ = os.Remove(tmpName) }() // no-op once the rename succeeds
+	if _, err := tmp.Write(b); err != nil {
+		_ = tmp.Close()
+		return fmt.Errorf("support: write temp state: %w", err)
+	}
+	if err := tmp.Close(); err != nil {
+		return fmt.Errorf("support: close temp state: %w", err)
+	}
+	if err := os.Rename(tmpName, s.path); err != nil {
+		return fmt.Errorf("support: replace state %s: %w", s.path, err)
+	}
+	return nil
+}
@@ -0,0 +1,178 @@
+package support
+
+import (
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+)
+
+// statePath returns a path inside a fresh temp directory for a store file.
+func statePath(t *testing.T) string {
+	t.Helper()
+	return filepath.Join(t.TempDir(), "support.json")
+}
+
+func TestOpenMissingReturnsEmpty(t *testing.T) {
+	s, err := Open(statePath(t))
+	if err != nil {
+		t.Fatalf("Open missing: %v", err)
+	}
+	if _, ok := s.Get(42); ok {
+		t.Error("Get on an empty store returned a record")
+	}
+}
+
+func TestOpenCorruptReturnsError(t *testing.T) {
+	path := statePath(t)
+	if err := os.WriteFile(path, []byte("{not json"), 0o600); err != nil {
+		t.Fatalf("write corrupt: %v", err)
+	}
+	if _, err := Open(path); err == nil {
+		t.Fatal("Open on a corrupt file: expected an error, got nil")
+	}
+}
+
+func TestSetTopicPersistsAndReloads(t *testing.T) {
+	path := statePath(t)
+	s := New(path)
+	if err := s.SetTopic(7, 100, 101, "Ann", "Lee", "annlee"); err != nil {
+		t.Fatalf("SetTopic: %v", err)
+	}
+
+	// In-memory view.
+	got, ok := s.Get(7)
+	if !ok || got.TopicID != 100 || got.HeaderMsgID != 101 || got.Username != "annlee" {
+		t.Fatalf("Get = %+v ok=%v, want topic 100 / header 101 / annlee", got, ok)
+	}
+	if got.CreatedAt.IsZero() {
+		t.Error("CreatedAt not set on first contact")
+	}
+	if uid, ok := s.ByTopic(100); !ok || uid != 7 {
+		t.Errorf("ByTopic(100) = %d ok=%v, want 7/true", uid, ok)
+	}
+
+	// Reloaded from disk.
+	reload, err := Open(path)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	got, ok = reload.Get(7)
+	if !ok || got.TopicID != 100 || got.FirstName != "Ann" {
+		t.Fatalf("reloaded Get = %+v ok=%v, want topic 100 / Ann", got, ok)
+	}
+	if uid, ok := reload.ByTopic(100); !ok || uid != 7 {
+		t.Errorf("reloaded ByTopic(100) = %d ok=%v, want 7/true", uid, ok)
+	}
+}
+
+func TestSetTopicRecreateReindexes(t *testing.T) {
+	s := New(statePath(t))
+	if err := s.SetTopic(7, 100, 101, "Ann", "", ""); err != nil {
+		t.Fatalf("SetTopic: %v", err)
+	}
+	if err := s.SetTopic(7, 200, 201, "Ann", "", ""); err != nil {
+		t.Fatalf("SetTopic recreate: %v", err)
+	}
+	if _, ok := s.ByTopic(100); ok {
+		t.Error("ByTopic(100) still resolves after recreate; the stale topic was not dropped")
+	}
+	if uid, ok := s.ByTopic(200); !ok || uid != 7 {
+		t.Errorf("ByTopic(200) = %d ok=%v, want 7/true", uid, ok)
+	}
+}
+
+// TestBlockSurvivesTopicRecreate is the field-isolation guarantee: a topic recreate
+// (SetTopic) must not clear a block set concurrently from the buttons.
+func TestBlockSurvivesTopicRecreate(t *testing.T) {
+	s := New(statePath(t))
+	if err := s.SetTopic(7, 100, 101, "Ann", "", "annlee"); err != nil {
+		t.Fatalf("SetTopic: %v", err)
+	}
+	if err := s.SetBlocked(7, true); err != nil {
+		t.Fatalf("SetBlocked: %v", err)
+	}
+	if err := s.SetTopic(7, 200, 201, "Ann", "", "annlee"); err != nil {
+		t.Fatalf("SetTopic recreate: %v", err)
+	}
+	if !s.Blocked(7) {
+		t.Error("block was cleared by a topic recreate")
+	}
+}
+
+func TestAppendAndClearMsgs(t *testing.T) {
+	s := New(statePath(t))
+	if err := s.SetTopic(7, 100, 101, "Ann", "", ""); err != nil {
+		t.Fatalf("SetTopic: %v", err)
+	}
+	for _, id := range []int{500, 501, 502} {
+		if err := s.AppendMsg(7, id); err != nil {
+			t.Fatalf("AppendMsg %d: %v", id, err)
+		}
+	}
+	got, _ := s.Get(7)
+	if len(got.RelayedMsgIDs) != 3 {
+		t.Fatalf("RelayedMsgIDs = %v, want 3 ids", got.RelayedMsgIDs)
+	}
+
+	ids, err := s.ClearMsgs(7)
+	if err != nil {
+		t.Fatalf("ClearMsgs: %v", err)
+	}
+	if len(ids) != 3 || ids[0] != 500 || ids[2] != 502 {
+		t.Errorf("ClearMsgs = %v, want [500 501 502]", ids)
+	}
+	got, _ = s.Get(7)
+	if len(got.RelayedMsgIDs) != 0 {
+		t.Errorf("RelayedMsgIDs after clear = %v, want empty", got.RelayedMsgIDs)
+	}
+	if got.HeaderMsgID != 101 || got.TopicID != 100 {
+		t.Errorf("clear disturbed the topic/header: %+v", got)
+	}
+
+	// A second clear is a no-op.
+	if ids, err := s.ClearMsgs(7); err != nil || ids != nil {
+		t.Errorf("second ClearMsgs = %v, %v, want nil, nil", ids, err)
+	}
+}
+
+func TestGetReturnsDetachedCopy(t *testing.T) {
+	s := New(statePath(t))
+	if err := s.SetTopic(7, 100, 101, "Ann", "", ""); err != nil {
+		t.Fatalf("SetTopic: %v", err)
+	}
+	if err := s.AppendMsg(7, 500); err != nil {
+		t.Fatalf("AppendMsg: %v", err)
+	}
+	got, _ := s.Get(7)
+	got.RelayedMsgIDs[0] = 999 // mutate the copy
+	again, _ := s.Get(7)
+	if again.RelayedMsgIDs[0] != 500 {
+		t.Error("Get returned a slice that aliases the store's state")
+	}
+}
+
+// TestConcurrentMutationsSafe exercises the mutex under parallel writers; run with
+// -race to catch data races.
+func TestConcurrentMutationsSafe(t *testing.T) {
+	s := New(statePath(t))
+	if err := s.SetTopic(7, 100, 101, "Ann", "", ""); err != nil {
+		t.Fatalf("SetTopic: %v", err)
+	}
+	var wg sync.WaitGroup
+	for i := range 20 {
+		wg.Add(1)
+		go func(n int) {
+			defer wg.Done()
+			_ = s.AppendMsg(7, 1000+n)
+			_ = s.SetBlocked(7, n%2 == 0)
+			_, _ = s.Get(7)
+			_, _ = s.ByTopic(100)
+		}(i)
+	}
+	wg.Wait()
+	got, ok := s.Get(7)
+	if !ok || len(got.RelayedMsgIDs) != 20 {
+		t.Errorf("after concurrent appends: %d ids, want 20", len(got.RelayedMsgIDs))
+	}
+}