package server

import (
	"context"
	"errors"
	"fmt"
	"os"
	"sync"
	"time"

	"go.uber.org/zap"

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

// Billing ops flush: every pod owns an in-process drive9_tenant_file_ops_total
// counter, so every pod periodically flushes its deltas to the central meta
// DB (billing_ops_delta). The billing worker aggregates those rows per hour;
// no external metrics store is on the billing data path.

// billingFlushHandle coordinates the periodic flush loop with shutdown so the
// final flush cannot overlap an in-flight one (the delta table is
// idempotent per cumulative point, and the single-flight mutex additionally
// keeps snapshot/insert/mark pairs from interleaving).
type billingFlushHandle struct {
	cancel context.CancelFunc
	wg     sync.WaitGroup
	podID  string
}

// StartBillingOpsFlush launches the periodic flush loop. It is a no-op when
// the central meta store is unavailable (e.g. tests without a control plane)
// or interval <= 0 (explicitly disabled).
func (s *Server) StartBillingOpsFlush(ctx context.Context, interval time.Duration) {
	if s.meta == nil {
		return
	}
	if interval <= 0 {
		s.billingFlushMu.Lock()
		s.billingFlushDisabled = true
		s.billingFlushMu.Unlock()
		return
	}
	loopCtx, cancel := context.WithCancel(context.Background())
	handle := &billingFlushHandle{cancel: cancel, podID: billingFlushPodID()}
	s.billingFlushMu.Lock()
	s.billingFlush = handle
	s.billingFlushMu.Unlock()

	handle.wg.Add(1)
	go func() {
		defer handle.wg.Done()
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-loopCtx.Done():
				return
			case <-ticker.C:
			}
			// The loop's flush ctx derives from the caller's (stopCtx): on
			// SIGTERM an in-flight insert is canceled and rolls back; the
			// identical batch is retried by the final flush (same-batch
			// pending retry + INSERT IGNORE make this safe).
			flushCtx, flushCancel := context.WithTimeout(ctx, 30*time.Second)
			if err := s.FlushBillingOps(flushCtx, handle.podID); err != nil {
				// Alertable: sustained flush failures mean ops usage is
				// delayed (and at risk once the pending-batch cap trips).
				metrics.RecordOperation("server", "billing_ops_flush", metrics.ResultForError(err), 0)
				logger.Warn(ctx, "billing_ops_flush_failed", zap.Error(err))
			}
			flushCancel()
		}
	}()
}

// StopBillingOpsFlush cancels the periodic loop, waits for it to exit, then
// runs the final flush with its own timeout (started only after the loop has
// exited, so an in-flight flush never eats the final flush's budget). Call
// during graceful shutdown, before store teardown. When the flush was
// explicitly disabled (interval <= 0), no lifetime flush happens at shutdown.
func (s *Server) StopBillingOpsFlush() {
	s.billingFlushMu.Lock()
	handle := s.billingFlush
	disabled := s.billingFlushDisabled
	s.billingFlushMu.Unlock()
	if handle != nil {
		handle.cancel()
		handle.wg.Wait()
	}
	if disabled {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	podID := ""
	if handle != nil {
		podID = handle.podID
	}
	if err := s.FlushBillingOps(ctx, podID); err != nil {
		logger.Warn(context.Background(), "billing_ops_final_flush_failed", zap.Error(err))
	}
}

// FlushBillingOps persists the pending ops deltas once. It is single-flighted
// per server: concurrent callers serialize so snapshot/insert/mark pairs
// cannot interleave. A failed insert keeps the SAME snapshot pending and the
// next call retries that exact batch (the idempotency key only dedupes
// identical points; re-snapshotting after an ambiguous commit would produce a
// different cumulative and double-count). Baselines advance on success — and,
// as the deliberate escape valve, on a batch dropped after
// maxBillingFlushFailures consecutive non-context failures (those ops go
// unbilled; alert on billing_ops_flush=dropped_pending_batch).
//
// podID is part of the idempotency key and must be stable across retries of
// the same pending batch; pass "" to use the process identity (the periodic
// and shutdown paths always pass the fixed one).
func (s *Server) FlushBillingOps(ctx context.Context, podID string) error {
	if s.meta == nil {
		return nil
	}
	if podID == "" {
		podID = billingFlushPodID()
	}
	s.billingFlushMu.Lock()
	defer s.billingFlushMu.Unlock()
	// Publish stranded-cap drops outside the tracker/registry locks. Uses the
	// same component/operation as dropped_pending_batch so alerts on
	// deliberately discarded billing data share one convention.
	if dropped := metrics.FileOpsFlushStrandedDropped(); dropped > s.billingFlushStrandedReported {
		metrics.RecordOperationCount("server", "billing_ops_flush", "stranded_dropped", dropped-s.billingFlushStrandedReported)
		s.billingFlushStrandedReported = dropped
	}
	deltas := s.billingFlushPending
	if deltas == nil {
		deltas = metrics.SnapshotFileOpsDeltas()
	}
	if len(deltas) == 0 {
		s.billingFlushPending = nil
		return nil
	}
	rows := make([]meta.BillingOpsDelta, 0, len(deltas))
	for _, d := range deltas {
		rows = append(rows, meta.BillingOpsDelta{
			TenantID:    d.TenantID,
			OrgID:       d.OrgID,
			Direction:   d.Direction,
			TrafficPath: d.TrafficPath,
			Ops:         d.Ops,
			Cumulative:  d.Cumulative(),
			Gen:         d.Gen(),
		})
	}
	if err := s.meta.InsertBillingOpsDeltas(ctx, podID, time.Now().UTC(), rows); err != nil {
		s.billingFlushPending = deltas
		// Context cancellation/deadline (shutdown, slow DB) is not a
		// deterministic batch failure: the insert rolled back and the
		// identical batch is safely retried, so it never counts toward the
		// drop cap. Only errors that can be deterministic wedge the cap.
		if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
			s.billingFlushFailures++
		}
		if s.billingFlushFailures >= maxBillingFlushFailures {
			// A deterministically failing batch would otherwise wedge every
			// future flush on this pod. Drop it loudly: baselines advance past
			// the batch (those ops go unbilled — alert on the
			// billing_ops_flush=dropped_pending_batch metric) and flushing
			// resumes.
			metrics.RecordOperation("server", "billing_ops_flush", "dropped_pending_batch", 0)
			logger.Error(ctx, "billing_ops_flush_pending_batch_dropped",
				zap.Int("consecutive_failures", s.billingFlushFailures), zap.Error(err))
			metrics.MarkFileOpsFlushed(deltas)
			s.billingFlushPending = nil
			s.billingFlushFailures = 0
			return fmt.Errorf("insert billing ops deltas (batch dropped after %d failures): %w", maxBillingFlushFailures, err)
		}
		return fmt.Errorf("insert billing ops deltas: %w", err)
	}
	metrics.MarkFileOpsFlushed(deltas)
	s.billingFlushPending = nil
	s.billingFlushFailures = 0
	return nil
}

// maxBillingFlushFailures caps consecutive failures of the same pending
// batch before it is dropped (loudly) to unwedge future flushes.
const maxBillingFlushFailures = 10

func billingFlushPodID() string {
	hostname, err := os.Hostname()
	if err != nil || hostname == "" {
		hostname = "unknown"
	}
	// Guard against the pod_id column length: a truncated-but-stable hostname
	// is fine, a silently truncated one with a colliding format is not.
	if len(hostname) > 100 {
		hostname = hostname[:100]
	}
	// Boot timestamp keeps the idempotency key unique across process restarts
	// on the same host (PID reuse).
	return fmt.Sprintf("%s/%d/%d", hostname, os.Getpid(), processBootUnixNano)
}

var processBootUnixNano = time.Now().UnixNano()