kra-new/internal/data/data.go

package data

import (
	"context"
	"fmt"
	"log/slog"
	"sync"
	"sync/atomic"
	"time"

	"github.com/google/wire"
	"github.com/redis/go-redis/v9"
	"gorm.io/gorm"
	"kra/internal/config"
	dataintegration "kra/internal/data/integration"
	datapayment "kra/internal/data/payment"
	datasystem "kra/internal/data/system"
	datatask "kra/internal/data/task"
	"kra/internal/integration/runtimeconfig"
	"kra/internal/integration/storage"
	"kra/pkg/module"
)

var ProviderSet = wire.NewSet(
	NewData,
	NewIntegrationRuntime,
	wire.Bind(new(datasystem.Provider), new(*Data)),
	wire.Bind(new(datasystem.DatabaseProvider), new(*Data)),
	wire.Bind(new(dataintegration.Provider), new(*Data)),
	wire.Bind(new(datatask.Provider), new(*Data)),
	wire.Bind(new(datapayment.Provider), new(*Data)),
	datasystem.ProviderSet,
	dataintegration.ProviderSet,
	datatask.ProviderSet,
	datapayment.ProviderSet,
)

func NewIntegrationRuntime(data *Data) *runtimeconfig.Store {
	if data == nil {
		return runtimeconfig.NewStore()
	}
	return data.IntegrationRuntime()
}

type Data struct {
	initMu        sync.Mutex
	configMu      sync.Mutex
	databaseReady atomic.Bool
	gormDB        *reloadableDB
	redis         *reloadableRedis
	mongo         *reloadableMongo
	runtime       *config.Store
	integrations  *runtimeconfig.Store
	storage       *storage.Reloadable
	dbListMu      sync.RWMutex
	dbList        map[string]*gorm.DB
	appLogger     *slog.Logger
	auditLog      *dataScopeAuditWriter
	catalog       module.Catalog
}

// DB exposes the active primary database to narrowly scoped data submodules.
func (d *Data) DB() *gorm.DB {
	if d == nil || d.gormDB == nil {
		return nil
	}
	return d.gormDB.DB()
}

// DatabaseReady reports whether the configured primary database has been
// initialized. It is intentionally small so system repositories do not depend
// on the full Data implementation.
func (d *Data) DatabaseReady() bool {
	return d != nil && d.databaseReady.Load()
}

// Runtime exposes the immutable runtime configuration snapshot to data
// submodules that need system settings while keeping Data itself private.
func (d *Data) Runtime() *config.Store {
	if d == nil {
		return nil
	}
	return d.runtime
}

// IntegrationRuntime exposes database-backed integration configuration to
// long-lived adapters without making config.yaml part of their lifecycle.
func (d *Data) IntegrationRuntime() *runtimeconfig.Store {
	if d == nil {
		return nil
	}
	return d.integrations
}

// Database resolves the primary or a named database for repositories such as
// the system export module.
func (d *Data) Database(name string) (*gorm.DB, error) {
	return d.database(name)
}

func (d *Data) RedisClient() redis.UniversalClient {
	if d == nil || d.redis == nil {
		return nil
	}
	return d.redis.load()
}

func (d *Data) logger() *slog.Logger {
	if d != nil && d.appLogger != nil {
		return d.appLogger
	}
	return slog.Default()
}

func openDatabaseList(configs []*config.Database, appLogger ...*slog.Logger) (map[string]*gorm.DB, error) {
	items := make(map[string]*gorm.DB)
	for _, config := range configs {
		if config == nil || config.Disable || config.AliasName == "" {
			continue
		}
		db, err := openDatabase(config, false, "", appLogger...)
		if err != nil {
			for _, opened := range items {
				if sqlDB, dbErr := opened.DB(); dbErr == nil {
					_ = sqlDB.Close()
				}
			}
			return nil, fmt.Errorf("open database %q: %w", config.AliasName, err)
		}
		items[config.AliasName] = db
	}
	return items, nil
}

func closeDatabaseList(items map[string]*gorm.DB) {
	for _, db := range items {
		if sqlDB, err := db.DB(); err == nil {
			_ = sqlDB.Close()
		}
	}
}

func (d *Data) replaceDatabaseList(items map[string]*gorm.DB) {
	d.dbListMu.Lock()
	old := d.dbList
	d.dbList = items
	d.dbListMu.Unlock()
	closeDatabaseList(old)
}

func (d *Data) database(name string) (*gorm.DB, error) {
	if name == "" {
		return d.gormDB.DB(), nil
	}
	d.dbListMu.RLock()
	db := d.dbList[name]
	d.dbListMu.RUnlock()
	if db == nil {
		return nil, fmt.Errorf("database %q not found", name)
	}
	return db, nil
}

func NewData(runtime *config.Store, appLogger *slog.Logger, storageManager *storage.Reloadable, catalog module.Catalog) (*Data, func(), error) {
	if appLogger == nil {
		appLogger = slog.Default()
	}
	c := runtime.Data()
	if c == nil {
		c = &config.Data{}
	}
	if c.Database == nil {
		// An empty database block starts the service on the bootstrap database
		// so /init/checkdb
		// and /init/initdb remain available.
		c.Database = &config.Database{}
	}
	d := &Data{runtime: runtime, integrations: runtimeconfig.NewStore(), appLogger: appLogger, storage: storageManager, catalog: catalog}
	var stopConfigWatcher func()
	var cleanupOnce sync.Once
	cleanup := func() {
		cleanupOnce.Do(func() {
			if stopConfigWatcher != nil {
				stopConfigWatcher()
			}
			if d.auditLog != nil {
				d.auditLog.Close()
			}
			if d.gormDB != nil {
				d.gormDB.close()
			}
			closeDatabaseList(d.dbList)
			if d.redis != nil {
				d.redis.close()
			}
			if d.mongo != nil {
				d.mongo.close()
			}
		})
	}
	initialized := false
	defer func() {
		if !initialized {
			cleanup()
		}
	}()
	usingFallback := !databaseConnectionConfigured(c.Database)
	var db *gorm.DB
	var err error
	if !usingFallback {
		db, err = openDatabase(c.Database, false, "", appLogger)
	}
	if usingFallback || err != nil {
		// The initialization endpoint must remain available when the configured
		// target database has not been created yet.
		if err != nil {
			appLogger.Warn("configured database unavailable before initialization", "mod", "system", "error", err)
		}
		db, err = openFallbackDatabase(appLogger)
		if err != nil {
			return nil, nil, fmt.Errorf("open bootstrap database: %w", err)
		}
		usingFallback = true
	}
	d.databaseReady.Store(!usingFallback)
	d.gormDB = newReloadableDB(db, d.enqueueDataScopeAudit)
	d.auditLog = newDataScopeAuditWriter(d, appLogger)
	d.dbList, err = openDatabaseList(c.DatabaseList, appLogger)
	if err != nil {
		return nil, nil, err
	}
	for _, item := range d.dbList {
		registerDataScopeCallbacks(item, d.enqueueDataScopeAudit)
	}
	admin := runtime.Admin()
	if admin == nil {
		admin = &config.Admin{}
	}
	disableAutoMigrate := admin.System != nil && admin.System.DisableAutoMigrate
	if !usingFallback && !disableAutoMigrate {
		if err = migrateAll(db, catalog); err != nil {
			return nil, nil, fmt.Errorf("migrate tables: %w", err)
		}
	}
	if !usingFallback {
		storageConfig, storageErr := resolveStorageIntegrationConfig(db, admin.Storage)
		if storageErr != nil {
			return nil, nil, fmt.Errorf("load storage integration configuration: %w", storageErr)
		}
		emailConfig, emailErr := resolveEmailIntegrationConfig(db, admin.Email)
		if emailErr != nil {
			return nil, nil, fmt.Errorf("load email integration configuration: %w", emailErr)
		}
		admin.Storage = storageConfig
		admin.Email = emailConfig
		runtime.Replace(&config.Config{Server: runtime.Server(), Data: c, Admin: admin})
		if err = d.loadIntegrationRuntime(db); err != nil {
			return nil, nil, fmt.Errorf("load integration runtime: %w", err)
		}
		activeStorage, storageErr := storage.New(admin)
		if storageErr != nil {
			return nil, nil, fmt.Errorf("initialize storage: %w", storageErr)
		}
		if storageManager != nil {
			storageManager.Replace(activeStorage)
		}
		if db.Migrator().HasTable("sys_integration_configs") {
			if removeErr := d.removeIntegrationConfigFromFile(); removeErr != nil {
				appLogger.Warn("remove legacy integration configuration from file", "mod", "integration", "error", removeErr)
			}
		}
	}
	if storageManager == nil {
		return nil, nil, fmt.Errorf("storage manager is nil")
	}
	useRedis := admin != nil && admin.System != nil && admin.System.UseRedis
	d.redis = newReloadableRedis(openRedis(c.Redis, useRedis, appLogger))
	useMongo := admin != nil && admin.System != nil && admin.System.UseMongo
	mongoClient, err := openMongo(c.Mongo, useMongo)
	if err != nil {
		appLogger.Error("mongo unavailable", "mod", "mongo", "error", err)
		mongoClient = nil
	}
	d.mongo = newReloadableMongo(mongoClient)
	stopConfigWatcher = d.watchConfig()
	initialized = true
	return d, cleanup, nil
}

func openRedis(config *config.Redis, enabled bool, appLogger ...*slog.Logger) redis.UniversalClient {
	if !enabled || config == nil || (config.Addr == "" && len(config.ClusterAddrs) == 0) {
		return nil
	}
	var candidate redis.UniversalClient
	if config.UseCluster {
		addresses := config.ClusterAddrs
		if len(addresses) == 0 && config.Addr != "" {
			addresses = []string{config.Addr}
		}
		candidate = redis.NewClusterClient(&redis.ClusterOptions{Addrs: addresses, Password: config.Password})
	} else {
		options := &redis.Options{Addr: config.Addr, Network: config.Network, Password: config.Password, DB: int(config.DB)}
		if config.ReadTimeout > 0 {
			options.ReadTimeout = config.ReadTimeout
		}
		if config.WriteTimeout > 0 {
			options.WriteTimeout = config.WriteTimeout
		}
		candidate = redis.NewClient(options)
	}
	pingCtx, cancel := context.WithTimeout(context.Background(), 800*time.Millisecond)
	defer cancel()
	if err := candidate.Ping(pingCtx).Err(); err != nil {
		log := slog.Default()
		if len(appLogger) > 0 && appLogger[0] != nil {
			log = appLogger[0]
		}
		log.Warn("redis unavailable, using in-memory cache", "mod", "redis", "error", err)
		_ = candidate.Close()
		return nil
	}
	return candidate
}

func (d *Data) activateDatabase(db *gorm.DB, config *config.Database) {
	d.gormDB.replace(db, d.enqueueDataScopeAudit)
	d.runtime.UpdateDatabase(config)
	d.databaseReady.Store(true)
}