bb71e7b1c7
CI / changes (pull_request) Successful in 2s
CI / unit (pull_request) Successful in 12s
CI / integration (pull_request) Successful in 26s
CI / ui (pull_request) Successful in 1m11s
CI / conformance (pull_request) Successful in 10s
CI / gate (pull_request) Successful in 0s
CI / deploy (pull_request) Successful in 1m56s
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).
254 lines
7.3 KiB
Go
254 lines
7.3 KiB
Go
package botlink
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"net"
|
|
"testing"
|
|
"time"
|
|
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/credentials/insecure"
|
|
"google.golang.org/grpc/status"
|
|
"google.golang.org/grpc/test/bufconn"
|
|
|
|
botlinkv1 "scrabble/pkg/proto/botlink/v1"
|
|
telegramv1 "scrabble/pkg/proto/telegram/v1"
|
|
)
|
|
|
|
// fakeBot is a test bot that dials the hub, registers, and acks commands with a
|
|
// fixed delivered flag (or never, when ack is false).
|
|
type fakeBot struct {
|
|
delivered bool
|
|
ack bool
|
|
received chan *botlinkv1.Command
|
|
}
|
|
|
|
// startHub registers a Hub (no chat-eligibility resolver) on an in-memory gRPC
|
|
// server and returns the hub plus a dialer for fake bots.
|
|
func startHub(t *testing.T) (*Hub, func(t *testing.T) botlinkv1.BotLinkClient) {
|
|
return startHubWith(t, nil)
|
|
}
|
|
|
|
// startHubWith is startHub with an explicit chat-eligibility resolver, for the
|
|
// ResolveChatEligibility tests.
|
|
func startHubWith(t *testing.T, resolve EligibilityResolver) (*Hub, func(t *testing.T) botlinkv1.BotLinkClient) {
|
|
t.Helper()
|
|
lis := bufconn.Listen(1 << 20)
|
|
hub := NewHub(nil, nil, resolve)
|
|
srv := grpc.NewServer()
|
|
botlinkv1.RegisterBotLinkServer(srv, hub)
|
|
go func() { _ = srv.Serve(lis) }()
|
|
t.Cleanup(srv.Stop)
|
|
|
|
dial := func(t *testing.T) botlinkv1.BotLinkClient {
|
|
t.Helper()
|
|
conn, err := grpc.NewClient("passthrough:///bufnet",
|
|
grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return lis.Dial() }),
|
|
grpc.WithTransportCredentials(insecure.NewCredentials()),
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("dial: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = conn.Close() })
|
|
return botlinkv1.NewBotLinkClient(conn)
|
|
}
|
|
return hub, dial
|
|
}
|
|
|
|
// connect runs a fake bot against client until ctx is cancelled, returning once the
|
|
// hub has registered it.
|
|
func (f *fakeBot) connect(t *testing.T, ctx context.Context, hub *Hub, client botlinkv1.BotLinkClient) {
|
|
t.Helper()
|
|
f.received = make(chan *botlinkv1.Command, 8)
|
|
stream, err := client.Link(ctx)
|
|
if err != nil {
|
|
t.Fatalf("link: %v", err)
|
|
}
|
|
if err := stream.Send(&botlinkv1.FromBot{Msg: &botlinkv1.FromBot_Hello{Hello: &botlinkv1.Hello{InstanceId: "test"}}}); err != nil {
|
|
t.Fatalf("hello: %v", err)
|
|
}
|
|
go func() {
|
|
for {
|
|
msg, err := stream.Recv()
|
|
if err != nil {
|
|
return
|
|
}
|
|
cmd := msg.GetCommand()
|
|
f.received <- cmd
|
|
if !f.ack {
|
|
continue
|
|
}
|
|
_ = stream.Send(&botlinkv1.FromBot{Msg: &botlinkv1.FromBot_Ack{Ack: &botlinkv1.Ack{
|
|
CommandId: cmd.GetCommandId(),
|
|
Delivered: f.delivered,
|
|
}}})
|
|
}
|
|
}()
|
|
waitConnected(t, hub, 1)
|
|
}
|
|
|
|
// waitConnected blocks until the hub reports n connected bots.
|
|
func waitConnected(t *testing.T, hub *Hub, n int) {
|
|
t.Helper()
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for time.Now().Before(deadline) {
|
|
hub.mu.Lock()
|
|
got := len(hub.links)
|
|
hub.mu.Unlock()
|
|
if got == n {
|
|
return
|
|
}
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
t.Fatalf("hub did not reach %d connected bots", n)
|
|
}
|
|
|
|
func TestHubSendAwaitDelivered(t *testing.T) {
|
|
hub, dial := startHub(t)
|
|
ctx := t.Context()
|
|
bot := &fakeBot{delivered: true, ack: true}
|
|
bot.connect(t, ctx, hub, dial(t))
|
|
|
|
delivered, err := hub.SendAwait(ctx, SendToUserCommand("42", "hi"))
|
|
if err != nil {
|
|
t.Fatalf("SendAwait: %v", err)
|
|
}
|
|
if !delivered {
|
|
t.Fatal("delivered = false, want true")
|
|
}
|
|
select {
|
|
case cmd := <-bot.received:
|
|
if cmd.GetSendToUser().GetExternalId() != "42" {
|
|
t.Errorf("received external_id = %q, want 42", cmd.GetSendToUser().GetExternalId())
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatal("bot received no command")
|
|
}
|
|
}
|
|
|
|
func TestHubSendAwaitNoBot(t *testing.T) {
|
|
hub, _ := startHub(t)
|
|
_, err := hub.SendAwait(context.Background(), SendToUserCommand("42", "hi"))
|
|
if !errors.Is(err, ErrNoBot) {
|
|
t.Errorf("err = %v, want ErrNoBot", err)
|
|
}
|
|
}
|
|
|
|
func TestHubSendAwaitDeadline(t *testing.T) {
|
|
hub, dial := startHub(t)
|
|
ctx := t.Context()
|
|
bot := &fakeBot{ack: false} // receives but never acks
|
|
bot.connect(t, ctx, hub, dial(t))
|
|
|
|
awaitCtx, awaitCancel := context.WithTimeout(ctx, 100*time.Millisecond)
|
|
defer awaitCancel()
|
|
delivered, err := hub.SendAwait(awaitCtx, SendToUserCommand("42", "hi"))
|
|
if err != nil {
|
|
t.Fatalf("SendAwait err = %v, want nil on deadline", err)
|
|
}
|
|
if delivered {
|
|
t.Error("delivered = true, want false on deadline")
|
|
}
|
|
}
|
|
|
|
func TestHubSendAsync(t *testing.T) {
|
|
hub, dial := startHub(t)
|
|
ctx := t.Context()
|
|
bot := &fakeBot{ack: false}
|
|
bot.connect(t, ctx, hub, dial(t))
|
|
|
|
hub.Send(NotifyCommand("42", "your_turn", nil, "en"))
|
|
select {
|
|
case cmd := <-bot.received:
|
|
if cmd.GetNotify().GetExternalId() != "42" {
|
|
t.Errorf("received external_id = %q, want 42", cmd.GetNotify().GetExternalId())
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatal("bot received no command")
|
|
}
|
|
}
|
|
|
|
func TestHubSendChatGate(t *testing.T) {
|
|
hub, dial := startHub(t)
|
|
ctx := t.Context()
|
|
bot := &fakeBot{ack: false}
|
|
bot.connect(t, ctx, hub, dial(t))
|
|
|
|
hub.Send(ChatGateCommand("42", true))
|
|
select {
|
|
case cmd := <-bot.received:
|
|
cg := cmd.GetChatGate()
|
|
if cg.GetExternalId() != "42" || !cg.GetAllow() {
|
|
t.Errorf("chat_gate = %+v, want external_id=42 allow=true", cg)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatal("bot received no command")
|
|
}
|
|
}
|
|
|
|
func TestHubResolveChatEligibility(t *testing.T) {
|
|
var gotExt string
|
|
_, dial := startHubWith(t, func(_ context.Context, ext string) (bool, bool, error) {
|
|
gotExt = ext
|
|
return true, ext == "good", nil
|
|
})
|
|
client := dial(t)
|
|
ctx := t.Context()
|
|
|
|
resp, err := client.ResolveChatEligibility(ctx, &botlinkv1.ChatEligibilityRequest{ExternalId: "good"})
|
|
if err != nil {
|
|
t.Fatalf("ResolveChatEligibility: %v", err)
|
|
}
|
|
if gotExt != "good" {
|
|
t.Errorf("resolver external_id = %q, want good", gotExt)
|
|
}
|
|
if !resp.GetRegistered() || !resp.GetEligible() {
|
|
t.Errorf("resp = %+v, want registered+eligible", resp)
|
|
}
|
|
|
|
resp, err = client.ResolveChatEligibility(ctx, &botlinkv1.ChatEligibilityRequest{ExternalId: "muted"})
|
|
if err != nil {
|
|
t.Fatalf("ResolveChatEligibility(muted): %v", err)
|
|
}
|
|
if !resp.GetRegistered() || resp.GetEligible() {
|
|
t.Errorf("resp = %+v, want registered but not eligible", resp)
|
|
}
|
|
}
|
|
|
|
func TestHubResolveChatEligibilityUnconfigured(t *testing.T) {
|
|
_, dial := startHub(t) // nil resolver
|
|
client := dial(t)
|
|
_, err := client.ResolveChatEligibility(t.Context(), &botlinkv1.ChatEligibilityRequest{ExternalId: "x"})
|
|
if status.Code(err) != codes.Unavailable {
|
|
t.Fatalf("err = %v, want Unavailable", err)
|
|
}
|
|
}
|
|
|
|
func TestRelayServerNoBot(t *testing.T) {
|
|
hub, _ := startHub(t)
|
|
relay := NewRelayServer(hub, 200*time.Millisecond)
|
|
if _, err := relay.SendToUser(context.Background(), &telegramv1.SendToUserRequest{ExternalId: "42", Text: "hi"}); err == nil {
|
|
t.Fatal("expected an error with no bot connected")
|
|
}
|
|
}
|
|
|
|
// TestRecordHealthLastOKMonotonic checks a bot Health report advances the last-ok liveness stamp but
|
|
// a stale or reordered report (an earlier second) never rewinds it.
|
|
func TestRecordHealthLastOKMonotonic(t *testing.T) {
|
|
hub := NewHub(nil, nil, nil)
|
|
hub.recordHealth(&botlinkv1.Health{LastOkUnix: 100})
|
|
if got := hub.lastOK.Load(); got != 100 {
|
|
t.Fatalf("lastOK = %d, want 100", got)
|
|
}
|
|
hub.recordHealth(&botlinkv1.Health{LastOkUnix: 90}) // reordered/stale — must not rewind
|
|
if got := hub.lastOK.Load(); got != 100 {
|
|
t.Errorf("lastOK rewound to %d, want 100", got)
|
|
}
|
|
hub.recordHealth(&botlinkv1.Health{LastOkUnix: 150})
|
|
if got := hub.lastOK.Load(); got != 150 {
|
|
t.Errorf("lastOK = %d, want 150", got)
|
|
}
|
|
}
|