vantage/server/cmd/main.go

package main

import (
	"context"
	"errors"
	"log"
	"net/http"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/api"
	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/auth"
	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/bus"
	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/db"
	grpcserver "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/grpc"
	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/monitorsched"
	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/services"
	"github.com/gin-gonic/gin"
)

func main() {
	mongoURI := getEnv("MONGO_URI", "mongodb://localhost:27017")

	// Two flags, so the schema work can be lifted out of the serving pods.
	//
	// Under Docker Compose neither is set and nothing changes: one process
	// migrates and then serves. Under Kubernetes with more than one replica
	// that is unsafe — every pod would run MigrateOrgToInstance at once, and
	// renaming collections while a sibling reads them is not a race anyone
	// wins. The chart therefore runs a pre-upgrade Job with MIGRATE_ONLY and
	// starts the Deployment with SKIP_MIGRATIONS.
	migrateOnly := boolEnv("VANTAGE_MIGRATE_ONLY")
	skipMigrations := boolEnv("VANTAGE_SKIP_MIGRATIONS")

	// Not required in migrate-only mode: that process never serves gRPC, and
	// demanding it would put an agent-facing address in a Job that has no
	// business knowing one.
	if !migrateOnly && os.Getenv("GRPC_HOST") == "" {
		log.Fatal("GRPC_HOST is required (host:port agents dial for gRPC)")
	}

	// DB name comes from the MONGO_URI path; "vantage" is the fallback.
	if err := db.Connect(mongoURI, "vantage"); err != nil {
		log.Fatalf("failed to connect to MongoDB: %v", err)
	}
	log.Println("connected to MongoDB")

	if migrateOnly || !skipMigrations {
		runSchemaSetup()
	} else {
		log.Println("VANTAGE_SKIP_MIGRATIONS set: assuming migrations ran elsewhere")
	}

	if migrateOnly {
		log.Println("VANTAGE_MIGRATE_ONLY set: schema setup complete, exiting")
		return
	}

	serve()
}

// runSchemaSetup performs every write that must happen exactly once before the
// application serves: migrations, index builders and default step seeding. It
// is fatal on anything that would leave the schema half-moved.
func runSchemaSetup() {
	// Migrations 0001 to 0003 still speak the pre-rename shape (orgs, org_id),
	// so they must run before 0004 renames everything underneath them.
	if err := services.RunMigrations(); err != nil {
		log.Fatalf("migration failed: %v", err)
	}
	// 0002 must precede 0003: 0003 can create a "default" org, which pushes
	// 0002 into its ambiguous multi-org branch.
	if err := services.MigrateSettingsOrg(); err != nil {
		log.Fatalf("settings org migration failed: %v", err)
	}
	if err := services.MigrateMissedOrgScopes(); err != nil {
		log.Fatalf("missed org scope migration failed: %v", err)
	}

	// 0004 renames orgs to instances. It must run BEFORE the index builders:
	// EnsureAuthIndexes creates instances.slug, which would create an empty
	// instances collection and make 0004 refuse to rename onto it.
	migCtx, migCancel := context.WithTimeout(context.Background(), 10*time.Minute)
	migErr := services.MigrateOrgToInstance(migCtx, db.Database)
	migCancel()
	if migErr != nil {
		log.Fatalf("instance rename migration failed: %v", migErr)
	}

	assertCtx, assertCancel := context.WithTimeout(context.Background(), 30*time.Second)
	assertErr := services.AssertNoScopedCollectionMissed(assertCtx, db.Database)
	assertCancel()
	if assertErr != nil {
		log.Fatalf("scoped collection check failed: %v", assertErr)
	}

	if err := services.EnsureAuthIndexes(); err != nil {
		log.Fatalf("failed to ensure auth indexes: %v", err)
	}

	if err := services.EnsureSecretIndexes(); err != nil {
		log.Printf("warning: failed to ensure secret indexes: %v", err)
	}

	if err := services.EnsureSettingsIndexes(); err != nil {
		log.Fatalf("failed to ensure settings indexes: %v", err)
	}

	if err := services.EnsureWorkflowIndexes(); err != nil {
		log.Printf("warning: failed to ensure workflow indexes: %v", err)
	}

	if instanceIDs, err := services.ListInstanceIDs(); err != nil {
		log.Printf("warning: failed to list instances for default step seeding: %v", err)
	} else {
		for _, instanceID := range instanceIDs {
			if created, updated, err := services.SeedDefaultSteps(instanceID); err != nil {
				log.Printf("warning: failed to seed default steps for instance %s: %v", instanceID, err)
			} else {
				log.Printf("default steps seeded for instance %s: %d created, %d updated", instanceID, created, updated)
			}
		}
	}

}

func serve() {
	redisAddr := getEnv("REDIS_ADDR", "localhost:6379")
	redisUser := os.Getenv("REDIS_USERNAME")
	redisPass := os.Getenv("REDIS_PASSWORD")
	if err := auth.InitRedis(redisAddr, redisUser, redisPass); err != nil {
		log.Fatalf("failed to connect to Redis: %v", err)
	}
	log.Println("connected to Redis")

	// The bus carries agent commands and step results between replicas. It is
	// not optional even on a single-replica deployment: dispatch takes the same
	// path either way, so the code exercised in production is the code
	// exercised everywhere.
	if err := bus.Init(redisAddr, redisUser, redisPass); err != nil {
		log.Fatalf("failed to connect the message bus: %v", err)
	}
	log.Printf("message bus ready as node %s", bus.NodeID())

	// Cancelled on SIGTERM/SIGINT. Everything below that takes a context — the
	// housekeeping jobs, the leader lock — stops when the pod is asked to.
	ctx, shutdown := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer shutdown()

	stopGRPC, err := grpcserver.StartGRPC(9090)
	if err != nil {
		log.Fatalf("gRPC server error: %v", err)
	}

	// Everything below runs on exactly one replica at a time.
	//
	// These are cluster-singleton jobs, not per-pod work: N replicas would mean
	// every monitor check firing N times, every incident notification delivered
	// to the customer N times, every retention sweep deleting concurrently, and
	// N reapers racing to purge the same instance. They share one lock rather
	// than holding four, because they are one role — housekeeping — and
	// splitting them would only spread that role across pods for no benefit.
	bus.RunAsLeader(ctx, "housekeeping", func(jobCtx context.Context) {
		services.StartLogSweeper(jobCtx)
		services.StartAuditSweeper(jobCtx)
		services.StartReaper(jobCtx)
		monitorsched.Start(jobCtx)

		ticker := time.NewTicker(2 * time.Minute)
		defer ticker.Stop()
		for {
			select {
			case <-jobCtx.Done():
				return
			case <-ticker.C:
				if err := services.MarkOfflineServers(); err != nil {
					log.Printf("mark offline error: %v", err)
				}
			}
		}
	})

	r := gin.New()
	r.Use(gin.Recovery())
	r.Use(gin.LoggerWithConfig(gin.LoggerConfig{SkipPaths: []string{"/api/console/tunnel"}}))
	r.Use(corsMiddleware())
	api.RegisterRoutes(r)

	srv := &http.Server{Addr: ":8080", Handler: r}
	go func() {
		log.Println("REST server listening on :8080")
		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			log.Fatalf("REST server error: %v", err)
		}
	}()

	<-ctx.Done()
	log.Println("shutdown signal received")

	// gRPC first, and this ordering is the point of the whole exercise. Stopping
	// it runs each CommandStream handler's deferred release, which clears that
	// agent's presence claim; until that happens another replica will keep
	// dispatching commands to this process. Draining HTTP first would leave the
	// claims held for the length of the drain.
	stopGRPC()

	drainCtx, cancelDrain := context.WithTimeout(context.Background(), httpDrainTimeout)
	defer cancelDrain()
	if err := srv.Shutdown(drainCtx); err != nil {
		log.Printf("REST server shutdown: %v", err)
	}
	log.Println("shutdown complete")
}

// How long in-flight REST requests are given to finish. Console tunnels are
// long-lived WebSockets that will not end on their own, so this is a ceiling
// rather than a target; the relays behind them are already gone by this point.
const httpDrainTimeout = 10 * time.Second

func corsMiddleware() gin.HandlerFunc {
	return func(c *gin.Context) {
		c.Header("Access-Control-Allow-Origin", "*")
		c.Header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
		c.Header("Access-Control-Allow-Headers", "Content-Type, Authorization")
		if c.Request.Method == "OPTIONS" {
			c.AbortWithStatus(204)
			return
		}
		c.Next()
	}
}

func getEnv(key, fallback string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return fallback
}

// boolEnv reads a flag env var. Anything other than a recognised truthy value
// is false, so a typo leaves the safe default (migrate here, serve here) rather
// than silently skipping schema setup.
func boolEnv(key string) bool {
	switch strings.ToLower(strings.TrimSpace(os.Getenv(key))) {
	case "1", "true", "yes", "on":
		return true
	}
	return false
}