go-sip/cmd/sip-go-agent/dispatcher_sip.go

package main

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"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/logging"
	"git.ipao.vip/rogee/go-sip/internal/mq"
	"git.ipao.vip/rogee/go-sip/internal/rpc"
	"git.ipao.vip/rogee/go-sip/internal/store"
	"github.com/google/uuid"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

// runSIPOnlyDispatcher owns no business queue, task discovery, recording
// listener, OSS signer, call executor or call admission. It consumes only the
// assigned control queue; unrelated controls are requeued and stop this lane.
func runSIPOnlyDispatcher(ctx context.Context, settings config.DispatcherRuntimeEnvironment) (result error) {
	endpoints, err := config.LoadAgentEndpoints(settings.AgentEndpointsFile)
	if err != nil {
		return err
	}
	if len(endpoints) != 1 {
		return errors.New("SIP-only Dispatcher requires exactly one assigned Agent")
	}
	endpoint := endpoints[0]
	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
	}
	clientTLS, err := rpc.NewClientTLSConfig(ca, cert, key, endpoint.ServerName)
	if err != nil {
		return errors.New("SIP-only Dispatcher mTLS Agent certificate is invalid")
	}
	httpClient, err := dispatcherSaaSHTTPClient("sip-only", ca)
	if err != nil {
		return err
	}
	_, reader, err := dispatcherConfigurationClient("sip-only", httpClient)
	if err != nil {
		return err
	}
	db, err := store.Open(settings.SQLitePath)
	if err != nil {
		return err
	}
	defer func() { result = errors.Join(result, db.Close()) }()
	if err := db.CloseAdmission(settings.DispatcherID); err != nil {
		return 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(endpoint.Address, grpc.WithTransportCredentials(credentials.NewTLS(clientTLS)), grpc.WithChainUnaryInterceptor(logging.UnaryClientInterceptor("dispatcher", "agent")))
	if err != nil {
		return errors.New("SIP-only Dispatcher cannot create pinned Agent connection")
	}
	defer func() { result = errors.Join(result, connection.Close()) }()
	agent := agentpb.NewAgentControlServiceClient(connection)
	coordinator := dispatcher.NewAgentCoordinator(time.Now)
	if err := coordinator.Register(endpoint.AgentID, agent); err != nil {
		return err
	}
	probeCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
	status, err := coordinator.Probe(probeCtx, endpoint.AgentID, endpoint.CellID)
	if err != nil {
		cancel()
		return fmt.Errorf("probe SIP-only assigned Agent: %w", err)
	}
	epoch, err := uuid.NewRandom()
	if err != nil {
		cancel()
		return err
	}
	session, err := coordinator.Activate(probeCtx, settings.DispatcherID, endpoint.AgentID, endpoint.CellID, status.BootId, epoch.String(), 0)
	cancel()
	if err != nil {
		return fmt.Errorf("activate SIP-only Agent session: %w", err)
	}
	originator := &dispatcher.ApprovedOriginator{DispatcherID: settings.DispatcherID, Client: agent, Meta: func(callCtx context.Context) (*agentpb.RequestMeta, error) {
		return coordinator.ApprovedMeta(callCtx, endpoint.AgentID)
	}}
	lane := dispatcher.SIPOnly{DispatcherID: settings.DispatcherID, Store: db, Client: reader, ApplySIP: originator.ApplySIP, VerifySIP: originator.VerifySIP}
	if _, err := broker.DrainControlPredeclared(ctx, broker.ControlQueue(), lane.HandleNotification); err != nil {
		return fmt.Errorf("drain assigned SIP controls before startup: %w", err)
	}
	if err := lane.Sync(ctx); err != nil {
		return fmt.Errorf("initial native SIP-only sync: %w", err)
	}
	serveCtx, stop := context.WithCancel(ctx)
	defer stop()
	renewErrors := make(chan error, 1)
	go func() {
		renewErr := maintainAgentSession(serveCtx, session, 5*time.Minute, func(c context.Context, previous dispatcher.AgentSession) (dispatcher.AgentSession, error) {
			return coordinator.Renew(c, settings.DispatcherID, previous)
		})
		renewErrors <- renewErr
		stop()
	}()
	consumer, err := broker.StartPredeclaredConsumer(serveCtx, broker.ControlQueue(), lane.HandleNotification)
	if err != nil {
		return err
	}
	defer func() {
		waitCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		result = errors.Join(result, consumer.Stop(waitCtx))
	}()
	consumerErrors := make(chan error, 1)
	go func() { consumerErrors <- consumer.Wait(context.Background()); stop() }()
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return nil
		case err := <-renewErrors:
			if err != nil {
				return fmt.Errorf("SIP-only Agent session renewal failed: %w", err)
			}
			return errors.New("SIP-only Agent session stopped unexpectedly")
		case err := <-consumerErrors:
			if err != nil {
				return fmt.Errorf("SIP-only control queue stopped: %w", err)
			}
			return errors.New("SIP-only control queue stopped unexpectedly")
		case err := <-broker.Done():
			return fmt.Errorf("SIP-only broker connection lost: %v", err)
		case <-ticker.C:
			_, pending, err := db.SIPState(settings.DispatcherID)
			if err != nil {
				return err
			}
			if pending == 0 {
				continue
			}
			readCtx, cancel := context.WithTimeout(serveCtx, 20*time.Second)
			err = lane.Sync(readCtx)
			cancel()
			if err != nil {
				slog.Warn("SIP-only revision remains durably pending", "dispatcher_id", settings.DispatcherID, "pending_revision", pending, "error", err)
			}
		}
	}
}