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"
	"time"

	"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

// invoiceMintTimeout bounds one synchronous invoice-mint round-trip (the gateway commands the bot
// and awaits the Ack carrying the createInvoiceLink result), so a hung or slow bot cannot stall the
// caller's wallet-order request indefinitely.
const invoiceMintTimeout = 15 * time.Second

// EligibilityResolver answers a Telegram identity's moderated-chat write eligibility
// for the bot's join-time ResolveChatEligibility query: registered reports whether the
// identity maps to an account, eligible is the final gate the bot acts on (registered
// and neither admin-suspended nor chat-muted). The gateway backs it with the backend
// chat-access endpoint.
type EligibilityResolver func(ctx context.Context, externalID string) (registered, eligible bool, err error)

// PreCheckoutResolver validates a Telegram Stars pre_checkout_query for the bot's
// ValidatePreCheckout query: it answers whether the order may still be charged (ok) and, when not,
// a short reason to show the payer. The gateway backs it with the backend intake.
type PreCheckoutResolver func(ctx context.Context, orderID string, amount int64, currency string) (ok bool, reason string, err error)

// PaymentForwarder delivers a completed Telegram Stars payment for the bot's ForwardPayment call:
// it credits the order through the backend intake and reports whether it was credited (or already
// had been). A non-nil error is a transient failure the bot retries. The gateway backs it with the
// backend intake.
type PaymentForwarder func(ctx context.Context, orderID, chargeID string, amount, telegramUserID int64) (credited bool, err error)

// 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
	eligibility EligibilityResolver
	// precheck and forward back the Telegram Stars payment RPCs; nil until SetPaymentBridge wires
	// them, in which case the RPCs report Unavailable. Set once before the gRPC server serves.
	precheck PreCheckoutResolver
	forward  PaymentForwarder

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

	seq atomic.Uint64

	connected metric.Int64UpDownCounter
	commands  metric.Int64Counter
	// tgErrors counts the remote bot's Bot API failures it reports over the stream, by kind
	// (connect / api / rate_limited). lastOK holds the wall-clock second of the bot's most recent
	// successful Bot API call, read by the bot_tg_last_ok_unix observable gauge for a staleness alert.
	tgErrors metric.Int64Counter
	lastOK   atomic.Int64
}

// 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. resolve answers the bot's join-time chat-eligibility query
// (nil rejects it as unavailable). A nil meter disables metrics; a nil logger is
// tolerated.
func NewHub(log *zap.Logger, meter metric.Meter, resolve EligibilityResolver) *Hub {
	if log == nil {
		log = zap.NewNop()
	}
	h := &Hub{
		log:         log,
		eligibility: resolve,
		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)."))
		h.tgErrors, _ = meter.Int64Counter("bot_tg_errors_total",
			metric.WithDescription("Telegram Bot API failures the remote bot reports over the bot-link, by kind (connect, api, rate_limited)."))
		// The bot's last successful Bot API second, reported over the stream. An observable gauge
		// reads the atomic on each collection; it observes nothing until a bot has reported, so a
		// never-connected gateway shows no data (the connected-bots alert covers that case).
		_, _ = meter.Int64ObservableGauge("bot_tg_last_ok_unix",
			metric.WithDescription("Unix second of the remote bot's most recent successful Bot API call (0/absent until reported)."),
			metric.WithInt64Callback(func(_ context.Context, o metric.Int64Observer) error {
				if v := h.lastOK.Load(); v > 0 {
					o.Observe(v)
				}
				return nil
			}))
	}
	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)
		}
		if hb := msg.GetHealth(); hb != nil {
			h.recordHealth(hb)
		}
	}
}

// recordHealth folds one bot Health report into the gateway's cumulative Bot API metrics: the three
// delta counters are added under their kind label, and the last-ok stamp (an absolute wall-clock
// second) advances the liveness gauge (monotonically — a stale reordered report cannot rewind it).
func (h *Hub) recordHealth(hb *botlinkv1.Health) {
	if h.tgErrors != nil {
		ctx := context.Background()
		if v := hb.GetConnectFailures(); v > 0 {
			h.tgErrors.Add(ctx, int64(v), metric.WithAttributes(attribute.String("kind", "connect")))
		}
		if v := hb.GetApiErrors(); v > 0 {
			h.tgErrors.Add(ctx, int64(v), metric.WithAttributes(attribute.String("kind", "api")))
		}
		if v := hb.GetRateLimited(); v > 0 {
			h.tgErrors.Add(ctx, int64(v), metric.WithAttributes(attribute.String("kind", "rate_limited")))
		}
	}
	for {
		cur := h.lastOK.Load()
		if hb.GetLastOkUnix() <= cur {
			break
		}
		if h.lastOK.CompareAndSwap(cur, hb.GetLastOkUnix()) {
			break
		}
	}
}

// ResolveChatEligibility serves the bot's join-time query: whether the Telegram user
// identified in the request may write in the moderated discussion chat. It delegates
// to the configured resolver (the backend chat-access endpoint), unlike the streamed
// Commands it is a plain request/response over the same mTLS channel.
func (h *Hub) ResolveChatEligibility(ctx context.Context, req *botlinkv1.ChatEligibilityRequest) (*botlinkv1.ChatEligibilityResponse, error) {
	if h.eligibility == nil {
		return nil, status.Error(codes.Unavailable, "chat eligibility resolver not configured")
	}
	registered, eligible, err := h.eligibility(ctx, req.GetExternalId())
	if err != nil {
		h.log.Warn("resolve chat eligibility failed", zap.String("external_id", req.GetExternalId()), zap.Error(err))
		return nil, status.Error(codes.Internal, "resolve chat eligibility")
	}
	return &botlinkv1.ChatEligibilityResponse{Registered: registered, Eligible: eligible}, nil
}

// SetPaymentBridge wires the Telegram Stars payment resolvers, backing ValidatePreCheckout and
// ForwardPayment. It must be called before the gRPC server starts serving (the fields are read
// without a lock, relying on that happens-before). A nil resolver leaves its RPC reporting
// Unavailable.
func (h *Hub) SetPaymentBridge(precheck PreCheckoutResolver, forward PaymentForwarder) {
	h.precheck = precheck
	h.forward = forward
}

// ValidatePreCheckout serves the bot's pre_checkout validation over the same mTLS channel: it
// delegates to the configured resolver (the backend intake) and returns whether the order may be
// charged. A resolver failure maps to Internal, which the bot treats as a decline (fail-closed).
func (h *Hub) ValidatePreCheckout(ctx context.Context, req *botlinkv1.PreCheckoutRequest) (*botlinkv1.PreCheckoutResponse, error) {
	if h.precheck == nil {
		return nil, status.Error(codes.Unavailable, "payment bridge not configured")
	}
	ok, reason, err := h.precheck(ctx, req.GetOrderId(), req.GetAmount(), req.GetCurrency())
	if err != nil {
		h.log.Warn("validate pre_checkout failed", zap.String("order", req.GetOrderId()), zap.Error(err))
		return nil, status.Error(codes.Internal, "validate pre_checkout")
	}
	return &botlinkv1.PreCheckoutResponse{Ok: ok, Reason: reason}, nil
}

// ForwardPayment serves the bot's completed-payment delivery: it credits the order through the
// configured forwarder (the backend intake) and reports the durable outcome. A forwarder failure
// maps to Internal, which the bot treats as transient and retries; a clean response (credited true
// or false) lets the bot forget the outbox row.
func (h *Hub) ForwardPayment(ctx context.Context, req *botlinkv1.ForwardPaymentRequest) (*botlinkv1.ForwardPaymentResponse, error) {
	if h.forward == nil {
		return nil, status.Error(codes.Unavailable, "payment bridge not configured")
	}
	credited, err := h.forward(ctx, req.GetOrderId(), req.GetTelegramPaymentChargeId(), req.GetAmount(), req.GetTelegramUserId())
	if err != nil {
		h.log.Warn("forward telegram payment failed", zap.String("order", req.GetOrderId()), zap.Error(err))
		return nil, status.Error(codes.Internal, "forward payment")
	}
	return &botlinkv1.ForwardPaymentResponse{Credited: credited}, nil
}

// MintInvoice commands the connected bot to mint a Telegram Stars invoice link for the order and
// returns the link. It is a bounded, synchronous round-trip (invoiceMintTimeout): no bot connected
// returns ErrNoBot, and a bot error or an empty link is an error the caller surfaces as a failed
// order launch.
func (h *Hub) MintInvoice(ctx context.Context, title, description, orderID string, amountStars int64) (string, error) {
	cctx, cancel := context.WithTimeout(ctx, invoiceMintTimeout)
	defer cancel()
	ack, err := h.sendAwaitAck(cctx, CreateInvoiceCommand(title, description, orderID, amountStars))
	if err != nil {
		return "", err
	}
	if ack.GetResult() == "" {
		return "", errors.New("botlink: bot returned an empty invoice link")
	}
	return ack.GetResult(), nil
}

// sendAwaitAck enqueues a command and waits for the bot's full Ack (or ctx). It mirrors SendAwait
// but returns the Ack — the invoice-mint path needs its result field — and reports ctx expiry as an
// error rather than a not-delivered, since a timed-out mint has no link to return.
func (h *Hub) sendAwaitAck(ctx context.Context, cmd *botlinkv1.Command) (*botlinkv1.Ack, error) {
	l, ok := h.pick()
	if !ok {
		h.count("dropped")
		return nil, 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 nil, ctx.Err()
	}

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

// 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)
}