geo/server/internal/bootstrap/bootstrap.go

package bootstrap

import (
	"context"
	"fmt"
	"path/filepath"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/jackc/pgx/v5/pgxpool"
	goredis "github.com/redis/go-redis/v9"
	"go.uber.org/zap"

	"github.com/geo-platform/tenant-api/internal/shared/auditlog"
	"github.com/geo-platform/tenant-api/internal/shared/auth"
	"github.com/geo-platform/tenant-api/internal/shared/cache"
	"github.com/geo-platform/tenant-api/internal/shared/config"
	"github.com/geo-platform/tenant-api/internal/shared/llm"
	"github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq"
	"github.com/geo-platform/tenant-api/internal/shared/middleware"
	"github.com/geo-platform/tenant-api/internal/shared/objectstorage"
	"github.com/geo-platform/tenant-api/internal/shared/observability"
	"github.com/geo-platform/tenant-api/internal/shared/repository/postgres"
	"github.com/geo-platform/tenant-api/internal/shared/repository/redis"
	"github.com/geo-platform/tenant-api/internal/shared/response"
	"github.com/geo-platform/tenant-api/internal/shared/retrieval"
	"github.com/geo-platform/tenant-api/internal/shared/stream"
	tenantapp "github.com/geo-platform/tenant-api/internal/tenant/app"
	"github.com/geo-platform/tenant-api/internal/tenant/prompts"
	"github.com/geo-platform/tenant-api/internal/tenant/repository"
)

type App struct {
	Config             *config.Config
	Logger             *zap.Logger
	DB                 *pgxpool.Pool
	MonitoringDB       *pgxpool.Pool
	RabbitMQ           *rabbitmq.Client
	Redis              *goredis.Client
	Engine             *gin.Engine
	JWT                *auth.Manager
	Sessions           *auth.SessionStore
	LoginGuard         *auth.LoginGuard
	LLM                llm.Client
	RetrievalProvider  retrieval.Provider
	VectorStore        retrieval.VectorStore
	ObjectStorage      objectstorage.Client
	AuditLogs          *auditlog.AsyncWriter
	GenerationStreams  *stream.GenerationHub
	DesktopTaskStreams *stream.DesktopTaskHub
	DesktopDispatch    *stream.DesktopDispatchHub
	Cache              cache.Cache
	KolProfiles        repository.KolProfileRepository
	KolPackages        repository.KolPackageRepository
	KolMarketplace     repository.KolMarketplaceRepository
	KolPrompts         repository.KolPromptRepository
	KolSubscriptions   repository.KolSubscriptionRepository
	KolAssists         repository.KolAssistRepository
	KolPromptAsset     *tenantapp.KolPromptAsset
}

func New(configPath string) (*App, error) {
	cfg, err := config.Load(configPath)
	if err != nil {
		return nil, fmt.Errorf("load config: %w", err)
	}
	prompts.SetConfigFile(filepath.Join(filepath.Dir(configPath), "prompts.yml"))

	logger, err := observability.NewLogger(cfg.Log.Level, cfg.Log.Format)
	if err != nil {
		return nil, fmt.Errorf("init logger: %w", err)
	}

	ctx := context.Background()

	pool, err := postgres.NewPool(ctx, cfg.Database)
	if err != nil {
		return nil, fmt.Errorf("init postgres: %w", err)
	}

	if err := syncMembershipPlans(ctx, pool, cfg.Membership); err != nil {
		pool.Close()
		return nil, fmt.Errorf("sync membership plans: %w", err)
	}

	monitoringPool, err := postgres.NewPool(ctx, cfg.MonitoringDatabase)
	if err != nil {
		pool.Close()
		return nil, fmt.Errorf("init monitoring postgres: %w", err)
	}

	rdb, err := redis.NewClient(ctx, cfg.Redis)
	if err != nil {
		logger.Warn("redis unavailable during startup; continuing with degraded auth/cache fallback", zap.Error(err))
		rdb = redis.NewLazyClient(cfg.Redis)
	}

	mqClient, err := rabbitmq.New(ctx, cfg.RabbitMQ)
	if err != nil {
		pool.Close()
		monitoringPool.Close()
		_ = rdb.Close()
		return nil, fmt.Errorf("init rabbitmq: %w", err)
	}

	jwtMgr := auth.NewManager(cfg.JWT.Secret, cfg.JWT.AccessTTL, cfg.JWT.RefreshTTL)
	sessions := auth.NewSessionStore(rdb)
	loginGuard := auth.NewLoginGuard(rdb, auth.DefaultLoginGuardConfig("tenant"))
	llmClient := llm.New(cfg.LLM)
	retrievalProvider := retrieval.NewProvider(cfg.Retrieval, logger)
	vectorStore := retrieval.NewQdrantStore(cfg.Qdrant, logger)
	objectStorageClient := objectstorage.New(cfg.ObjectStorage, logger)
	auditLogs := auditlog.NewAsyncWriter(pool, logger)
	generationStreams := stream.NewGenerationHub(mqClient)
	desktopTaskStreams := stream.NewDesktopTaskHub(mqClient)
	desktopDispatchBindings := []string{
		cfg.RabbitMQ.DesktopDispatchPublishPrefix + ".*",
		cfg.RabbitMQ.DesktopDispatchMonitorPrefix + ".*",
		cfg.RabbitMQ.DesktopDispatchMonitorPrefix + ".*.*",
	}
	desktopDispatch := stream.NewDesktopDispatchHub(mqClient, logger, desktopDispatchBindings)
	appCache := cache.New(cfg.Cache.Driver, rdb)
	kolProfiles := repository.NewKolProfileRepository(pool)
	kolPackages := repository.NewKolPackageRepository(pool)
	kolMarketplace := repository.NewKolMarketplaceRepository(pool)
	kolPrompts := repository.NewKolPromptRepository(pool)
	kolSubscriptions := repository.NewKolSubscriptionRepository(pool)
	kolAssists := repository.NewKolAssistRepository(pool)
	kolPromptAsset := tenantapp.NewKolPromptAsset(objectStorageClient)

	if cfg.Server.Mode == "release" {
		gin.SetMode(gin.ReleaseMode)
	}
	engine := gin.New()

	engine.Use(
		middleware.Recovery(logger),
		middleware.RequestID(),
		middleware.Logger(logger),
		middleware.CORS(),
	)

	engine.GET("/api/health/live", func(c *gin.Context) {
		response.Success(c, gin.H{"status": "alive"})
	})
	engine.GET("/api/health/ready", func(c *gin.Context) {
		if err := pool.Ping(c.Request.Context()); err != nil {
			response.Error(c, response.ErrServiceUnavailable(50301, "postgres_unavailable", "database is not reachable"))
			return
		}
		if err := monitoringPool.Ping(c.Request.Context()); err != nil {
			response.Error(c, response.ErrServiceUnavailable(50303, "monitoring_postgres_unavailable", "monitoring database is not reachable"))
			return
		}
		if err := rdb.Ping(c.Request.Context()).Err(); err != nil {
			response.Error(c, response.ErrServiceUnavailable(50302, "redis_unavailable", "redis is not reachable"))
			return
		}
		if err := mqClient.Ping(c.Request.Context()); err != nil {
			response.Error(c, response.ErrServiceUnavailable(50304, "rabbitmq_unavailable", "rabbitmq is not reachable"))
			return
		}
		response.Success(c, gin.H{"status": "ready"})
	})

	return &App{
		Config:             cfg,
		Logger:             logger,
		DB:                 pool,
		MonitoringDB:       monitoringPool,
		RabbitMQ:           mqClient,
		Redis:              rdb,
		Engine:             engine,
		JWT:                jwtMgr,
		Sessions:           sessions,
		LoginGuard:         loginGuard,
		LLM:                llmClient,
		RetrievalProvider:  retrievalProvider,
		VectorStore:        vectorStore,
		ObjectStorage:      objectStorageClient,
		AuditLogs:          auditLogs,
		GenerationStreams:  generationStreams,
		DesktopTaskStreams: desktopTaskStreams,
		DesktopDispatch:    desktopDispatch,
		Cache:              appCache,
		KolProfiles:        kolProfiles,
		KolPackages:        kolPackages,
		KolMarketplace:     kolMarketplace,
		KolPrompts:         kolPrompts,
		KolSubscriptions:   kolSubscriptions,
		KolAssists:         kolAssists,
		KolPromptAsset:     kolPromptAsset,
	}, nil
}

// syncMembershipPlans upserts all configured plans in a single transaction so
// that a partial failure (or concurrent startup) never leaves the `plans` table
// in a half-synced state.
func syncMembershipPlans(ctx context.Context, pool *pgxpool.Pool, membership config.MembershipConfig) error {
	tx, err := pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() {
		_ = tx.Rollback(ctx)
	}()

	if err := repository.NewTenantPlanRepository(tx).UpsertConfiguredPlans(ctx, membership); err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (a *App) Close() {
	if a.AuditLogs != nil {
		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		if err := a.AuditLogs.Close(ctx); err != nil {
			a.Logger.Warn("shutdown with pending audit logs", zap.Error(err))
		}
		cancel()
	}
	a.DB.Close()
	if a.MonitoringDB != nil {
		a.MonitoringDB.Close()
	}
	if a.RabbitMQ != nil {
		_ = a.RabbitMQ.Close()
	}
	_ = a.Redis.Close()
	_ = a.Logger.Sync()
}