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

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
}

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

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

	for {
		msg, err := stream.Recv()
		if err != nil {
			return err
		}
		cmd := msg.GetCommand()
		if cmd == nil {
			continue
		}
		delivered, herr := c.exec.Handle(ctx, cmd)
		ack := &botlinkv1.Ack{CommandId: cmd.GetCommandId(), Delivered: delivered}
		if herr != nil {
			ack.Error = herr.Error()
		}
		if err := stream.Send(&botlinkv1.FromBot{Msg: &botlinkv1.FromBot_Ack{Ack: ack}}); err != nil {
			return err
		}
	}
}

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