package config

import (
	"context"
	"crypto/sha256"
	"errors"
	"fmt"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/tidbcloud/service-proxy/pkg/audit"
)

const (
	// closeTimeout is the maximum time to wait for Run() to finish during Close().
	closeTimeout = 5 * time.Second

	// defaultPollInterval is the default interval for polling config file changes.
	defaultPollInterval = 5 * time.Second
)

// Watcher monitors configuration file changes and notifies handlers.
// It uses polling + content hash to detect changes, which works for both
// regular files and Kubernetes ConfigMap symlink-based updates.
type Watcher struct {
	manager *Manager

	auditHandlers []AuditConfigChangeHandler
	mu            sync.RWMutex

	lastAuditConfig *audit.Config

	pollInterval time.Duration

	// lastConfigHash is the sha256 of the last seen config file content.
	// It's updated even when reload/apply fails, so we only retry when content changes.
	lastConfigHash [sha256.Size]byte
	hasConfigHash  bool

	lastReadErrMsg string

	// done signals that Run() has exited.
	done chan struct{}
}

// NewWatcher creates a new configuration watcher.
func NewWatcher(manager *Manager) (*Watcher, error) {
	return newWatcher(manager, defaultPollInterval)
}

func newWatcher(manager *Manager, pollInterval time.Duration) (*Watcher, error) {
	if pollInterval <= 0 {
		return nil, errors.New("poll interval must be positive")
	}

	// Initialize lastAuditConfig with current config to avoid false positive
	// on the first poll tick.
	// Ensure manager has loaded config before creating watcher.
	cfg := manager.Get()
	if cfg == nil {
		return nil, errors.New("config manager not initialized: call manager.Load() first")
	}

	data, err := os.ReadFile(manager.GetFilePath())
	if err != nil {
		return nil, fmt.Errorf("read config: %w", err)
	}

	w := &Watcher{
		manager:         manager,
		lastAuditConfig: copyAuditConfig(&cfg.Audit),
		pollInterval:    pollInterval,
		lastConfigHash:  sha256.Sum256(data),
		hasConfigHash:   true,
		done:            make(chan struct{}),
	}

	return w, nil
}

// RegisterAuditHandler registers a handler for audit config changes.
func (w *Watcher) RegisterAuditHandler(handler AuditConfigChangeHandler) {
	w.mu.Lock()
	defer w.mu.Unlock()
	w.auditHandlers = append(w.auditHandlers, handler)
}

// Run starts watching for configuration changes.
// It blocks until the context is cancelled.
func (w *Watcher) Run(ctx context.Context) error {
	defer close(w.done)

	ticker := time.NewTicker(w.pollInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			w.checkAndReload()
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}

func (w *Watcher) checkAndReload() {
	data, err := os.ReadFile(w.manager.GetFilePath())
	if err != nil {
		if w.lastReadErrMsg != err.Error() {
			w.lastReadErrMsg = err.Error()
			log.Error("failed to read config file", zap.String("filePath", w.manager.GetFilePath()), zap.Error(err))
		}
		return
	}
	if w.lastReadErrMsg != "" {
		w.lastReadErrMsg = ""
		log.Info("config file readable again", zap.String("filePath", w.manager.GetFilePath()))
	}

	sum := sha256.Sum256(data)
	if w.hasConfigHash && sum == w.lastConfigHash {
		return
	}

	// Update hash even if reload/apply fails: do not retry until content changes again.
	w.lastConfigHash = sum
	w.hasConfigHash = true

	log.Info("config file content changed, reloading", zap.String("filePath", w.manager.GetFilePath()))
	w.handleConfigChange(data)
}

// handleConfigChange reloads configuration and notifies handlers.
// The lastAuditConfig is only updated if all handlers succeed. If any handler fails,
// we don't retry until the config file content changes again.
func (w *Watcher) handleConfigChange(data []byte) {
	// Reload configuration using Manager
	if err := w.manager.loadFromBytes(data); err != nil {
		log.Error("failed to reload config", zap.Error(err), zap.String("filePath", w.manager.GetFilePath()))
		return
	}

	cfg := w.manager.Get()

	// Check if audit config changed (only hot-reloadable fields)
	w.mu.Lock()
	if !w.auditConfigChanged(&cfg.Audit) {
		w.mu.Unlock()
		log.Debug("audit config unchanged, skipping notification")
		return
	}

	// Copy handlers while holding the lock
	handlers := make([]AuditConfigChangeHandler, len(w.auditHandlers))
	copy(handlers, w.auditHandlers)
	w.mu.Unlock()

	log.Info("audit config changed, notifying handlers",
		zap.Bool("enabled", cfg.Audit.Enabled),
		zap.Duration("flushInterval", cfg.Audit.FlushInterval),
		zap.Bool("includeGetRequests", cfg.Audit.IncludeGetRequests),
		zap.Int("maxBodySize", cfg.Audit.MaxBodySize),
	)

	// Track if all handlers succeed
	allSucceeded := true
	for i, handler := range handlers {
		func() {
			defer func() {
				if r := recover(); r != nil {
					log.Error("handler panicked during audit config change",
						zap.Int("handler_index", i),
						zap.Any("panic", r),
						zap.Stack("stack"))
					allSucceeded = false
				}
			}()
			if err := handler.OnAuditConfigChange(&cfg.Audit); err != nil {
				log.Error("handler failed to process audit config change",
					zap.Int("handler_index", i),
					zap.Error(err))
				allSucceeded = false
			}
		}()
	}

	// Only update lastAuditConfig if all handlers succeeded.
	// This ensures that on next config change, if the config hasn't changed further,
	// we'll still retry the failed handlers.
	if allSucceeded {
		w.mu.Lock()
		w.lastAuditConfig = copyAuditConfig(&cfg.Audit)
		w.mu.Unlock()
		log.Info("audit config change applied successfully")
	} else {
		log.Warn("some handlers failed, will retry on next config change")
	}
}

// auditConfigChanged checks if hot-reloadable audit config fields changed.
func (w *Watcher) auditConfigChanged(newCfg *audit.Config) bool {
	if w.lastAuditConfig == nil {
		return true
	}
	old := w.lastAuditConfig

	return old.Enabled != newCfg.Enabled ||
		old.FlushInterval != newCfg.FlushInterval ||
		old.IncludeGetRequests != newCfg.IncludeGetRequests ||
		old.MaxBodySize != newCfg.MaxBodySize ||
		!sliceSetEqual(old.SensitiveQueryParams, newCfg.SensitiveQueryParams) ||
		!sliceSetEqual(old.SensitiveBodyFields, newCfg.SensitiveBodyFields)
}

// sliceSetEqual compares two string slices as sets (order-independent, case-insensitive).
// Duplicate elements are ignored.
func sliceSetEqual(a, b []string) bool {
	aSet := make(map[string]struct{}, len(a))
	for _, s := range a {
		aSet[strings.ToLower(s)] = struct{}{}
	}

	bSet := make(map[string]struct{}, len(b))
	for _, s := range b {
		bSet[strings.ToLower(s)] = struct{}{}
	}

	if len(aSet) != len(bSet) {
		return false
	}

	for s := range aSet {
		if _, ok := bSet[s]; !ok {
			return false
		}
	}
	return true
}

// copyAuditConfig creates a deep copy of audit config for comparison.
func copyAuditConfig(cfg *audit.Config) *audit.Config {
	cpy := *cfg
	if cfg.SensitiveQueryParams != nil {
		cpy.SensitiveQueryParams = make([]string, len(cfg.SensitiveQueryParams))
		copy(cpy.SensitiveQueryParams, cfg.SensitiveQueryParams)
	}
	if cfg.SensitiveBodyFields != nil {
		cpy.SensitiveBodyFields = make([]string, len(cfg.SensitiveBodyFields))
		copy(cpy.SensitiveBodyFields, cfg.SensitiveBodyFields)
	}
	return &cpy
}

// Close waits for Run() to finish with a timeout.
// The caller must cancel the context passed to Run() before calling Close().
// Returns an error if Run() doesn't finish within the timeout.
func (w *Watcher) Close() error {
	select {
	case <-w.done:
		return nil
	case <-time.After(closeTimeout):
		return errors.New("watcher close timeout: context was not cancelled")
	}
}