scrabble-game/platform/telegram/internal/botlink/client.go

package botlink

import (
	"context"
	"time"

	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/keepalive"

	botlinkv1 "scrabble/pkg/proto/botlink/v1"
	"scrabble/platform/telegram/internal/health"
)

const (
	// clientKeepaliveTime is how often the bot pings the gateway to hold the WAN
	// connection open (kept above the gateway's MinTime to avoid an enforcement ban).
	clientKeepaliveTime = 30 * time.Second
	// clientKeepaliveTimeout bounds the wait for a keepalive ping reply.
	clientKeepaliveTimeout = 10 * time.Second
)

// ClientConfig configures the bot's dial side of the reverse bot-link.
type ClientConfig struct {
	// GatewayAddr is the gateway bot-link endpoint to dial.
	GatewayAddr string
	// InstanceID identifies this bot to the gateway.
	InstanceID string
	// OwnsUpdates reports whether this bot runs the exclusive getUpdates long-poll.
	OwnsUpdates bool
	// Creds is the transport credentials for the dial (mutual TLS in production,
	// built from pkg/mtls).
	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
// it receives and re-dialing after any break. The same mTLS connection also serves
// the unary chat-eligibility query the bot makes on a chat join.
type Client struct {
	cfg    ClientConfig
	exec   *Executor
	log    *zap.Logger
	conn   *grpc.ClientConn
	client botlinkv1.BotLinkClient
}

// NewClient builds the bot-link client over the executor, dialing the gateway. The
// gRPC connection is lazy, so the dial does not block on the gateway being up; the
// caller must Close it. The bot-link command stream is opened by Run.
func NewClient(cfg ClientConfig, exec *Executor, log *zap.Logger) (*Client, error) {
	if log == nil {
		log = zap.NewNop()
	}
	conn, err := grpc.NewClient(cfg.GatewayAddr,
		grpc.WithTransportCredentials(cfg.Creds),
		grpc.WithStatsHandler(otelgrpc.NewClientHandler()),
		grpc.WithKeepaliveParams(keepalive.ClientParameters{
			Time:                clientKeepaliveTime,
			Timeout:             clientKeepaliveTimeout,
			PermitWithoutStream: true,
		}),
	)
	if err != nil {
		return nil, err
	}
	return &Client{cfg: cfg, exec: exec, log: log, conn: conn, client: botlinkv1.NewBotLinkClient(conn)}, nil
}

// Close releases the bot-link connection.
func (c *Client) Close() error { return c.conn.Close() }

// ResolveChatEligibility asks the gateway whether the Telegram user identified by
// externalID may write in the moderated chat. The bot calls it on a chat join, over
// the same mTLS connection as the command stream.
func (c *Client) ResolveChatEligibility(ctx context.Context, externalID string) (bool, error) {
	resp, err := c.client.ResolveChatEligibility(ctx, &botlinkv1.ChatEligibilityRequest{ExternalId: externalID})
	if err != nil {
		return false, err
	}
	return resp.GetEligible(), nil
}

// ValidatePreCheckout asks the gateway whether a Telegram Stars pre_checkout_query for orderID
// paying amount in currency may be approved, before the charge. The bot calls it on every
// pre_checkout_query over the same mTLS connection and approves only on ok; the reason is a short
// decline message (already localised by the backend) to show the payer.
func (c *Client) ValidatePreCheckout(ctx context.Context, orderID string, amount int64, currency string) (ok bool, reason string, err error) {
	resp, err := c.client.ValidatePreCheckout(ctx, &botlinkv1.PreCheckoutRequest{OrderId: orderID, Amount: amount, Currency: currency})
	if err != nil {
		return false, "", err
	}
	return resp.GetOk(), resp.GetReason(), nil
}

// ForwardPayment delivers a completed Stars payment to the gateway for crediting. The bot calls it
// from the outbox drain; it reports whether the order was credited (or already had been). A non-nil
// error is a transient failure the bot retries.
func (c *Client) ForwardPayment(ctx context.Context, orderID, chargeID string, amount, telegramUserID int64) (credited bool, err error) {
	resp, err := c.client.ForwardPayment(ctx, &botlinkv1.ForwardPaymentRequest{
		OrderId:                 orderID,
		TelegramPaymentChargeId: chargeID,
		Amount:                  amount,
		TelegramUserId:          telegramUserID,
	})
	if err != nil {
		return false, err
	}
	return resp.GetCredited(), nil
}

// Run keeps the bot-link command stream open, re-opening it after each break, until
// ctx is cancelled. The gRPC connection auto-reconnects the transport underneath.
func (c *Client) Run(ctx context.Context) error {
	for ctx.Err() == nil {
		if err := c.serve(ctx, c.client); err != nil && ctx.Err() == nil {
			c.log.Warn("bot-link stream ended", zap.Error(err))
		}
		if !sleep(ctx, c.cfg.ReconnectDelay) {
			break
		}
	}
	return ctx.Err()
}

// serve opens one Link stream, registers with a Hello, then executes commands and
// replies with an Ack each until the stream ends.
func (c *Client) serve(ctx context.Context, client botlinkv1.BotLinkClient) error {
	stream, err := client.Link(ctx)
	if err != nil {
		return err
	}
	if err := stream.Send(&botlinkv1.FromBot{Msg: &botlinkv1.FromBot_Hello{Hello: &botlinkv1.Hello{
		InstanceId:  c.cfg.InstanceID,
		OwnsUpdates: c.cfg.OwnsUpdates,
	}}}); err != nil {
		return err
	}
	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 {
		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)
		}
	}
}

// sleep waits for d or until ctx is cancelled, reporting whether it waited the full
// duration.
func sleep(ctx context.Context, d time.Duration) bool {
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-t.C:
		return true
	}
}