scrabble-game/gateway/internal/botlink/hub.go

// Package botlink is the gateway side of the reverse Telegram bot channel: a remote
// bot dials the gateway and opens one long-lived mTLS gRPC stream
// (pkg/proto/botlink/v1), over which the gateway pushes send Commands and the bot
// returns an Ack per command. The Hub registers connected bots and routes commands:
// out-of-app push is fire-and-forget (Send), admin sends await the bot Ack
// (SendAwait). Delivery is best-effort, at-most-once — a command lost across a
// reconnect is not replayed. See docs/ARCHITECTURE.md.
package botlink

import (
	"context"
	"errors"
	"strconv"
	"sync"
	"sync/atomic"

	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/metric"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"

	botlinkv1 "scrabble/pkg/proto/botlink/v1"
)

// ErrNoBot is returned by SendAwait when no bot is currently connected.
var ErrNoBot = errors.New("botlink: no bot connected")

// outboundBuffer is the per-link command queue depth; a full queue drops commands
// (at-most-once under backpressure).
const outboundBuffer = 64

// Hub registers connected bots and routes send commands to them. A single bot is
// expected today; the registry already holds a set so adding more later needs no
// rewrite.
type Hub struct {
	botlinkv1.UnimplementedBotLinkServer

	log *zap.Logger

	mu      sync.Mutex
	links   map[*link]struct{}
	pending map[string]chan *botlinkv1.Ack

	seq atomic.Uint64

	connected metric.Int64UpDownCounter
	commands  metric.Int64Counter
}

// link is one connected bot's outbound queue and identity.
type link struct {
	instanceID  string
	ownsUpdates bool
	out         chan *botlinkv1.ToBot
}

// NewHub builds a Hub. A nil meter disables metrics; a nil logger is tolerated.
func NewHub(log *zap.Logger, meter metric.Meter) *Hub {
	if log == nil {
		log = zap.NewNop()
	}
	h := &Hub{
		log:     log,
		links:   make(map[*link]struct{}),
		pending: make(map[string]chan *botlinkv1.Ack),
	}
	if meter != nil {
		h.connected, _ = meter.Int64UpDownCounter("botlink_connected_bots",
			metric.WithDescription("Number of Telegram bots currently connected to the gateway bot-link."))
		h.commands, _ = meter.Int64Counter("botlink_commands_total",
			metric.WithDescription("Bot-link send commands by result (delivered, not_delivered, dropped, error)."))
	}
	return h
}

// Link implements botlinkv1.BotLinkServer: it registers the dialing bot, drains
// queued commands to it, and resolves Acks until the stream ends.
func (h *Hub) Link(stream grpc.BidiStreamingServer[botlinkv1.FromBot, botlinkv1.ToBot]) error {
	first, err := stream.Recv()
	if err != nil {
		return err
	}
	hello := first.GetHello()
	if hello == nil {
		return status.Error(codes.InvalidArgument, "first bot-link message must be Hello")
	}
	l := &link{
		instanceID:  hello.GetInstanceId(),
		ownsUpdates: hello.GetOwnsUpdates(),
		out:         make(chan *botlinkv1.ToBot, outboundBuffer),
	}
	h.register(l)
	defer h.unregister(l)

	ctx := stream.Context()
	// A dedicated goroutine owns stream.Send; the Recv loop below owns stream.Recv.
	go func() {
		for {
			select {
			case <-ctx.Done():
				return
			case msg := <-l.out:
				if err := stream.Send(msg); err != nil {
					return
				}
			}
		}
	}()

	for {
		msg, err := stream.Recv()
		if err != nil {
			return err
		}
		if ack := msg.GetAck(); ack != nil {
			h.resolve(ack)
		}
	}
}

// register adds a connected bot.
func (h *Hub) register(l *link) {
	h.mu.Lock()
	h.links[l] = struct{}{}
	n := len(h.links)
	h.mu.Unlock()
	if h.connected != nil {
		h.connected.Add(context.Background(), 1)
	}
	h.log.Info("bot connected",
		zap.String("instance_id", l.instanceID),
		zap.Bool("owns_updates", l.ownsUpdates),
		zap.Int("connected", n))
}

// unregister removes a bot. l.out is left for the GC; its sender goroutine has
// already exited via the stream context, and stale enqueues simply never send
// (at-most-once).
func (h *Hub) unregister(l *link) {
	h.mu.Lock()
	delete(h.links, l)
	n := len(h.links)
	h.mu.Unlock()
	if h.connected != nil {
		h.connected.Add(context.Background(), -1)
	}
	h.log.Info("bot disconnected", zap.String("instance_id", l.instanceID), zap.Int("connected", n))
}

// pick returns one connected bot.
func (h *Hub) pick() (*link, bool) {
	h.mu.Lock()
	defer h.mu.Unlock()
	for l := range h.links {
		return l, true
	}
	return nil, false
}

// Send enqueues a fire-and-forget command to a connected bot, dropping it (with a
// warning) when no bot is connected or the queue is full.
func (h *Hub) Send(cmd *botlinkv1.Command) {
	l, ok := h.pick()
	if !ok {
		h.count("dropped")
		h.log.Warn("bot-link send dropped: no bot connected")
		return
	}
	cmd.CommandId = h.nextID()
	select {
	case l.out <- &botlinkv1.ToBot{Command: cmd}:
	default:
		h.count("dropped")
		h.log.Warn("bot-link send dropped: outbound queue full", zap.String("instance_id", l.instanceID))
	}
}

// SendAwait enqueues a command and waits for the bot's Ack or until ctx is done.
// It returns ErrNoBot when no bot is connected, the delivered flag on a clean Ack,
// or (false, nil) when ctx (the deadline) fires before an Ack arrives.
func (h *Hub) SendAwait(ctx context.Context, cmd *botlinkv1.Command) (bool, error) {
	l, ok := h.pick()
	if !ok {
		h.count("dropped")
		return false, ErrNoBot
	}
	id := h.nextID()
	cmd.CommandId = id
	ackc := make(chan *botlinkv1.Ack, 1)
	h.mu.Lock()
	h.pending[id] = ackc
	h.mu.Unlock()
	defer func() {
		h.mu.Lock()
		delete(h.pending, id)
		h.mu.Unlock()
	}()

	select {
	case l.out <- &botlinkv1.ToBot{Command: cmd}:
	case <-ctx.Done():
		h.count("dropped")
		return false, ctx.Err()
	}

	select {
	case ack := <-ackc:
		if e := ack.GetError(); e != "" {
			h.count("error")
			return false, errors.New(e)
		}
		h.count(deliveredLabel(ack.GetDelivered()))
		return ack.GetDelivered(), nil
	case <-ctx.Done():
		h.count("error")
		return false, nil
	}
}

// resolve routes an Ack to its waiting SendAwait, if any.
func (h *Hub) resolve(ack *botlinkv1.Ack) {
	h.mu.Lock()
	c, ok := h.pending[ack.GetCommandId()]
	h.mu.Unlock()
	if ok {
		select {
		case c <- ack:
		default:
		}
	}
}

// nextID returns a process-unique command id.
func (h *Hub) nextID() string {
	return strconv.FormatUint(h.seq.Add(1), 10)
}

// count records one command result, when metrics are enabled.
func (h *Hub) count(result string) {
	if h.commands == nil {
		return
	}
	h.commands.Add(context.Background(), 1, metric.WithAttributes(resultAttr(result)))
}

// deliveredLabel maps the delivered flag to a metric result label.
func deliveredLabel(delivered bool) string {
	if delivered {
		return "delivered"
	}
	return "not_delivered"
}

// resultAttr is the metric attribute carrying a command result label.
func resultAttr(result string) attribute.KeyValue {
	return attribute.String("result", result)
}