package backend

import (
	"context"
	"os"
	"strconv"
	"strings"
	"time"

	"go.uber.org/zap"

	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/metrics"
)

const (
	defaultPurgePollInterval = time.Minute
	// defaultMutationLogRetention bounds how long applied mutation log rows
	// are kept for replay/audit before the purge worker deletes them. The
	// table is append-only otherwise (observed in prod: 151M rows / 64GB in
	// under three months), and every reader only cares about pending rows or
	// rows younger than any in-flight scan watermark. Tune with
	// DRIVE9_QUOTA_MUTATION_LOG_RETENTION_HOURS.
	defaultMutationLogRetention = 7 * 24 * time.Hour
	// purgeBatchLimit bounds one DELETE statement so each purge round stays a
	// short PK-range transaction instead of one long lock-holding delete.
	purgeBatchLimit = 5000
	// purgeMaxRoundsPerTick caps the delete rounds per tick so a huge backlog
	// drains over many ticks, leaving headroom for foreground traffic.
	purgeMaxRoundsPerTick = 40
)

func purgePollInterval() time.Duration {
	d := envDurationMS("DRIVE9_QUOTA_PURGE_INTERVAL_MS", defaultPurgePollInterval)
	if d <= 0 {
		return defaultPurgePollInterval
	}
	// Floor the interval: each tick runs a PK binary search plus up to
	// purgeMaxRoundsPerTick DELETE batches, so a sub-second value becomes a
	// tight loop against the meta DB.
	if d < time.Second {
		return time.Second
	}
	return d
}

// maxMutationLogRetentionHours caps the retention override: without it a huge
// DRIVE9_QUOTA_MUTATION_LOG_RETENTION_HOURS value overflows
// time.Duration(h)*time.Hour to negative, moving the purge cutoff into the
// future and deleting every applied row.
const maxMutationLogRetentionHours = 10 * 365 * 24 // 10 years

func mutationLogRetentionHours() time.Duration {
	h := envIntHours("DRIVE9_QUOTA_MUTATION_LOG_RETENTION_HOURS", int(defaultMutationLogRetention/time.Hour))
	if h <= 0 {
		return defaultMutationLogRetention
	}
	if h > maxMutationLogRetentionHours {
		return time.Duration(maxMutationLogRetentionHours) * time.Hour
	}
	return time.Duration(h) * time.Hour
}

// envIntHours reads a positive-integer hours override; unset or invalid
// values fall back to def. Zero/negative means "use the default" rather than
// "disable" so a mis-set variable can never silently stop the purge.
func envIntHours(name string, def int) int {
	raw := strings.TrimSpace(os.Getenv(name))
	if raw == "" {
		return def
	}
	h, err := strconv.Atoi(raw)
	if err != nil {
		return def
	}
	return h
}

// MutationLogPurgeWorker periodically deletes old applied rows from
// quota_mutation_log. Like MutationReplayWorker it runs only on the leader
// (started from the same server wiring), so no cross-pod coordination is
// needed: batched deletes are idempotent by nature.
type MutationLogPurgeWorker struct {
	store  MetaQuotaStore
	cancel context.CancelFunc
	done   chan struct{}
}

// StartMutationLogPurgeWorker starts the background purge loop.
// Returns nil if store is nil (central quota not wired).
func StartMutationLogPurgeWorker(store MetaQuotaStore) *MutationLogPurgeWorker {
	if store == nil {
		return nil
	}
	ctx, cancel := context.WithCancel(backgroundWithTrace())
	w := &MutationLogPurgeWorker{
		store:  store,
		cancel: cancel,
		done:   make(chan struct{}),
	}
	go w.run(ctx)
	return w
}

// Stop gracefully shuts down the purge worker.
func (w *MutationLogPurgeWorker) Stop() {
	if w == nil {
		return
	}
	w.cancel()
	<-w.done
}

func (w *MutationLogPurgeWorker) run(ctx context.Context) {
	defer close(w.done)
	logger.Info(ctx, "mutation_log_purge_worker_started",
		zap.Duration("retention", mutationLogRetentionHours()),
		zap.Duration("poll_interval", purgePollInterval()))
	ticker := time.NewTicker(purgePollInterval())
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			logger.Info(ctx, "mutation_log_purge_worker_stopped")
			return
		case <-ticker.C:
			w.purgeOnce(ctx)
		}
	}
}

// purgeOnce deletes up to purgeMaxRoundsPerTick batches of expired applied
// rows. Each batch is its own short transaction; a partial failure stops the
// tick and retries on the next one.
func (w *MutationLogPurgeWorker) purgeOnce(ctx context.Context) {
	start := time.Now()
	cutoff := time.Now().UTC().Add(-mutationLogRetentionHours())
	var total int64
	for round := 0; round < purgeMaxRoundsPerTick; round++ {
		deleted, err := w.store.PurgeAppliedMutations(ctx, cutoff, purgeBatchLimit)
		if err != nil {
			logger.Warn(ctx, "mutation_log_purge_batch_failed", zap.Error(err))
			metrics.RecordOperation("mutation_log_purge", "delete", metrics.ResultForError(err), time.Since(start))
			return
		}
		total += deleted
		if deleted < purgeBatchLimit {
			break
		}
	}
	metrics.RecordOperation("mutation_log_purge", "delete", "ok", time.Since(start))
	if total > 0 {
		logger.Info(ctx, "mutation_log_purge_tick_complete",
			zap.Int64("deleted", total),
			zap.Time("cutoff", cutoff),
			zap.Float64("duration_ms", float64(time.Since(start).Milliseconds())))
	}
}