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 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) 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, 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() }