mxaccessgw/clients/go/mxgateway/client.go

// Package mxgateway is the Go client for the MXAccess Gateway gRPC service.
//
// The package wraps the generated gRPC contract with session-oriented helpers
// for invoking MXAccess commands, streaming events, and browsing the Galaxy
// Repository. Authentication uses an API-key bearer token attached as gRPC
// metadata on every call.
//
// Typical use opens a Client with Dial, opens a Session, invokes commands such
// as Register, AddItem, Advise, and Write, and consumes events via
// SubscribeEvents. Galaxy Repository browse RPCs are exposed through
// GalaxyClient.
package mxgateway

import (
	"context"
	"crypto/tls"
	"errors"
	"strings"
	"time"

	pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
	"google.golang.org/grpc"
	"google.golang.org/grpc/connectivity"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/protobuf/types/known/durationpb"
)

const (
	defaultDialTimeout = 10 * time.Second
	defaultCallTimeout = 30 * time.Second
)

// Client owns a gateway gRPC connection and exposes session-oriented helpers.
type Client struct {
	conn *grpc.ClientConn
	raw  pb.MxAccessGatewayClient
	opts Options
}

// Dial opens a gRPC connection to the gateway and configures auth metadata,
// transport security, and blocking dial cancellation from ctx.
func Dial(ctx context.Context, opts Options) (*Client, error) {
	if opts.Endpoint == "" {
		return nil, errors.New("mxgateway: endpoint is required")
	}

	dialCtx := ctx
	cancel := func() {}
	if opts.DialTimeout > 0 {
		dialCtx, cancel = context.WithTimeout(ctx, opts.DialTimeout)
	} else if _, ok := ctx.Deadline(); !ok {
		dialCtx, cancel = context.WithTimeout(ctx, defaultDialTimeout)
	}
	defer cancel()

	transportCredentials, err := resolveTransportCredentials(opts)
	if err != nil {
		return nil, err
	}

	dialOptions := []grpc.DialOption{
		grpc.WithTransportCredentials(transportCredentials),
		grpc.WithUnaryInterceptor(unaryAuthInterceptor(opts.APIKey)),
		grpc.WithStreamInterceptor(streamAuthInterceptor(opts.APIKey)),
	}
	dialOptions = append(dialOptions, opts.DialOptions...)

	conn, err := dialConnected(dialCtx, opts.Endpoint, dialOptions)
	if err != nil {
		return nil, &GatewayError{Op: "dial", Err: err}
	}

	return NewClient(conn, opts), nil
}

// NewClient wraps an existing gRPC connection. The caller owns closing conn
// unless it calls Close on the returned Client.
func NewClient(conn *grpc.ClientConn, opts Options) *Client {
	return &Client{
		conn: conn,
		raw:  pb.NewMxAccessGatewayClient(conn),
		opts: opts,
	}
}

// RawClient returns the generated gRPC client for command-specific parity tests.
func (c *Client) RawClient() RawGatewayClient {
	return c.raw
}

// OpenSession creates a gateway-backed MXAccess session.
func (c *Client) OpenSession(ctx context.Context, opts OpenSessionOptions) (*Session, error) {
	reply, err := c.OpenSessionRaw(ctx, opts.Request())
	if err != nil {
		return nil, err
	}

	return newSession(c, reply), nil
}

// OpenSessionRaw sends a raw OpenSession request and validates protocol status.
func (c *Client) OpenSessionRaw(ctx context.Context, req *OpenSessionRequest) (*OpenSessionReply, error) {
	if req == nil {
		return nil, errors.New("mxgateway: open session request is required")
	}

	callCtx, cancel := c.callContext(ctx)
	defer cancel()

	reply, err := c.raw.OpenSession(callCtx, req)
	if err != nil {
		return nil, &GatewayError{Op: "open session", Err: err}
	}
	if err := EnsureProtocolSuccess("open session", reply.GetProtocolStatus(), nil); err != nil {
		return reply, err
	}

	return reply, nil
}

// Invoke sends a raw MXAccess command request and validates protocol and
// MXAccess status fields while preserving the raw reply on typed errors.
func (c *Client) Invoke(ctx context.Context, req *MxCommandRequest) (*MxCommandReply, error) {
	if req == nil {
		return nil, errors.New("mxgateway: command request is required")
	}

	callCtx, cancel := c.callContext(ctx)
	defer cancel()

	reply, err := c.raw.Invoke(callCtx, req)
	if err != nil {
		return nil, &GatewayError{Op: "invoke", Err: err}
	}
	if err := EnsureProtocolSuccess("invoke", reply.GetProtocolStatus(), reply); err != nil {
		return reply, err
	}
	if err := EnsureMxAccessSuccess("invoke", reply); err != nil {
		return reply, err
	}

	return reply, nil
}

// CloseSessionRaw sends a raw CloseSession request and validates protocol
// status.
func (c *Client) CloseSessionRaw(ctx context.Context, req *CloseSessionRequest) (*CloseSessionReply, error) {
	if req == nil {
		return nil, errors.New("mxgateway: close session request is required")
	}

	callCtx, cancel := c.callContext(ctx)
	defer cancel()

	reply, err := c.raw.CloseSession(callCtx, req)
	if err != nil {
		return nil, &GatewayError{Op: "close session", Err: err}
	}
	if err := EnsureProtocolSuccess("close session", reply.GetProtocolStatus(), nil); err != nil {
		return reply, err
	}

	return reply, nil
}

// StreamEventsRaw starts the generated event stream for callers that need direct
// control over Recv.
func (c *Client) StreamEventsRaw(ctx context.Context, req *StreamEventsRequest) (RawEventStream, error) {
	if req == nil {
		return nil, errors.New("mxgateway: stream events request is required")
	}

	stream, err := c.raw.StreamEvents(ctx, req)
	if err != nil {
		return nil, &GatewayError{Op: "stream events", Err: err}
	}

	return stream, nil
}

// Close closes the underlying gRPC connection.
func (c *Client) Close() error {
	if c == nil || c.conn == nil {
		return nil
	}

	return c.conn.Close()
}

func (c *Client) callContext(ctx context.Context) (context.Context, context.CancelFunc) {
	timeout := c.opts.CallTimeout
	if timeout == 0 {
		timeout = defaultCallTimeout
	}
	if timeout < 0 {
		return ctx, func() {}
	}
	if deadline, ok := ctx.Deadline(); ok {
		timeoutDeadline := time.Now().Add(timeout)
		if deadline.Before(timeoutDeadline) {
			return ctx, func() {}
		}
	}
	return context.WithTimeout(ctx, timeout)
}

func resolveTransportCredentials(opts Options) (credentials.TransportCredentials, error) {
	if opts.TransportCredentials != nil {
		return opts.TransportCredentials, nil
	}
	if opts.Plaintext {
		return insecure.NewCredentials(), nil
	}
	if opts.CACertFile != "" {
		return credentials.NewClientTLSFromFile(opts.CACertFile, opts.ServerNameOverride)
	}
	if opts.TLSConfig != nil {
		cfg := opts.TLSConfig.Clone()
		if opts.ServerNameOverride != "" {
			cfg.ServerName = opts.ServerNameOverride
		}
		return credentials.NewTLS(cfg), nil
	}

	return credentials.NewTLS(tlsConfigForOptions(opts)), nil
}

// tlsConfigForOptions returns the *tls.Config for the no-CA, no-custom-config TLS path.
// It returns nil when the caller should use a different credentials path (CA file or custom TLSConfig).
// Exposed as an internal helper so unit tests can assert the InsecureSkipVerify posture.
func tlsConfigForOptions(opts Options) *tls.Config {
	// CA file and custom TLSConfig take their own paths in resolveTransportCredentials.
	if opts.CACertFile != "" || opts.TLSConfig != nil {
		return nil
	}
	return &tls.Config{
		MinVersion:         tls.VersionTLS12,
		ServerName:         opts.ServerNameOverride,
		InsecureSkipVerify: !opts.RequireCertificateValidation, //nolint:gosec // internal tool; self-signed gateway cert expected; opt-in strict via RequireCertificateValidation
	}
}

// OpenSessionOptions describes fields used to create an OpenSessionRequest.
type OpenSessionOptions struct {
	// RequestedBackend selects the gateway worker backend (empty for default).
	RequestedBackend string
	// ClientSessionName is a human-readable name recorded on the session.
	ClientSessionName string
	// ClientCorrelationID echoes through gateway logs and replies for tracing.
	ClientCorrelationID string
	// CommandTimeout sets the per-command timeout the gateway forwards to the
	// worker; zero leaves the gateway default in place.
	CommandTimeout time.Duration
}

// Request returns the raw protobuf OpenSessionRequest for these options.
func (o OpenSessionOptions) Request() *OpenSessionRequest {
	req := &OpenSessionRequest{
		RequestedBackend:    o.RequestedBackend,
		ClientSessionName:   o.ClientSessionName,
		ClientCorrelationId: o.ClientCorrelationID,
	}
	if o.CommandTimeout > 0 {
		req.CommandTimeout = durationpb.New(o.CommandTimeout)
	}
	return req
}

// dialConnected creates a connection with grpc.NewClient and then waits for it
// to become READY, bounded by ctx. This preserves the connect-or-fail contract
// the removed grpc.DialContext + grpc.WithBlock combination gave Dial callers.
// The endpoint is resolved with the passthrough scheme unless it already names a
// scheme, matching DialContext's historical default (NewClient defaults to dns).
func dialConnected(ctx context.Context, endpoint string, dialOptions []grpc.DialOption) (*grpc.ClientConn, error) {
	target := endpoint
	if !strings.Contains(target, "://") {
		target = "passthrough:///" + target
	}
	conn, err := grpc.NewClient(target, dialOptions...)
	if err != nil {
		return nil, err
	}
	conn.Connect()
	for {
		state := conn.GetState()
		if state == connectivity.Ready {
			return conn, nil
		}
		if state == connectivity.Shutdown {
			_ = conn.Close()
			return nil, errors.New("connection shut down while connecting")
		}
		if !conn.WaitForStateChange(ctx, state) {
			_ = conn.Close()
			return nil, ctx.Err()
		}
	}
}