lislgosms/gateway/cmd/gateway/main.go

package main

import (
	"context"
	"log"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"

	"cmpp-platform/gateway/internal/control"
	"cmpp-platform/gateway/internal/health"
	"cmpp-platform/gateway/internal/inbound"
	platformmetrics "cmpp-platform/gateway/internal/metrics"
	"cmpp-platform/gateway/internal/queue"
	"cmpp-platform/gateway/internal/ratelimit"
	"cmpp-platform/gateway/internal/resultoutbox"
	"cmpp-platform/gateway/internal/submitworker"
	"cmpp-platform/gateway/internal/upstream"

	"github.com/redis/go-redis/v9"
)

func main() {
	addr := os.Getenv("GATEWAY_HEALTH_ADDR")
	if addr == "" {
		addr = ":8090"
	}
	cmppAddr := os.Getenv("GATEWAY_CMPP_ADDR")
	if cmppAddr == "" {
		cmppAddr = ":17890"
	}
	apiBaseURL := os.Getenv("API_BASE_URL")
	upstreamManager := &upstream.Manager{APIBaseURL: apiBaseURL}
	var worker *submitworker.Worker
	var resultOutbox *resultoutbox.Outbox
	channelLimiter, err := ratelimit.New(os.Getenv("REDIS_URL"))
	if err != nil {
		log.Fatalf("gateway channel rate limiter init failed: %v", err)
	}
	presenceStore, err := inbound.NewRedisPresenceStore(os.Getenv("REDIS_URL"))
	if err != nil {
		log.Printf("gateway downstream presence store init failed: %v", err)
	}
	recoveryStore, err := inbound.NewRedisRecoveryStore(os.Getenv("REDIS_URL"))
	if err != nil {
		log.Printf("gateway downstream recovery store init failed: %v", err)
	}

	go func() {
		log.Printf("cmpp gateway inbound server listening on %s", cmppAddr)
		inboundConcurrency := positiveEnvInt("GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY", 64)
		if err := (inbound.Server{
			Addr:                 cmppAddr,
			APIBaseURL:           apiBaseURL,
			HTTPClient:           inbound.NewAPIHTTPClient(16),
			SubmitHTTPClient:     inbound.NewAPIHTTPClient(inboundConcurrency),
			PresenceStore:        presenceStore,
			RecoveryStore:        recoveryStore,
			GatewayInstanceID:    getenv("GATEWAY_INSTANCE_ID", hostname()),
			MaxSubmitConcurrency: inboundConcurrency,
			SecurityEventToken:   os.Getenv("SECURITY_EVENT_TOKEN"),
		}).ListenAndServe(); err != nil {
			log.Fatalf("gateway inbound server stopped: %v", err)
		}
	}()

	if os.Getenv("GATEWAY_SUBMIT_WORKER_DISABLED") != "true" {
		worker, err = submitworker.New(os.Getenv("REDIS_URL"), upstreamManager)
		if err != nil {
			log.Printf("gateway submit worker init failed: %v", err)
		} else {
			worker.Stream = getenv("GATEWAY_SUBMIT_STREAM", "gateway.submit.commands")
			worker.Group = getenv("GATEWAY_SUBMIT_GROUP", "cmpp-gateway")
			worker.Consumer = getenv("GATEWAY_SUBMIT_CONSUMER", "gateway-1")
			worker.Concurrency = positiveEnvInt("GATEWAY_SUBMIT_WORKER_CONCURRENCY", 64)
			worker.APIBaseURL = apiBaseURL
			resultOutbox = resultoutbox.New(worker.Redis)
			resultOutbox.Stream = getenv("GATEWAY_SUBMIT_RESULT_STREAM", "gateway.submit.results")
			resultOutbox.Group = getenv("GATEWAY_SUBMIT_RESULT_GROUP", "cmpp-api-callback")
			resultOutbox.Consumer = getenv("GATEWAY_SUBMIT_RESULT_CONSUMER", "gateway-1")
			resultOutbox.APIBaseURL = apiBaseURL
			resultOutbox.Concurrency = positiveEnvInt("GATEWAY_SUBMIT_RESULT_WORKER_CONCURRENCY", 8)
			worker.ResultOutbox = resultOutbox
			upstreamManager.SubmitSegmentPublisher = resultOutbox
			go func() {
				log.Printf("cmpp gateway submit worker consuming stream=%s group=%s consumer=%s", worker.Stream, worker.Group, worker.Consumer)
				if err := worker.Run(context.Background()); err != nil {
					log.Printf("gateway submit worker stopped: %v", err)
				}
			}()
			go func() {
				log.Printf("cmpp gateway result Outbox consuming stream=%s group=%s consumer=%s", resultOutbox.StreamName(), resultOutbox.GroupName(), resultOutbox.Consumer)
				if err := resultOutbox.Run(context.Background()); err != nil {
					log.Printf("gateway result Outbox worker stopped: %v", err)
				}
			}()
		}
	}

	mux := http.NewServeMux()
	mux.Handle("/health", health.Handler())
	mux.Handle("/metrics", platformmetrics.Handler(func(ctx context.Context) platformmetrics.Snapshot {
		desired, connected := upstreamManager.ConnectionCounts()
		snapshot := platformmetrics.Snapshot{
			UpstreamDesired: desired, UpstreamConnected: connected,
			DownstreamConnected: inbound.ActiveConnectionCount(), SubmitWorkerUp: worker != nil,
		}
		if worker != nil {
			snapshot.SubmitWorkerConcurrency = worker.ConfiguredConcurrency()
			snapshot.SubmitWorkerInFlight = worker.InFlight()
		}
		if resultOutbox != nil {
			snapshot.ResultWorkerUp = true
			snapshot.ResultWorkerConcurrency = resultOutbox.ConfiguredConcurrency()
			snapshot.ResultWorkerInFlight = resultOutbox.InFlight()
		}
		snapshot.InboundSubmitConcurrency, snapshot.InboundSubmitInFlight = inbound.SubmitSlotSnapshot()
		if worker == nil || worker.Redis == nil {
			return snapshot
		}
		pending, err := worker.Redis.XPending(ctx, worker.Stream, worker.Group).Result()
		if err != nil {
			return snapshot
		}
		snapshot.QueueAvailable = true
		snapshot.QueuePending = pending.Count
		groups, err := worker.Redis.XInfoGroups(ctx, worker.Stream).Result()
		if err == nil {
			for _, group := range groups {
				if group.Name == worker.Group {
					snapshot.QueueLag = group.Lag
					break
				}
			}
		}
		entries, err := worker.Redis.XPendingExt(ctx, &redis.XPendingExtArgs{Stream: worker.Stream, Group: worker.Group, Start: "-", End: "+", Count: 1}).Result()
		if err == nil && len(entries) > 0 {
			milliseconds, parseErr := strconv.ParseInt(strings.SplitN(entries[0].ID, "-", 2)[0], 10, 64)
			if parseErr == nil {
				snapshot.QueueOldestAgeSeconds = max(0, time.Since(time.UnixMilli(milliseconds)).Seconds())
			}
		}
		if resultOutbox != nil {
			resultPending, resultErr := worker.Redis.XPending(ctx, resultOutbox.StreamName(), resultOutbox.GroupName()).Result()
			if resultErr == nil {
				snapshot.ResultQueueAvailable = true
				snapshot.ResultQueuePending = resultPending.Count
			}
			resultGroups, resultErr := worker.Redis.XInfoGroups(ctx, resultOutbox.StreamName()).Result()
			if resultErr == nil {
				for _, group := range resultGroups {
					if group.Name == resultOutbox.GroupName() {
						snapshot.ResultQueueLag = group.Lag
						break
					}
				}
			}
		}
		return snapshot
	}))
	control.Register(mux, control.Server{
		APIBaseURL: apiBaseURL,
		Upstream:   upstreamManager,
		Limiter:    channelLimiter,
		Submit: func(ctx context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
			result, submitErr := upstreamManager.Submit(ctx, command)
			if resultOutbox == nil {
				return result, submitErr
			}
			if publishErr := resultOutbox.PublishSubmitResult(ctx, command, result); publishErr != nil {
				return result, publishErr
			}
			return result, submitErr
		},
		RecoveryCandidates: func(ctx context.Context) ([]inbound.DownstreamPresence, error) {
			return inbound.ListRecoveryCandidates(ctx, presenceStore)
		},
		RecoveryStatuses: func(ctx context.Context) ([]inbound.DownstreamRecoveryStatus, error) {
			if recoveryStore == nil {
				return nil, nil
			}
			return recoveryStore.ListRecoveryStatuses(ctx)
		},
	})

	log.Printf("cmpp gateway control server listening on %s", addr)
	if err := http.ListenAndServe(addr, mux); err != nil {
		log.Fatalf("gateway control server stopped: %v", err)
	}
}

func getenv(key string, fallback string) string {
	value := os.Getenv(key)
	if value == "" {
		return fallback
	}
	return value
}

func positiveEnvInt(key string, fallback int) int {
	value, err := strconv.Atoi(os.Getenv(key))
	if err != nil || value <= 0 {
		return fallback
	}
	return value
}

func hostname() string {
	name, err := os.Hostname()
	if err != nil || name == "" {
		return "gateway-1"
	}
	return name
}