package tenantmanager

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sync"

	"go.uber.org/zap"

	"github.com/everything-on-mysql/cassandra-on-mysql/internal/config"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/cqlerrors"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/executor"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/mysqlbackend"
)

var ErrMySQLLimitExceeded = errors.New("tenant mysql connection limit exceeded")

type poolManager interface {
	GetPool(user, password string) (poolConnProvider, error)
	Close() error
}

type poolStatsProvider interface {
	GetStats() map[string]*mysqlbackend.PoolStats
}

type poolConnProvider interface {
	Conn(ctx context.Context) (*sql.Conn, error)
}

type poolManagerAdapter struct {
	pm *mysqlbackend.PoolManager
}

func (a *poolManagerAdapter) GetPool(user, password string) (poolConnProvider, error) {
	if a == nil || a.pm == nil {
		return nil, fmt.Errorf("pool manager not configured")
	}
	return a.pm.GetPool(user, password)
}

func (a *poolManagerAdapter) Close() error {
	if a == nil || a.pm == nil {
		return nil
	}
	return a.pm.Close()
}

func (a *poolManagerAdapter) GetStats() map[string]*mysqlbackend.PoolStats {
	if a == nil || a.pm == nil {
		return nil
	}
	return a.pm.GetStats()
}

// TenantManager owns tenant-scoped runtime state.
type TenantManager struct {
	mu      sync.RWMutex
	tenants map[string]*TenantContext
	poolCfg *mysqlbackend.PoolConfig
	limits  config.TenancyLimitConfig
	logger  *zap.Logger
}

// TenantStats aggregates usage across tenants.
type TenantStats struct {
	Tenants    int
	ProxyInUse int
	MySQLInUse int
}

// TenantPoolStats holds MySQL pool stats for a tenant.
type TenantPoolStats struct {
	TenantID string
	Pools    map[string]*mysqlbackend.PoolStats
}

// TenantContext contains per-tenant runtime components.
type TenantContext struct {
	tenantID     string
	poolManager  poolManager
	executor     *executor.Executor
	proxyLimiter *Limiter
	mysqlLimiter *Limiter
}

// Limiter enforces a maximum concurrent count.
type Limiter struct {
	ch chan struct{}
}

func newLimiter(max int) *Limiter {
	if max <= 0 {
		return nil
	}
	return &Limiter{ch: make(chan struct{}, max)}
}

func (l *Limiter) Acquire() bool {
	if l == nil {
		return true
	}
	select {
	case l.ch <- struct{}{}:
		return true
	default:
		return false
	}
}

func (l *Limiter) Release() {
	if l == nil {
		return
	}
	select {
	case <-l.ch:
	default:
	}
}

func (l *Limiter) Count() int {
	if l == nil {
		return 0
	}
	return len(l.ch)
}

// NewTenantManager initializes a tenant manager with the provided configuration.
func NewTenantManager(cfg config.TenancyConfig, poolCfg *mysqlbackend.PoolConfig, logger *zap.Logger) (*TenantManager, error) {
	if poolCfg == nil {
		return nil, fmt.Errorf("pool config is required")
	}
	if logger == nil {
		logger = zap.NewNop()
	}
	return &TenantManager{
		tenants: make(map[string]*TenantContext),
		poolCfg: poolCfg,
		limits:  cfg.Limits,
		logger:  logger,
	}, nil
}

// GetTenant returns the tenant context, creating it if needed.
func (m *TenantManager) GetTenant(tenantID string) (*TenantContext, error) {
	if m == nil {
		return nil, fmt.Errorf("tenant manager is nil")
	}
	if tenantID == "" {
		return nil, fmt.Errorf("tenant ID is required")
	}
	m.mu.RLock()
	if tenant, ok := m.tenants[tenantID]; ok {
		m.mu.RUnlock()
		return tenant, nil
	}
	m.mu.RUnlock()

	m.mu.Lock()
	defer m.mu.Unlock()
	if tenant, ok := m.tenants[tenantID]; ok {
		return tenant, nil
	}

	logger := m.logger.With(zap.String("tenant", tenantID))
	poolMgr := mysqlbackend.NewPoolManager(m.poolCfg, logger)
	exec := executor.NewExecutor(poolMgr, logger)
	exec.Start()
	tenant := &TenantContext{
		tenantID:     tenantID,
		poolManager:  &poolManagerAdapter{pm: poolMgr},
		executor:     exec,
		proxyLimiter: newLimiter(m.limits.MaxProxyConnections),
		mysqlLimiter: newLimiter(m.limits.MaxMySQLConnections),
	}
	m.tenants[tenantID] = tenant
	return tenant, nil
}

// PruneIdleTenants removes tenant contexts without active proxy or MySQL leases.
func (m *TenantManager) PruneIdleTenants() int {
	if m == nil {
		return 0
	}
	var toClose []poolManager
	var executors []*executor.Executor
	m.mu.Lock()
	for tenantID, tenant := range m.tenants {
		if tenant == nil || tenant.isIdle() {
			delete(m.tenants, tenantID)
			if tenant != nil && tenant.poolManager != nil {
				toClose = append(toClose, tenant.poolManager)
			}
			if tenant != nil && tenant.executor != nil {
				executors = append(executors, tenant.executor)
			}
		}
	}
	m.mu.Unlock()

	for _, exec := range executors {
		exec.Close()
	}
	for _, pm := range toClose {
		if err := pm.Close(); err != nil {
			m.logger.Error("failed to close tenant pool manager", zap.Error(err))
		}
	}
	return len(toClose)
}

// Stats returns aggregated limiter usage across tenants.
func (m *TenantManager) Stats() TenantStats {
	if m == nil {
		return TenantStats{}
	}
	stats := TenantStats{}
	m.mu.RLock()
	stats.Tenants = len(m.tenants)
	for _, tenant := range m.tenants {
		if tenant == nil {
			continue
		}
		stats.ProxyInUse += tenant.proxyLimiter.Count()
		stats.MySQLInUse += tenant.mysqlLimiter.Count()
	}
	m.mu.RUnlock()
	return stats
}

// PoolStats returns MySQL pool stats across all tenants.
func (m *TenantManager) PoolStats() []TenantPoolStats {
	if m == nil {
		return nil
	}
	m.mu.RLock()
	tenants := make([]*TenantContext, 0, len(m.tenants))
	for _, tenant := range m.tenants {
		tenants = append(tenants, tenant)
	}
	m.mu.RUnlock()

	out := make([]TenantPoolStats, 0, len(tenants))
	for _, tenant := range tenants {
		if tenant == nil || tenant.poolManager == nil {
			continue
		}
		provider, ok := tenant.poolManager.(poolStatsProvider)
		if !ok {
			continue
		}
		stats := provider.GetStats()
		if len(stats) == 0 {
			continue
		}
		out = append(out, TenantPoolStats{
			TenantID: tenant.tenantID,
			Pools:    stats,
		})
	}
	return out
}

// Close closes all tenant pool managers.
func (m *TenantManager) Close() error {
	if m == nil {
		return nil
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	var firstErr error
	for tenantID, tenant := range m.tenants {
		if tenant == nil {
			continue
		}
		if tenant.executor != nil {
			tenant.executor.Close()
		}
		if tenant.poolManager == nil {
			continue
		}
		if err := tenant.poolManager.Close(); err != nil && firstErr == nil {
			firstErr = err
			m.logger.Error("failed to close tenant pool manager", zap.String("tenant", tenantID), zap.Error(err))
		}
	}
	return firstErr
}

// TenantID returns the tenant identifier.
func (t *TenantContext) TenantID() string {
	if t == nil {
		return ""
	}
	return t.tenantID
}

// Executor returns the tenant-scoped executor.
func (t *TenantContext) Executor() *executor.Executor {
	if t == nil {
		return nil
	}
	return t.executor
}

// AcquireProxySlot reserves a proxy connection slot.
func (t *TenantContext) AcquireProxySlot() bool {
	if t == nil {
		return false
	}
	return t.proxyLimiter.Acquire()
}

// ReleaseProxySlot releases a proxy connection slot.
func (t *TenantContext) ReleaseProxySlot() {
	if t == nil {
		return
	}
	t.proxyLimiter.Release()
}

func (t *TenantContext) isIdle() bool {
	if t == nil {
		return true
	}
	if t.proxyLimiter == nil || t.mysqlLimiter == nil {
		return false
	}
	return t.proxyLimiter.Count() == 0 && t.mysqlLimiter.Count() == 0
}

// AcquireConnLease implements mysqlbackend.ConnLeaseProvider.
func (t *TenantContext) AcquireConnLease(kind mysqlbackend.ConnLeaseKind) (func(), error) {
	if t == nil {
		return nil, fmt.Errorf("tenant context is nil")
	}
	if !t.mysqlLimiter.Acquire() {
		switch kind {
		case mysqlbackend.ConnLeaseKindBulk:
			return nil, cqlerrors.Overloaded("tenant MySQL connection limit exceeded for bulk batch workers")
		default:
			return nil, cqlerrors.Overloaded("tenant MySQL connection limit exceeded")
		}
	}
	var once sync.Once
	release := func() {
		once.Do(func() {
			t.mysqlLimiter.Release()
		})
	}
	return release, nil
}

// LeaseMySQLConn leases a backend connection for the tenant.
func (t *TenantContext) LeaseMySQLConn(ctx context.Context, user, password string) (*sql.Conn, func(), error) {
	if t == nil || t.poolManager == nil {
		return nil, nil, fmt.Errorf("tenant pool manager not configured")
	}
	if !t.mysqlLimiter.Acquire() {
		return nil, nil, ErrMySQLLimitExceeded
	}
	pool, err := t.poolManager.GetPool(user, password)
	if err != nil {
		t.mysqlLimiter.Release()
		return nil, nil, fmt.Errorf("failed to get connection pool: %w", err)
	}
	conn, err := pool.Conn(ctx)
	if err != nil {
		t.mysqlLimiter.Release()
		return nil, nil, fmt.Errorf("failed to get connection: %w", err)
	}

	var once sync.Once
	release := func() {
		once.Do(func() {
			_ = conn.Close()
			t.mysqlLimiter.Release()
		})
	}
	return conn, release, nil
}

// LeaseMySQLSlot reserves a tenant MySQL slot without pinning a connection.
func (t *TenantContext) LeaseMySQLSlot() (func(), error) {
	if t == nil {
		return nil, fmt.Errorf("tenant context is nil")
	}
	if !t.mysqlLimiter.Acquire() {
		return nil, ErrMySQLLimitExceeded
	}
	var once sync.Once
	release := func() {
		once.Do(func() {
			t.mysqlLimiter.Release()
		})
	}
	return release, nil
}