package metrics

import "testing"

// drainFileOpsForTest flushes any tracker state left by earlier tests and
// returns a per-tenant delta extractor, so assertions never depend on
// process-global counts or test order.
func drainFileOpsForTest(t *testing.T) (deltasFor func(tenantID string) []FileOpsDelta) {
	t.Helper()
	MarkFileOpsFlushed(SnapshotFileOpsDeltas())
	return func(tenantID string) []FileOpsDelta {
		t.Helper()
		var out []FileOpsDelta
		for _, d := range SnapshotFileOpsDeltas() {
			if d.TenantID == tenantID {
				out = append(out, d)
			}
		}
		return out
	}
}

func TestFileOpsFlushDeltaTracking(t *testing.T) {
	const tenantID = "tenant-ops-flush"
	const orgID = "org-ops-flush"
	deltasFor := drainFileOpsForTest(t)
	t.Cleanup(func() { DeleteTenantCounters(tenantID) })

	RecordTenantFileOpsWithOrg(tenantID, orgID, "read", "db9", 3)
	RecordTenantFileOpsWithOrg(tenantID, orgID, "read", "db9", 2)
	RecordTenantFileOpsWithOrg(tenantID, orgID, "write", "s3", 1)

	deltas := deltasFor(tenantID)
	byDirection := map[string]int64{}
	for _, d := range deltas {
		byDirection[d.Direction+"/"+d.TrafficPath] = d.Ops
	}
	if byDirection["read/db9"] != 5 || byDirection["write/s3"] != 1 {
		t.Fatalf("deltas = %v", byDirection)
	}

	// Snapshot again without marking: deltas must not shrink (failed flush
	// retries the full accumulated value).
	if got := deltasFor(tenantID); len(got) != 2 {
		t.Fatalf("second snapshot lost unflushed deltas: %v", got)
	}

	MarkFileOpsFlushed(SnapshotFileOpsDeltas())
	if got := deltasFor(tenantID); len(got) != 0 {
		t.Fatalf("deltas after mark flushed: %+v", got)
	}

	// New increments appear as fresh deltas from the advanced baseline.
	RecordTenantFileOpsWithOrg(tenantID, orgID, "read", "db9", 4)
	deltas = deltasFor(tenantID)
	if len(deltas) != 1 || deltas[0].Ops != 4 {
		t.Fatalf("post-flush deltas = %+v, want single read/db9=4", deltas)
	}
	MarkFileOpsFlushed(SnapshotFileOpsDeltas())
}

func TestFileOpsFlushDropOnTenantEviction(t *testing.T) {
	const tenantID = "tenant-ops-flush-evict"
	deltasFor := drainFileOpsForTest(t)
	t.Cleanup(func() { DeleteTenantCounters(tenantID) })

	RecordTenantFileOpsWithOrg(tenantID, "org-evict", "read", "db9", 7)

	// Eviction with an unflushed delta moves it to the stranded list instead
	// of dropping it: the next flush still bills it.
	DeleteTenantCounters(tenantID)
	deltas := deltasFor(tenantID)
	if len(deltas) != 1 || deltas[0].Ops != 7 {
		t.Fatalf("stranded delta after eviction = %+v, want single delta of 7", deltas)
	}
	MarkFileOpsFlushed(SnapshotFileOpsDeltas())

	// A reactivated tenant starts from its fresh counter values.
	RecordTenantFileOpsWithOrg(tenantID, "org-evict", "read", "db9", 2)
	deltas = deltasFor(tenantID)
	if len(deltas) != 1 || deltas[0].Ops != 2 {
		t.Fatalf("deltas after reactivation = %+v, want single delta of 2", deltas)
	}
	MarkFileOpsFlushed(SnapshotFileOpsDeltas())
}

// TestFileOpsFlushSnapshotMarkRace pins that increments recorded between
// SnapshotFileOpsDeltas and MarkFileOpsFlushed stay pending instead of being
// treated as flushed.
func TestFileOpsFlushSnapshotMarkRace(t *testing.T) {
	const tenantID = "tenant-ops-flush-race"
	deltasFor := drainFileOpsForTest(t)
	t.Cleanup(func() { DeleteTenantCounters(tenantID) })

	RecordTenantFileOpsWithOrg(tenantID, "org-race", "read", "db9", 5)
	deltas := SnapshotFileOpsDeltas()
	// Increments land while the snapshot is being persisted.
	RecordTenantFileOpsWithOrg(tenantID, "org-race", "read", "db9", 3)
	MarkFileOpsFlushed(deltas)

	rest := deltasFor(tenantID)
	if len(rest) != 1 || rest[0].Ops != 3 {
		t.Fatalf("pending after race = %+v, want single delta of 3", rest)
	}
	MarkFileOpsFlushed(SnapshotFileOpsDeltas())
	if got := deltasFor(tenantID); len(got) != 0 {
		t.Fatalf("deltas after final mark = %+v, want none", got)
	}
}

// TestFileOpsFlushMarkAfterEvictionCannotResurrectBaseline pins that a Mark
// for a pre-eviction snapshot cannot write its absolute value as the baseline
// of the reactivated series (which would swallow new increments).
func TestFileOpsFlushMarkAfterEvictionCannotResurrectBaseline(t *testing.T) {
	const tenantID = "tenant-ops-flush-resurrect"
	deltasFor := drainFileOpsForTest(t)
	t.Cleanup(func() { DeleteTenantCounters(tenantID) })

	RecordTenantFileOpsWithOrg(tenantID, "org-resurrect", "read", "db9", 10)
	staleSnapshot := SnapshotFileOpsDeltas()

	// Tenant is evicted and reactivated while the snapshot is in flight.
	DeleteTenantCounters(tenantID)
	RecordTenantFileOpsWithOrg(tenantID, "org-resurrect", "read", "db9", 2)

	// The stale Mark acknowledges the stranded delta but must not set a
	// baseline of 10 on the new series (which would swallow the 2).
	MarkFileOpsFlushed(staleSnapshot)
	rest := deltasFor(tenantID)
	if len(rest) != 1 || rest[0].Ops != 2 {
		t.Fatalf("pending after stale mark = %+v, want single delta of 2", rest)
	}
	MarkFileOpsFlushed(SnapshotFileOpsDeltas())
	if got := deltasFor(tenantID); len(got) != 0 {
		t.Fatalf("deltas after final mark = %+v, want none", got)
	}
}

// TestFileOpsFlushGenerationsNeverRepeat pins that every series incarnation
// gets a fresh generation, so flush rows from before an eviction can never
// collide with post-eviction ones in the billing_ops_delta idempotency key.
func TestFileOpsFlushGenerationsNeverRepeat(t *testing.T) {
	const tenantID = "tenant-ops-flush-gen"
	deltasFor := drainFileOpsForTest(t)
	t.Cleanup(func() { DeleteTenantCounters(tenantID) })

	var gens []int64
	for i := 0; i < 3; i++ {
		RecordTenantFileOpsWithOrg(tenantID, "org-gen", "read", "db9", 1)
		deltas := deltasFor(tenantID)
		if len(deltas) != 1 {
			t.Fatalf("cycle %d: deltas = %+v", i, deltas)
		}
		gens = append(gens, deltas[0].Gen())
		MarkFileOpsFlushed(SnapshotFileOpsDeltas())
		DeleteTenantCounters(tenantID)
	}
	seen := map[int64]bool{}
	for _, g := range gens {
		if seen[g] {
			t.Fatalf("generation reused: %v", gens)
		}
		seen[g] = true
	}
}

// TestFileOpsFlushStrandedCapNoDeadlock drives the stranded cap through
// DeleteTenantCounters, which holds the registry write lock: emitting a
// metric from inside dropTenant would re-enter that non-reentrant lock and
// deadlock. The cap must only count; the metric is published by the server
// flush loop.
func TestFileOpsFlushStrandedCapNoDeadlock(t *testing.T) {
	old := maxStrandedFileOpsDeltas
	maxStrandedFileOpsDeltas = 2
	t.Cleanup(func() { maxStrandedFileOpsDeltas = old })
	drainFileOpsForTest(t)
	before := FileOpsFlushStrandedDropped()

	const tenantID = "tenant-ops-flush-cap"
	t.Cleanup(func() { DeleteTenantCounters(tenantID) })
	// Each evict-with-pending strands one entry; three evictions exceed cap=2.
	for i := 0; i < 3; i++ {
		RecordTenantFileOpsWithOrg(tenantID, "org-cap", "read", "db9", 1)
		DeleteTenantCounters(tenantID)
	}
	if got := FileOpsFlushStrandedDropped(); got != before+1 {
		t.Fatalf("stranded dropped = %d, want %d", got, before+1)
	}
	// The stranded list stays at the cap; all three deltas remain billable
	// except the one explicitly dropped.
	deltas := SnapshotFileOpsDeltas()
	if len(deltas) != 2 {
		t.Fatalf("stranded after cap = %d, want 2", len(deltas))
	}
	MarkFileOpsFlushed(deltas)
}

// TestFileOpsFlushDisabledGate pins that the record hot path does no
// bookkeeping when the flush is disabled (the production default).
func TestFileOpsFlushDisabledGate(t *testing.T) {
	SetFileOpsFlushEnabled(false)
	t.Cleanup(func() { SetFileOpsFlushEnabled(true) })
	drainFileOpsForTest(t)
	RecordTenantFileOpsWithOrg("tenant-ops-flush-gate", "org-gate", "read", "db9", 5)
	if got := SnapshotFileOpsDeltas(); len(got) != 0 {
		t.Fatalf("tracker recorded while disabled: %+v", got)
	}
	DeleteTenantCounters("tenant-ops-flush-gate")
}