366 lines
14 KiB
Go
366 lines
14 KiB
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"
|
|
sharedbootstrap "github.com/geo-platform/tenant-api/internal/shared/bootstrap"
|
|
"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/ipregion"
|
|
"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 {
|
|
ConfigStore *config.Store
|
|
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
|
|
PasswordCipher *auth.PasswordCipher
|
|
LLM llm.Client
|
|
ReloadableLLM *llm.ReloadableClient
|
|
RetrievalProvider retrieval.Provider
|
|
ReloadableRetrieval *retrieval.ReloadableProvider
|
|
VectorStore retrieval.VectorStore
|
|
ReloadableVectorStore *retrieval.ReloadableVectorStore
|
|
ObjectStorage objectstorage.Client
|
|
ReloadableObjectStorage *objectstorage.ReloadableClient
|
|
AuditLogs *auditlog.AsyncWriter
|
|
GenerationStreams *stream.GenerationHub
|
|
DesktopTaskStreams *stream.DesktopTaskHub
|
|
DesktopDispatch *stream.DesktopDispatchHub
|
|
Cache cache.Cache
|
|
IPRegions *ipregion.Resolver
|
|
BrandService *tenantapp.BrandService
|
|
QuestionExpansion *tenantapp.QuestionExpansionService
|
|
MonitoringService *tenantapp.MonitoringService
|
|
MediaSupplyService *tenantapp.MediaSupplyService
|
|
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) {
|
|
configStore, err := config.NewStore(configPath)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("load config: %w", err)
|
|
}
|
|
cfg := configStore.Current()
|
|
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)
|
|
passwordCipher, err := buildPasswordCipher(cfg.Auth.PasswordCipher)
|
|
if err != nil {
|
|
pool.Close()
|
|
monitoringPool.Close()
|
|
_ = rdb.Close()
|
|
_ = mqClient.Close()
|
|
return nil, fmt.Errorf("init password cipher: %w", err)
|
|
}
|
|
if passwordCipher == nil {
|
|
logger.Warn("auth password cipher disabled; login request payloads may expose plaintext passwords in browser developer tools")
|
|
}
|
|
loginGuardCfg := auth.DefaultLoginGuardConfig("tenant")
|
|
loginGuardCfg.Enabled = cfg.Auth.LoginGuard.Enabled
|
|
if !loginGuardCfg.Enabled {
|
|
logger.Warn("tenant login guard disabled by config; brute-force / rate-limit protection is off")
|
|
}
|
|
loginGuard := auth.NewLoginGuard(rdb, loginGuardCfg)
|
|
llmClient := llm.NewReloadableClient(llm.New(cfg.LLM))
|
|
retrievalProvider := retrieval.NewReloadableProvider(retrieval.NewProvider(cfg.Retrieval, logger))
|
|
vectorStore := retrieval.NewReloadableVectorStore(retrieval.NewQdrantStore(cfg.Qdrant, logger))
|
|
objectStorageClient := objectstorage.NewReloadableClient(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.NewWithOptions(cfg.Cache.Driver, rdb, cache.Options{
|
|
Namespace: cfg.Cache.Namespace,
|
|
JitterRatio: cfg.Cache.JitterRatio,
|
|
MetricsEnabled: cfg.Cache.MetricsEnabled,
|
|
AsyncFillEnabled: cfg.Cache.AsyncFillEnabled,
|
|
AsyncFillWorkers: cfg.Cache.AsyncFillWorkers,
|
|
AsyncFillBuffer: cfg.Cache.AsyncFillBuffer,
|
|
AsyncFillTimeout: cfg.Cache.AsyncFillTimeout,
|
|
L1Enabled: cfg.Cache.L1Enabled,
|
|
L1TTL: cfg.Cache.L1TTL,
|
|
DeleteScanCount: cfg.Cache.DeleteScanCount,
|
|
})
|
|
ipRegions, err := ipregion.NewResolver(cfg.IPRegion.V4XDBPath, cfg.IPRegion.V6XDBPath, logger)
|
|
if err != nil {
|
|
logger.Warn("ip region resolver unavailable; location-aware defaults will be degraded", zap.Error(err))
|
|
}
|
|
brandService := tenantapp.NewBrandService(pool, monitoringPool, auditLogs, cfg.BrandLibrary).WithCache(appCache)
|
|
questionExpansion := tenantapp.NewQuestionExpansionService(pool, llmClient, brandService).
|
|
WithCache(appCache).
|
|
WithIPRegionResolver(ipRegions)
|
|
monitoringService := tenantapp.NewMonitoringService(pool, monitoringPool, mqClient, cfg.MonitoringDispatch, cfg.BrandLibrary, logger).WithRedis(rdb)
|
|
mediaSupplyService := tenantapp.NewMediaSupplyService(pool, rdb, logger, configStore).WithObjectStorage(objectStorageClient)
|
|
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)
|
|
|
|
engine, err := sharedbootstrap.NewEngine(cfg.Server)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("init http engine: %w", err)
|
|
}
|
|
|
|
engine.Use(
|
|
middleware.Recovery(logger),
|
|
middleware.RequestID(),
|
|
middleware.Logger(logger),
|
|
)
|
|
if cfg.Server.SecurityHeaders.Enabled {
|
|
engine.Use(middleware.SecurityHeaders())
|
|
}
|
|
engine.Use(middleware.CORSWithConfig(middleware.CORSConfig{
|
|
AllowedOrigins: cfg.Server.AllowedOrigins,
|
|
AllowAll: len(cfg.Server.AllowedOrigins) == 0,
|
|
}))
|
|
|
|
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{
|
|
ConfigStore: configStore,
|
|
Logger: logger,
|
|
DB: pool,
|
|
MonitoringDB: monitoringPool,
|
|
RabbitMQ: mqClient,
|
|
Redis: rdb,
|
|
Engine: engine,
|
|
JWT: jwtMgr,
|
|
Sessions: sessions,
|
|
LoginGuard: loginGuard,
|
|
PasswordCipher: passwordCipher,
|
|
LLM: llmClient,
|
|
ReloadableLLM: llmClient,
|
|
RetrievalProvider: retrievalProvider,
|
|
ReloadableRetrieval: retrievalProvider,
|
|
VectorStore: vectorStore,
|
|
ReloadableVectorStore: vectorStore,
|
|
ObjectStorage: objectStorageClient,
|
|
ReloadableObjectStorage: objectStorageClient,
|
|
AuditLogs: auditLogs,
|
|
GenerationStreams: generationStreams,
|
|
DesktopTaskStreams: desktopTaskStreams,
|
|
DesktopDispatch: desktopDispatch,
|
|
Cache: appCache,
|
|
IPRegions: ipRegions,
|
|
BrandService: brandService,
|
|
QuestionExpansion: questionExpansion,
|
|
MonitoringService: monitoringService,
|
|
MediaSupplyService: mediaSupplyService,
|
|
KolProfiles: kolProfiles,
|
|
KolPackages: kolPackages,
|
|
KolMarketplace: kolMarketplace,
|
|
KolPrompts: kolPrompts,
|
|
KolSubscriptions: kolSubscriptions,
|
|
KolAssists: kolAssists,
|
|
KolPromptAsset: kolPromptAsset,
|
|
}, nil
|
|
}
|
|
|
|
func buildPasswordCipher(cfg config.PasswordCipherConfig) (*auth.PasswordCipher, error) {
|
|
if !cfg.Enabled {
|
|
return nil, nil
|
|
}
|
|
if cfg.PrivateKeyPEM == "" {
|
|
return auth.NewEphemeralPasswordCipher(cfg.KeyID)
|
|
}
|
|
return auth.NewPasswordCipher(cfg.PrivateKeyPEM, cfg.KeyID)
|
|
}
|
|
|
|
func (a *App) Config() *config.Config {
|
|
if a == nil || a.ConfigStore == nil {
|
|
return nil
|
|
}
|
|
return a.ConfigStore.Current()
|
|
}
|
|
|
|
func (a *App) StartConfigWatcher(ctx context.Context) {
|
|
if a == nil || a.ConfigStore == nil {
|
|
return
|
|
}
|
|
go func() {
|
|
if err := a.ConfigStore.Watch(ctx, a.Logger, a.applyReloadedConfig); err != nil && a.Logger != nil {
|
|
a.Logger.Warn("config hot reload watcher stopped", zap.Error(err))
|
|
}
|
|
}()
|
|
}
|
|
|
|
func (a *App) applyReloadedConfig(event config.ReloadEvent) {
|
|
if a == nil || event.Current == nil {
|
|
return
|
|
}
|
|
|
|
a.JWT.Update(event.Current.JWT.Secret, event.Current.JWT.AccessTTL, event.Current.JWT.RefreshTTL)
|
|
if a.ReloadableLLM != nil {
|
|
a.ReloadableLLM.Update(llm.New(event.Current.LLM))
|
|
}
|
|
if a.ReloadableRetrieval != nil {
|
|
a.ReloadableRetrieval.Update(retrieval.NewProvider(event.Current.Retrieval, a.Logger))
|
|
}
|
|
if a.ReloadableVectorStore != nil {
|
|
a.ReloadableVectorStore.Update(retrieval.NewQdrantStore(event.Current.Qdrant, a.Logger))
|
|
}
|
|
if a.ReloadableObjectStorage != nil {
|
|
a.ReloadableObjectStorage.Update(objectstorage.New(event.Current.ObjectStorage, a.Logger))
|
|
}
|
|
if a.BrandService != nil {
|
|
a.BrandService.UpdateConfig(event.Current.BrandLibrary)
|
|
}
|
|
if a.MonitoringService != nil {
|
|
a.MonitoringService.UpdateConfig(event.Current.MonitoringDispatch, event.Current.BrandLibrary)
|
|
}
|
|
|
|
if err := syncMembershipPlans(context.Background(), a.DB, event.Current.Membership); err != nil {
|
|
a.Logger.Warn("config reload membership plan sync failed", zap.Error(err))
|
|
}
|
|
|
|
if config.RestartRequired(event.Changes) && a.Logger != nil {
|
|
a.Logger.Warn("config reload changed connection or listener settings; restart process to apply those fields")
|
|
}
|
|
}
|
|
|
|
// 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.ConfigStore != nil {
|
|
a.ConfigStore.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()
|
|
}
|
|
if a.IPRegions != nil {
|
|
a.IPRegions.Close()
|
|
}
|
|
a.DB.Close()
|
|
if a.MonitoringDB != nil {
|
|
a.MonitoringDB.Close()
|
|
}
|
|
if a.RabbitMQ != nil {
|
|
_ = a.RabbitMQ.Close()
|
|
}
|
|
_ = a.Redis.Close()
|
|
_ = a.Logger.Sync()
|
|
}
|