feat(bot): report Telegram bot Bot API health to the gateway over the bot-link

The bot runs on its own host and exports no telemetry (otelcol is unreachable
from there), so it was a monitoring blind spot. Observe its Bot API health
centrally by wrapping the HTTP client — one place, no per-call-site
instrumentation — and relay it up the existing bot-link as a periodic Health
message the gateway turns into its own metrics.

- proto: add Health (delta connect / api / 429 counters + a last-ok stamp) to
  the FromBot oneof (additive, backward-compatible).
- bot: platform/telegram/internal/health wraps the Bot API HTTP client — it
  stamps liveness on any 2xx (so the getUpdates long-poll keeps it fresh even
  when idle), classifies transport/5xx failures (getUpdates vs other) and 429s,
  and honours a 429's Retry-After (bounded) so the bot backs off; a 4xx other
  than 429 is a normal per-request outcome and is not counted.
- bot-link client: flush the reporter as a Health message every 30s over a
  single-sender loop (a gRPC stream forbids concurrent Send).
- gateway: fold each report into bot_tg_errors_total{kind} and the
  bot_tg_last_ok_unix liveness gauge.
- grafana: alerts (bot disconnected, bot not reaching the Bot API, sustained
  429s) routed to the operator email — which does not go through the bot — plus
  a Telegram-bot dashboard.
- docs (ARCHITECTURE, compose comments); unit tests (observer classification,
  Retry-After, snapshot/commit, hub last-ok monotonicity).

platform/telegram/internal/bot/bot.go

@@ -7,19 +7,26 @@ package bot
 
 import (
 	"context"
+	"net/http"
 	"net/url"
 	"strconv"
 	"strings"
+	"time"
 
 	tgbot "github.com/go-telegram/bot"
 	"github.com/go-telegram/bot/models"
 	"go.uber.org/zap"
 	"golang.org/x/time/rate"
 
+	"scrabble/platform/telegram/internal/health"
 	"scrabble/platform/telegram/internal/outbox"
 	"scrabble/platform/telegram/internal/support"
 )
 
+// botPollTimeout is the getUpdates long-poll timeout. It matches the go-telegram/bot default; we set
+// it explicitly only because wiring the health observer (WithHTTPClient) also sets the poll timeout.
+const botPollTimeout = time.Minute
+
 // Config configures the bot wrapper.
 type Config struct {
 	// Token is the Bot API token.
@@ -54,6 +61,9 @@ type Config struct {
 	// pre_checkout_query updates and handles pre_checkout / successful_payment. The runtime
 	// dependencies (the validator, forwarder and outbox) are wired with SetPaymentHandlers.
 	AcceptPayments bool
+	// Health, when set, observes every Bot API request for health metrics (reported to the gateway
+	// over the bot-link) and honours a 429's Retry-After. Nil disables the observation.
+	Health *health.Reporter
 }
 
 // EligibilityResolver answers whether the Telegram user identified by externalID
@@ -159,6 +169,12 @@ func New(cfg Config, log *zap.Logger) (*Bot, error) {
 	if cfg.APIBaseURL != "" {
 		opts = append(opts, tgbot.WithServerURL(cfg.APIBaseURL))
 	}
+	if cfg.Health != nil {
+		// Observe every Bot API request centrally for health metrics and the 429 back-off. The
+		// client timeout leaves slack over the long-poll hold (botPollTimeout - 1s server-side).
+		base := &http.Client{Timeout: botPollTimeout + 10*time.Second}
+		opts = append(opts, tgbot.WithHTTPClient(botPollTimeout, cfg.Health.Wrap(base)))
+	}
 	api, err := tgbot.New(cfg.Token, opts...)
 	if err != nil {
 		return nil, err

platform/telegram/internal/botlink/client.go

@@ -11,6 +11,7 @@ import (
 	"google.golang.org/grpc/keepalive"
 
 	botlinkv1 "scrabble/pkg/proto/botlink/v1"
+	"scrabble/platform/telegram/internal/health"
 )
 
 const (
@@ -34,6 +35,11 @@ type ClientConfig struct {
 	Creds credentials.TransportCredentials
 	// ReconnectDelay is the pause before re-dialing after the stream ends.
 	ReconnectDelay time.Duration
+	// Health, when set with a positive HealthInterval, is flushed to the gateway as a botlink Health
+	// message every HealthInterval while the stream is open. Nil disables health reporting.
+	Health *health.Reporter
+	// HealthInterval is the cadence of the Health flush; 0 disables it.
+	HealthInterval time.Duration
 }
 
 // Client maintains the long-lived bot-link to the gateway, executing the commands
@@ -140,22 +146,59 @@ func (c *Client) serve(ctx context.Context, client botlinkv1.BotLinkClient) erro
 	}
 	c.log.Info("bot-link connected", zap.String("gateway", c.cfg.GatewayAddr), zap.Bool("owns_updates", c.cfg.OwnsUpdates))
 
+	// One goroutine owns stream.Recv, feeding commands to the loop below; the loop owns every
+	// stream.Send (the Acks and the periodic Health) — a gRPC stream forbids concurrent Send.
+	rctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	cmds := make(chan *botlinkv1.Command)
+	recvErr := make(chan error, 1)
+	go func() {
+		for {
+			msg, err := stream.Recv()
+			if err != nil {
+				recvErr <- err
+				return
+			}
+			if cmd := msg.GetCommand(); cmd != nil {
+				select {
+				case cmds <- cmd:
+				case <-rctx.Done():
+					return
+				}
+			}
+		}
+	}()
+
+	var healthTick <-chan time.Time
+	if c.cfg.Health != nil && c.cfg.HealthInterval > 0 {
+		t := time.NewTicker(c.cfg.HealthInterval)
+		defer t.Stop()
+		healthTick = t.C
+	}
+
 	for {
-		msg, err := stream.Recv()
-		if err != nil {
-			return err
-		}
-		cmd := msg.GetCommand()
-		if cmd == nil {
-			continue
-		}
-		delivered, result, herr := c.exec.Handle(ctx, cmd)
-		ack := &botlinkv1.Ack{CommandId: cmd.GetCommandId(), Delivered: delivered, Result: result}
-		if herr != nil {
-			ack.Error = herr.Error()
-		}
-		if err := stream.Send(&botlinkv1.FromBot{Msg: &botlinkv1.FromBot_Ack{Ack: ack}}); err != nil {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case err := <-recvErr:
 			return err
+		case cmd := <-cmds:
+			delivered, result, herr := c.exec.Handle(ctx, cmd)
+			ack := &botlinkv1.Ack{CommandId: cmd.GetCommandId(), Delivered: delivered, Result: result}
+			if herr != nil {
+				ack.Error = herr.Error()
+			}
+			if err := stream.Send(&botlinkv1.FromBot{Msg: &botlinkv1.FromBot_Ack{Ack: ack}}); err != nil {
+				return err
+			}
+		case <-healthTick:
+			// Snapshot without resetting; only commit (clear the deltas) after a successful send, so a
+			// failed flush loses nothing and the counts ride the next connection.
+			hb := c.cfg.Health.Snapshot()
+			if err := stream.Send(&botlinkv1.FromBot{Msg: &botlinkv1.FromBot_Health{Health: hb}}); err != nil {
+				return err
+			}
+			c.cfg.Health.Commit(hb)
 		}
 	}
 }

platform/telegram/internal/health/health.go

@@ -0,0 +1,145 @@
+// Package health tracks the Telegram bot's Bot API health and reports it to the gateway over the
+// bot-link. The bot exports no telemetry of its own — otelcol lives on the main host, unreachable
+// from the bot host — so the gateway turns these reports into its own metrics (see
+// gateway/internal/botlink). Health is observed CENTRALLY by wrapping the Bot API HTTP client
+// ([Reporter.Wrap]); every request is classified there with no per-call-site instrumentation.
+package health
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	botlinkv1 "scrabble/pkg/proto/botlink/v1"
+)
+
+// maxBackoff caps how long a single 429 makes the bot wait, so a hostile or buggy Retry-After can
+// never wedge the bot.
+const maxBackoff = 30 * time.Second
+
+// Reporter accumulates the bot's Bot API health between reports: three delta counters and the last
+// successful-response stamp. It is safe for concurrent use. The bot-link client periodically
+// [Reporter.Snapshot]s it, sends the snapshot as a botlink Health, and [Reporter.Commit]s it on a
+// successful send.
+type Reporter struct {
+	connectFailures atomic.Int64
+	apiErrors       atomic.Int64
+	rateLimited     atomic.Int64
+	lastOKUnix      atomic.Int64
+}
+
+// NewReporter returns an empty Reporter.
+func NewReporter() *Reporter { return &Reporter{} }
+
+// Snapshot reads the current delta counters and the last-ok stamp into a Health message WITHOUT
+// resetting them. The counters are cleared only by [Reporter.Commit] after a successful send, so a
+// failed send loses nothing.
+func (r *Reporter) Snapshot() *botlinkv1.Health {
+	return &botlinkv1.Health{
+		ConnectFailures: uint64(max(r.connectFailures.Load(), 0)),
+		ApiErrors:       uint64(max(r.apiErrors.Load(), 0)),
+		RateLimited:     uint64(max(r.rateLimited.Load(), 0)),
+		LastOkUnix:      r.lastOKUnix.Load(),
+	}
+}
+
+// Commit subtracts a just-sent snapshot's delta counters, so counts accrued between the snapshot and
+// the send survive into the next report. last_ok is a stamp, not a delta, so it is left as is.
+func (r *Reporter) Commit(sent *botlinkv1.Health) {
+	r.connectFailures.Add(-int64(sent.GetConnectFailures()))
+	r.apiErrors.Add(-int64(sent.GetApiErrors()))
+	r.rateLimited.Add(-int64(sent.GetRateLimited()))
+}
+
+// Wrap returns an http.Client-shaped observer over base for tgbot.WithHTTPClient: it stamps liveness
+// on every successful response, counts transport and 5xx failures (split getUpdates vs other) and
+// 429s, and honours a 429's Retry-After (bounded by maxBackoff) so the bot backs off instead of
+// hammering. A 4xx other than 429 (a user blocked the bot, a malformed request) is a normal
+// per-request outcome, not a health signal, and is not counted.
+func (r *Reporter) Wrap(base Doer) Doer { return &observer{base: base, r: r} }
+
+// Doer is the minimal HTTP surface the Bot API client needs; both *http.Client and [observer]
+// satisfy it, and it matches the go-telegram/bot HttpClient interface.
+type Doer interface {
+	Do(*http.Request) (*http.Response, error)
+}
+
+// observer is the Bot API HTTP client wrapper (see [Reporter.Wrap]).
+type observer struct {
+	base Doer
+	r    *Reporter
+}
+
+// Do performs the request and records its health outcome.
+func (o *observer) Do(req *http.Request) (*http.Response, error) {
+	resp, err := o.base.Do(req)
+	poll := strings.HasSuffix(req.URL.Path, "/getUpdates")
+	if err != nil {
+		// A cancelled context is a shutdown, not a Bot API failure.
+		if req.Context().Err() == nil {
+			o.fail(poll)
+		}
+		return resp, err
+	}
+	switch {
+	case resp.StatusCode == http.StatusTooManyRequests:
+		o.r.rateLimited.Add(1)
+		o.backoff(req.Context(), resp)
+	case resp.StatusCode >= 200 && resp.StatusCode < 300:
+		o.r.lastOKUnix.Store(time.Now().Unix())
+	case resp.StatusCode >= 500:
+		o.fail(poll)
+	}
+	return resp, err
+}
+
+// fail records a connect failure on the getUpdates long-poll or an api error otherwise.
+func (o *observer) fail(poll bool) {
+	if poll {
+		o.r.connectFailures.Add(1)
+	} else {
+		o.r.apiErrors.Add(1)
+	}
+}
+
+// backoff waits out a 429's Retry-After (bounded, cancellable). It buffers the response body first so
+// the wait holds no connection open and the Bot API library can still parse the 429 afterwards.
+func (o *observer) backoff(ctx context.Context, resp *http.Response) {
+	body, _ := io.ReadAll(resp.Body)
+	resp.Body.Close()
+	resp.Body = io.NopCloser(bytes.NewReader(body))
+	d := retryAfter(resp.Header.Get("Retry-After"), body)
+	if d <= 0 {
+		return
+	}
+	if d > maxBackoff {
+		d = maxBackoff
+	}
+	select {
+	case <-ctx.Done():
+	case <-time.After(d):
+	}
+}
+
+// retryAfter resolves the 429 back-off from the Retry-After header, falling back to the Bot API
+// body's parameters.retry_after (both are seconds). It returns 0 when neither gives a positive value.
+func retryAfter(header string, body []byte) time.Duration {
+	if secs, err := strconv.Atoi(strings.TrimSpace(header)); err == nil && secs > 0 {
+		return time.Duration(secs) * time.Second
+	}
+	var payload struct {
+		Parameters struct {
+			RetryAfter int `json:"retry_after"`
+		} `json:"parameters"`
+	}
+	if err := json.Unmarshal(body, &payload); err == nil && payload.Parameters.RetryAfter > 0 {
+		return time.Duration(payload.Parameters.RetryAfter) * time.Second
+	}
+	return 0
+}

platform/telegram/internal/health/health_test.go

@@ -0,0 +1,121 @@
+package health
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+// fakeDoer returns a canned response and error for every request.
+type fakeDoer struct {
+	resp *http.Response
+	err  error
+}
+
+func (f *fakeDoer) Do(*http.Request) (*http.Response, error) { return f.resp, f.err }
+
+// mkResp builds a response with a status, optional Retry-After header and body.
+func mkResp(status int, retryAfter, body string) *http.Response {
+	h := http.Header{}
+	if retryAfter != "" {
+		h.Set("Retry-After", retryAfter)
+	}
+	return &http.Response{StatusCode: status, Header: h, Body: io.NopCloser(strings.NewReader(body))}
+}
+
+// run sends one request for the Bot API method through a fresh observer and returns the reporter's
+// resulting snapshot. The 429 cases carry no Retry-After, so the back-off returns before any wait.
+func run(method string, resp *http.Response, err error) *Reporter {
+	r := NewReporter()
+	req := httptest.NewRequest(http.MethodPost, "https://api.telegram.org/bot123/"+method, nil)
+	_, _ = r.Wrap(&fakeDoer{resp: resp, err: err}).Do(req)
+	return r
+}
+
+func TestObserverClassifies(t *testing.T) {
+	cases := []struct {
+		name          string
+		method        string
+		resp          *http.Response
+		err           error
+		wantConnect   uint64
+		wantAPI       uint64
+		wantRate      uint64
+		wantLastOKSet bool
+	}{
+		{"poll ok stamps liveness", "getUpdates", mkResp(200, "", `{"ok":true}`), nil, 0, 0, 0, true},
+		{"send ok stamps liveness", "sendMessage", mkResp(200, "", `{"ok":true}`), nil, 0, 0, 0, true},
+		{"poll 5xx is a connect failure", "getUpdates", mkResp(502, "", ""), nil, 1, 0, 0, false},
+		{"send 5xx is an api error", "sendMessage", mkResp(500, "", ""), nil, 0, 1, 0, false},
+		{"429 is rate limited, not an error", "sendMessage", mkResp(429, "", `{"ok":false}`), nil, 0, 0, 1, false},
+		{"4xx other than 429 is not counted", "sendMessage", mkResp(403, "", ""), nil, 0, 0, 0, false},
+		{"poll transport error is a connect failure", "getUpdates", nil, errors.New("dial"), 1, 0, 0, false},
+		{"send transport error is an api error", "sendMessage", nil, errors.New("dial"), 0, 1, 0, false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			s := run(tc.method, tc.resp, tc.err).Snapshot()
+			if s.GetConnectFailures() != tc.wantConnect || s.GetApiErrors() != tc.wantAPI || s.GetRateLimited() != tc.wantRate {
+				t.Errorf("counters = connect %d api %d rate %d, want %d/%d/%d",
+					s.GetConnectFailures(), s.GetApiErrors(), s.GetRateLimited(), tc.wantConnect, tc.wantAPI, tc.wantRate)
+			}
+			if (s.GetLastOkUnix() > 0) != tc.wantLastOKSet {
+				t.Errorf("last_ok set = %v, want %v", s.GetLastOkUnix() > 0, tc.wantLastOKSet)
+			}
+		})
+	}
+}
+
+// TestObserverIgnoresShutdownTransportError checks a transport error under a cancelled context (a
+// shutdown) is not counted as a Bot API failure.
+func TestObserverIgnoresShutdownTransportError(t *testing.T) {
+	r := NewReporter()
+	req := httptest.NewRequest(http.MethodPost, "https://api.telegram.org/bot123/getUpdates", nil)
+	ctx, cancel := context.WithCancel(req.Context())
+	cancel()
+	_, _ = r.Wrap(&fakeDoer{err: errors.New("dial")}).Do(req.WithContext(ctx))
+	if s := r.Snapshot(); s.GetConnectFailures() != 0 {
+		t.Errorf("shutdown transport error must not count, got connect=%d", s.GetConnectFailures())
+	}
+}
+
+func TestRetryAfter(t *testing.T) {
+	cases := []struct {
+		header string
+		body   string
+		want   time.Duration
+	}{
+		{"5", "", 5 * time.Second},
+		{"", `{"parameters":{"retry_after":7}}`, 7 * time.Second},
+		{"3", `{"parameters":{"retry_after":9}}`, 3 * time.Second}, // header wins
+		{"", `{"ok":false}`, 0},
+		{"0", "", 0},
+		{"", "not json", 0},
+	}
+	for _, tc := range cases {
+		if got := retryAfter(tc.header, []byte(tc.body)); got != tc.want {
+			t.Errorf("retryAfter(%q, %q) = %v, want %v", tc.header, tc.body, got, tc.want)
+		}
+	}
+}
+
+// TestSnapshotCommit checks Commit clears only the sent deltas, so counts accrued between the
+// snapshot and the send survive into the next report.
+func TestSnapshotCommit(t *testing.T) {
+	r := NewReporter()
+	r.apiErrors.Add(3)
+	snap := r.Snapshot()
+	if snap.GetApiErrors() != 3 {
+		t.Fatalf("snapshot api_errors = %d, want 3", snap.GetApiErrors())
+	}
+	r.apiErrors.Add(2) // accrues after the snapshot
+	r.Commit(snap)
+	if got := r.Snapshot().GetApiErrors(); got != 2 {
+		t.Errorf("after commit api_errors = %d, want 2 (the post-snapshot accrual)", got)
+	}
+}