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"
	"go.mongodb.org/mongo-driver/mongo"
	"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
	resourceMu    sync.RWMutex
	databaseReady atomic.Bool
	gormDB        *reloadableDB
	redis         *reloadableRedis
	redisListMu   sync.RWMutex
	redisList     map[string]redis.UniversalClient
	mongo         *reloadableMongo
	runtime       *config.Store
	integrations  *runtimeconfig.Store
	storage       *storage.Reloadable
	dbListMu      sync.RWMutex
	dbList        map[string]*gorm.DB
	// Handles replaced by a hot reload of the named lists. They follow the same
	// grace period as the primary pool so in-flight queries are not cut off.
	retiredDBList    retiredSet[*gorm.DB]
	retiredRedisList retiredSet[redis.UniversalClient]
	appLogger        *slog.Logger
	auditLog         *dataScopeAuditWriter
	catalog          module.Catalog
	resourceHook     func()
}

// 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) {
	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 (d *Data) RedisClient() redis.UniversalClient {
	if d == nil || d.redis == nil {
		return nil
	}
	return d.redis.load()
}

// MongoClient returns the currently active Mongo client. The client remains
// owned by Data and is only exposed here for framework-level resource wiring.
func (d *Data) MongoClient() *mongo.Client {
	if d == nil || d.mongo == nil {
		return nil
	}
	return d.mongo.load()
}

// NamedDatabases returns a copy of configured secondary database handles for
// the process-wide resource registry. Data retains lifecycle ownership.
func (d *Data) NamedDatabases() map[string]*gorm.DB {
	if d == nil {
		return nil
	}
	d.dbListMu.RLock()
	defer d.dbListMu.RUnlock()
	if len(d.dbList) == 0 {
		return nil
	}
	result := make(map[string]*gorm.DB, len(d.dbList))
	for name, db := range d.dbList {
		if name != "" && db != nil {
			result[name] = db
		}
	}
	return result
}

// NamedRedisClients returns a copy of configured secondary Redis handles for
// the process-wide resource registry. Data retains lifecycle ownership.
func (d *Data) NamedRedisClients() map[string]redis.UniversalClient {
	if d == nil {
		return nil
	}
	d.redisListMu.RLock()
	defer d.redisListMu.RUnlock()
	if len(d.redisList) == 0 {
		return nil
	}
	result := make(map[string]redis.UniversalClient, len(d.redisList))
	for name, client := range d.redisList {
		if name != "" && client != nil {
			result[name] = client
		}
	}
	return result
}

// SetResourceHook registers a composition-root callback invoked after active
// database, Redis, or Mongo handles change. A callback keeps data independent
// from the global package while allowing the root to refresh shared handles.
func (d *Data) SetResourceHook(hook func()) {
	if d == nil {
		return
	}
	d.resourceMu.Lock()
	d.resourceHook = hook
	d.resourceMu.Unlock()
}

func (d *Data) notifyResources() {
	if d == nil {
		return
	}
	d.resourceMu.RLock()
	hook := d.resourceHook
	d.resourceMu.RUnlock()
	if hook != nil {
		hook()
	}
}

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 db != nil {
			closeGormDB(db)
		}
	}
}

// replaceDatabaseList swaps the named database handles. Superseded handles are
// retired instead of closed on the spot: a request that already resolved a
// handle would otherwise fail with "sql: database is closed" mid-flight.
func (d *Data) replaceDatabaseList(items map[string]*gorm.DB) {
	d.dbListMu.Lock()
	old := d.dbList
	d.dbList = items
	d.dbListMu.Unlock()
	for name, db := range old {
		if db == nil || db == items[name] {
			continue
		}
		d.retiredDBList.retire(db, closeGormDB)
	}
}

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 cleanupOnce sync.Once
	cleanup := func() {
		cleanupOnce.Do(func() {
			if d.auditLog != nil {
				d.auditLog.Close()
			}
			if d.gormDB != nil {
				d.gormDB.close()
			}
			closeDatabaseList(d.dbList)
			for _, db := range d.retiredDBList.drain() {
				closeGormDB(db)
			}
			if d.redis != nil {
				d.redis.close()
			}
			closeRedisList(d.redisList)
			for _, client := range d.retiredRedisList.drain() {
				closeRedisClient(client)
			}
			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))
	useRedisList := useRedis && admin.System.UseMultipoint
	d.redisList = openRedisList(c.RedisList, useRedisList, 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)
	initialized = true
	return d, cleanup, nil
}

// redisConnectionConfigured reports whether a Redis block names an endpoint.
// The reload path shares it with openRedis so "no endpoint configured" is never
// confused with "configured endpoint failed to answer".
func redisConnectionConfigured(config *config.Redis) bool {
	return config != nil && (config.Addr != "" || len(config.ClusterAddrs) > 0)
}

func openRedis(config *config.Redis, enabled bool, appLogger *slog.Logger) redis.UniversalClient {
	if !enabled || !redisConnectionConfigured(config) {
		return nil
	}
	if appLogger == nil {
		appLogger = slog.Default()
	}
	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 {
		appLogger.Warn("redis unavailable, using in-memory cache", "mod", "redis", "error", err)
		_ = candidate.Close()
		return nil
	}
	return candidate
}

func openRedisList(configs []*config.Redis, enabled bool, appLogger *slog.Logger) map[string]redis.UniversalClient {
	if !enabled || len(configs) == 0 {
		return nil
	}
	clients := make(map[string]redis.UniversalClient)
	for _, item := range configs {
		if item == nil || item.Name == "" {
			continue
		}
		if client := openRedis(item, true, appLogger); client != nil {
			clients[item.Name] = client
		}
	}
	if len(clients) == 0 {
		return nil
	}
	return clients
}

func closeRedisList(clients map[string]redis.UniversalClient) {
	for _, client := range clients {
		if client != nil {
			closeRedisClient(client)
		}
	}
}

// replaceRedisList mirrors replaceDatabaseList: superseded clients stay open for
// the retire grace period so in-flight commands are not aborted.
func (d *Data) replaceRedisList(clients map[string]redis.UniversalClient) {
	if d == nil {
		return
	}
	d.redisListMu.Lock()
	old := d.redisList
	d.redisList = clients
	d.redisListMu.Unlock()
	for name, client := range old {
		if client == nil || client == clients[name] {
			continue
		}
		d.retiredRedisList.retire(client, closeRedisClient)
	}
}

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