go-sip/cmd/sip-go-agent/dispatcher_command.go

package main

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"time"

	agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
	"git.ipao.vip/rogee/go-sip/internal/config"
	"git.ipao.vip/rogee/go-sip/internal/dispatcher"
	"git.ipao.vip/rogee/go-sip/internal/mq"
	"git.ipao.vip/rogee/go-sip/internal/oss"
	"git.ipao.vip/rogee/go-sip/internal/rpc"
	"git.ipao.vip/rogee/go-sip/internal/store"
	"github.com/google/uuid"
	"github.com/spf13/cobra"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

func newDispatcherCommand() *cobra.Command {
	var mode string
	command := &cobra.Command{
		Use: "dispatcher", Short: "Run the isolated Dispatcher", Args: cobra.NoArgs,
		SilenceUsage: true,
		RunE: func(cmd *cobra.Command, _ []string) error {
			return runDispatcher(cmd.Context(), mode)
		},
	}
	command.Flags().StringVar(&mode, "mode", "mock", "isolated Mock mode only")
	return command
}

// runDispatcher does not publish or consume until the predeclared MQ
// topology, Agent session and approved SIP load have all been verified.
func runDispatcher(ctx context.Context, mode string) (result error) {
	settings, err := config.LoadDispatcherRuntimeEnvironment(mode)
	if err != nil {
		return err
	}
	agentEndpoint, err := config.LoadMockAgentEndpoint(settings.AgentEndpointsFile)
	if err != nil {
		return err
	}
	ossConfiguration, err := config.LoadOSSConfig(settings.OSSConfigFile, settings.DispatcherID)
	if err != nil {
		return err
	}
	ca, err := readAgentPEM("MTLS_CA_FILE", settings.CAFile)
	if err != nil {
		return err
	}
	cert, err := readAgentPEM("MTLS_CERT_FILE", settings.CertFile)
	if err != nil {
		return err
	}
	key, err := readAgentPEM("MTLS_KEY_FILE", settings.KeyFile)
	if err != nil {
		return err
	}
	serverTLS, err := rpc.NewServerTLSConfig(ca, cert, key)
	if err != nil {
		return errors.New("Dispatcher mTLS listener certificate is invalid")
	}
	clientTLS, err := rpc.NewClientTLSConfig(ca, cert, key, agentEndpoint.ServerName)
	if err != nil {
		return errors.New("Dispatcher mTLS Agent certificate configuration is invalid")
	}
	httpClient, err := localMockHTTPClient(ca)
	if err != nil {
		return err
	}
	httpClient.Timeout = 10 * time.Second
	_, reader, err := dispatcherConfigurationClient(mode, httpClient)
	if err != nil {
		return err
	}
	ossClient, err := oss.NewClient(ossConfiguration)
	if err != nil {
		return fmt.Errorf("initialize bounded OSS grant signer: %w", err)
	}
	database, err := store.Open(settings.SQLitePath)
	if err != nil {
		return err
	}
	defer func() { result = errors.Join(result, database.Close()) }()
	if err := database.CloseAdmission(settings.DispatcherID); err != nil {
		return fmt.Errorf("close task admission before connecting external adapters: %w", err)
	}
	broker, err := mq.Open(settings.RabbitMQURL, settings.DispatcherID, 1)
	if err != nil {
		return err
	}
	defer func() { result = errors.Join(result, broker.Close()) }()
	connection, err := grpc.NewClient(agentEndpoint.Address, grpc.WithTransportCredentials(credentials.NewTLS(clientTLS)))
	if err != nil {
		return errors.New("Dispatcher cannot create verified Agent connection")
	}
	defer func() { result = errors.Join(result, connection.Close()) }()
	agentClient := agentpb.NewAgentControlServiceClient(connection)
	coordinator := dispatcher.NewAgentCoordinator(time.Now)
	if err := coordinator.Register(agentEndpoint.AgentID, agentClient); err != nil {
		return err
	}
	probeCtx, cancelProbe := context.WithTimeout(ctx, 5*time.Second)
	status, err := coordinator.Probe(probeCtx, agentEndpoint.AgentID, agentEndpoint.CellID)
	if err != nil {
		cancelProbe()
		return fmt.Errorf("probe assigned Agent before admission: %w", err)
	}
	epoch, err := uuid.NewRandom()
	if err != nil {
		cancelProbe()
		return fmt.Errorf("create Dispatcher epoch: %w", err)
	}
	session, err := coordinator.Activate(probeCtx, settings.DispatcherID, agentEndpoint.AgentID, agentEndpoint.CellID, status.BootId, epoch.String(), 0)
	if err != nil {
		cancelProbe()
		return fmt.Errorf("activate assigned Agent before admission: %w", err)
	}
	cancelProbe()
	originator := &dispatcher.ApprovedOriginator{
		DispatcherID: settings.DispatcherID,
		Client:       agentClient,
		Meta: func(callCtx context.Context) (*agentpb.RequestMeta, error) {
			return coordinator.ApprovedMeta(callCtx, agentEndpoint.AgentID)
		},
	}
	worker := &dispatcher.Runtime{
		Broker: broker,
		Bootstrap: dispatcher.Bootstrap{
			DispatcherID: settings.DispatcherID, Client: reader, Store: database, VerifySIP: originator.VerifySIP,
		},
		Execute: dispatcher.ExecuteController{
			DispatcherID: settings.DispatcherID, Store: database, Originator: originator, Publisher: broker, Now: time.Now,
		},
		Control: dispatcher.ControlController{
			DispatcherID: settings.DispatcherID, Store: database, Client: reader, Agent: originator, Now: time.Now,
		},
		PollInterval: time.Second, Logger: slog.Default(),
	}
	listener, err := net.Listen("tcp", settings.Listen)
	if err != nil {
		return fmt.Errorf("Dispatcher cannot listen on configured Mock address: %w", err)
	}
	server := grpc.NewServer(grpc.Creds(credentials.NewTLS(serverTLS)))
	agentpb.RegisterAgentControlServiceServer(server, &rpc.RecordingServer{
		DispatcherID: settings.DispatcherID, Store: database, OSS: ossClient,
		TrustedFingerprints: settings.PeerFingerprints, AuthorizeSession: coordinator.AuthorizeInboundMeta,
	})
	serveCtx, cancelServe := context.WithCancel(ctx)
	defer cancelServe()
	renewErrors := make(chan error, 1)
	go func() {
		renewErr := maintainAgentSession(serveCtx, session, 5*time.Minute, func(renewCtx context.Context, previous dispatcher.AgentSession) (dispatcher.AgentSession, error) {
			return coordinator.Renew(renewCtx, settings.DispatcherID, previous)
		})
		if renewErr != nil {
			closeErr := database.CloseAdmission(settings.DispatcherID)
			renewErrors <- errors.Join(renewErr, closeErr)
			cancelServe()
			return
		}
		renewErrors <- nil
	}()
	serverErrors := make(chan error, 1)
	go func() {
		serverErrors <- server.Serve(listener)
		cancelServe()
	}()
	runtimeErr := worker.Serve(serveCtx)
	cancelServe()
	server.GracefulStop()
	serverErr := <-serverErrors
	renewErr := <-renewErrors
	if renewErr != nil {
		if errors.Is(serverErr, grpc.ErrServerStopped) {
			serverErr = nil
		}
		return errors.Join(runtimeErr, fmt.Errorf("Agent session renewal failed: %w", renewErr), serverErr)
	}
	if serverErr != nil && !errors.Is(serverErr, grpc.ErrServerStopped) {
		return errors.Join(runtimeErr, fmt.Errorf("Dispatcher recording listener stopped: %w", serverErr))
	}
	if runtimeErr != nil {
		return runtimeErr
	}
	if ctx.Err() == nil {
		return errors.New("Dispatcher stopped without process shutdown")
	}
	return nil
}