package metrics

import (
	"sync"
	"sync/atomic"
)

// FileOpsDelta is one billing flush record: ops accumulated for one
// (tenant, org, direction, traffic_path) series since the last flush.
type FileOpsDelta struct {
	TenantID    string
	OrgID       string
	Direction   string
	TrafficPath string
	Ops         int64
	// absolute is the counter value at snapshot time; MarkFileOpsFlushed
	// advances the baseline to it so increments recorded between snapshot and
	// mark are never treated as flushed.
	absolute int64
	// gen identifies the series incarnation: a process-wide monotonic counter
	// assigned when a series is (re)created, so a counter reset after tenant
	// eviction can never reuse a generation this process already flushed.
	gen int64
	// stranded marks deltas from evicted tenants; they no longer have live
	// series state and are acknowledged by removing them from the stranded
	// list instead of touching baselines.
	stranded bool
}

// Cumulative returns the series' absolute counter value at snapshot time.
// Together with pod/series identity and the series generation it forms the
// idempotency key of the billing_ops_delta table.
func (d FileOpsDelta) Cumulative() int64 { return d.absolute }

// Gen returns the series incarnation generation at snapshot time. Every
// reincarnation after an eviction gets a generation this process has never
// used, so pre-eviction flush rows can never collide with post-eviction ones.
func (d FileOpsDelta) Gen() int64 { return d.gen }

// fileOpsSeriesState is the live state of one counter series: its absolute
// value and the incarnation generation assigned at (re)creation.
type fileOpsSeriesState struct {
	value int64
	gen   int64
}

// fileOpsFlushTracker mirrors the absolute values of the
// drive9_tenant_file_ops_total counter series so the server can flush deltas
// to the central meta DB for the billing worker, without depending on an
// external metrics store. Baselines advance after a successful flush — and,
// as the server's deliberate escape valve, when a wedged pending batch is
// dropped after maxBillingFlushFailures consecutive failures. A process
// restart simply restarts from the in-memory counter's fresh absolute values.
type fileOpsFlushTracker struct {
	mu          sync.Mutex
	current     map[fileOpsKey]fileOpsSeriesState
	lastFlushed map[fileOpsKey]int64
	// nextGen is the process-wide monotonic generation counter. Each series
	// incarnation gets a fresh value, so there is no per-tenant state to
	// prune and no generation reuse within a process lifetime.
	nextGen int64
	// inFlight is the high-water absolute value captured by snapshots that
	// have not been marked yet. dropTenant only strands the part of a series
	// not covered by any in-flight snapshot, so an eviction cannot duplicate
	// a delta that an in-flight flush may still persist. The residual loss
	// window is a FAILED flush (never marked) whose tenant is evicted before
	// the next successful flush — during a meta-DB outage, bounded by the
	// outage length.
	inFlight map[fileOpsKey]int64
	stranded []FileOpsDelta
	// strandedDropped counts stranded entries dropped at the cap. Emitted as
	// a metric by the server flush loop, never from inside the tracker (which
	// may be running under the registry write lock).
	strandedDropped int64
}

type fileOpsKey struct {
	tenantID    string
	orgID       string
	direction   string
	trafficPath string
}

var fileOpsFlush = newFileOpsFlushTracker()

// fileOpsFlushEnabled gates tracker work on the record hot path. The server
// disables it when the billing flush is not configured (interval 0), so
// default deployments pay nothing for tracking that would never be flushed.
var fileOpsFlushEnabled atomic.Bool

func init() {
	fileOpsFlushEnabled.Store(true)
}

// SetFileOpsFlushEnabled toggles ops-flush tracking. Drive9-server calls it
// at startup from DRIVE9_BILLING_OPS_FLUSH_INTERVAL; tests rely on the
// default (enabled).
func SetFileOpsFlushEnabled(enabled bool) {
	fileOpsFlushEnabled.Store(enabled)
}

// maxStrandedFileOpsDeltas bounds the stranded list when no flusher is
// configured or flushes keep failing across evictions.
var maxStrandedFileOpsDeltas = 100000

// FileOpsFlushStrandedDropped returns how many stranded deltas have been
// dropped at the cap. The server flush loop publishes this as a metric
// outside the registry lock.
func FileOpsFlushStrandedDropped() int64 {
	fileOpsFlush.mu.Lock()
	defer fileOpsFlush.mu.Unlock()
	return fileOpsFlush.strandedDropped
}

func newFileOpsFlushTracker() *fileOpsFlushTracker {
	return &fileOpsFlushTracker{
		current:     map[fileOpsKey]fileOpsSeriesState{},
		lastFlushed: map[fileOpsKey]int64{},
		inFlight:    map[fileOpsKey]int64{},
	}
}

func (t *fileOpsFlushTracker) add(tenantID, orgID, direction, trafficPath string, ops int64) {
	if !fileOpsFlushEnabled.Load() {
		return
	}
	key := fileOpsKey{tenantID: tenantID, orgID: orgID, direction: direction, trafficPath: trafficPath}
	t.mu.Lock()
	state, ok := t.current[key]
	if !ok {
		// New series incarnation (first record, or reactivation after
		// eviction dropped the old one): assign a generation this process
		// has never used.
		t.nextGen++
		state.gen = t.nextGen
	}
	state.value += ops
	t.current[key] = state
	t.mu.Unlock()
}

// SnapshotFileOpsDeltas returns the per-series deltas since the last
// successful MarkFileOpsFlushed, plus any stranded deltas from evicted
// tenants that are still owed. It does not advance baselines, so a failed
// flush can be retried without losing increments.
func SnapshotFileOpsDeltas() []FileOpsDelta {
	fileOpsFlush.mu.Lock()
	defer fileOpsFlush.mu.Unlock()
	out := make([]FileOpsDelta, 0, len(fileOpsFlush.stranded))
	out = append(out, fileOpsFlush.stranded...)
	for key, state := range fileOpsFlush.current {
		delta := state.value - fileOpsFlush.lastFlushed[key]
		if delta <= 0 {
			continue
		}
		if state.value > fileOpsFlush.inFlight[key] {
			fileOpsFlush.inFlight[key] = state.value
		}
		out = append(out, FileOpsDelta{
			TenantID:    key.tenantID,
			OrgID:       key.orgID,
			Direction:   key.direction,
			TrafficPath: key.trafficPath,
			Ops:         delta,
			absolute:    state.value,
			gen:         state.gen,
		})
	}
	return out
}

// MarkFileOpsFlushed acknowledges a snapshot: live-series baselines advance
// to the snapshotted absolute values (only when the series generation still
// matches, so an eviction between snapshot and mark cannot resurrect a stale
// baseline), and acknowledged stranded deltas are dropped. Call it with the
// exact slice returned by SnapshotFileOpsDeltas after the deltas have been
// durably persisted — or when deliberately dropping a wedged batch (the
// server's escape valve), where advancing past the batch is the point.
func MarkFileOpsFlushed(deltas []FileOpsDelta) {
	fileOpsFlush.mu.Lock()
	defer fileOpsFlush.mu.Unlock()
	for _, d := range deltas {
		if d.stranded {
			continue // handled below
		}
		key := fileOpsKey{tenantID: d.TenantID, orgID: d.OrgID, direction: d.Direction, trafficPath: d.TrafficPath}
		// The snapshot lifecycle is over for this key either way; clear the
		// in-flight marker even when the baseline advance is skipped.
		delete(fileOpsFlush.inFlight, key)
		state, live := fileOpsFlush.current[key]
		if !live || state.gen != d.gen {
			continue // series was evicted/recreated after the snapshot
		}
		if d.absolute > fileOpsFlush.lastFlushed[key] {
			fileOpsFlush.lastFlushed[key] = d.absolute
		}
	}
	acked := map[FileOpsDelta]bool{}
	for _, d := range deltas {
		if d.stranded {
			acked[d] = true
		}
	}
	if len(acked) == 0 {
		return
	}
	kept := fileOpsFlush.stranded[:0]
	for _, d := range fileOpsFlush.stranded {
		if !acked[d] {
			kept = append(kept, d)
		}
	}
	fileOpsFlush.stranded = kept
}

// dropTenant mirrors tenant series eviction. Clean series (nothing unflushed)
// are dropped outright; pending deltas move to the stranded list so they are
// still billed once flushing recovers, stamped with the dropped series
// incarnation. A delta already covered by an in-flight snapshot is not
// stranded — that flush will persist it (the documented residual loss window
// is a failed flush followed by eviction before the next successful one).
// The next incarnation of the same series gets a fresh generation, so a Mark
// for a pre-eviction snapshot cannot write a baseline onto it.
func (t *fileOpsFlushTracker) dropTenant(tenantID string) {
	t.mu.Lock()
	defer t.mu.Unlock()
	for key, state := range t.current {
		if key.tenantID != tenantID {
			continue
		}
		covered := t.lastFlushed[key]
		if t.inFlight[key] > covered {
			covered = t.inFlight[key]
		}
		if delta := state.value - covered; delta > 0 {
			// The stranded list is drained by the server's flush loop. When no
			// flusher is configured (or a meta outage outlasts evictions),
			// bound it: drop the oldest stranded deltas rather than growing
			// for the process lifetime. Only counted here — the metric is
			// published by the server flush loop, because dropTenant runs
			// under the registry write lock and emitting a metric would
			// re-enter it (non-reentrant deadlock).
			if len(t.stranded) >= maxStrandedFileOpsDeltas {
				t.stranded = t.stranded[1:]
				t.strandedDropped++
			}
			t.stranded = append(t.stranded, FileOpsDelta{
				TenantID:    key.tenantID,
				OrgID:       key.orgID,
				Direction:   key.direction,
				TrafficPath: key.trafficPath,
				Ops:         delta,
				absolute:    state.value,
				gen:         state.gen,
				stranded:    true,
			})
		}
		delete(t.current, key)
		delete(t.lastFlushed, key)
		delete(t.inFlight, key)
	}
}