package meta

import (
	"context"
	"testing"
	"time"

	"github.com/mem9-ai/drive9/pkg/testtidb"
)

// insertOldTenantNotify inserts one outbox row with an old created_at and
// returns its id.
func insertOldTenantNotify(t *testing.T, s *Store, tenantID string, age time.Duration) uint64 {
	t.Helper()
	res, err := s.DB().ExecContext(context.Background(),
		`INSERT INTO tenant_notify_outbox (tenant_id, work_mask, created_at) VALUES (?, ?, ?)`,
		tenantID, 1, time.Now().Add(-age))
	if err != nil {
		t.Fatalf("insert old outbox row: %v", err)
	}
	id, err := res.LastInsertId()
	if err != nil {
		t.Fatalf("last insert id: %v", err)
	}
	return uint64(id)
}

func listTenantNotifyIDs(t *testing.T, s *Store) map[uint64]bool {
	t.Helper()
	rows, err := s.ListTenantNotifySince(context.Background(), 0, 100)
	if err != nil {
		t.Fatalf("ListTenantNotifySince: %v", err)
	}
	ids := make(map[uint64]bool, len(rows))
	for _, r := range rows {
		ids[r.ID] = true
	}
	return ids
}

// TestDeleteTenantNotifyBeforeFreshFloor verifies the pruning floor: rows are
// pruned up to MIN(last_id) across fresh cursors, stale cursors are ignored,
// and age-only pruning applies when no fresh cursors remain.
func TestDeleteTenantNotifyBeforeFreshFloor(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	cutoff := func() time.Time { return time.Now().Add(-time.Hour) }

	// Case A: a fresh cursor holds the floor — rows above its last_id survive
	// even when older than the retention.
	testtidb.ResetMetaDB(t, s.DB())
	id1 := insertOldTenantNotify(t, s, "tenant-a", 2*time.Hour)
	id2 := insertOldTenantNotify(t, s, "tenant-a", 2*time.Hour)
	id3 := insertOldTenantNotify(t, s, "tenant-a", 2*time.Hour)
	if err := s.UpsertTenantOutboxCursor(ctx, "pod-fresh", id2); err != nil {
		t.Fatal(err)
	}
	n, err := s.DeleteTenantNotifyBefore(ctx, cutoff())
	if err != nil {
		t.Fatalf("DeleteTenantNotifyBefore case A: %v", err)
	}
	if n != 2 {
		t.Fatalf("case A deleted = %d, want 2 (floor at cursor last_id)", n)
	}
	ids := listTenantNotifyIDs(t, s)
	if ids[id1] || ids[id2] {
		t.Fatalf("case A: rows <= floor should be pruned: %v", ids)
	}
	if !ids[id3] {
		t.Fatalf("case A: row above fresh cursor floor must survive: %v", ids)
	}

	// Case B: the only cursor is stale (updated_at older than the freshness
	// bound) — it is ignored, so pruning falls back to age alone.
	testtidb.ResetMetaDB(t, s.DB())
	id1 = insertOldTenantNotify(t, s, "tenant-a", 2*time.Hour)
	id2 = insertOldTenantNotify(t, s, "tenant-a", 2*time.Hour)
	id3 = insertOldTenantNotify(t, s, "tenant-a", 2*time.Hour)
	if err := s.UpsertTenantOutboxCursor(ctx, "pod-stalled", id1); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx,
		`UPDATE tenant_outbox_cursor SET updated_at = ? WHERE pod_id = ?`,
		time.Now().Add(-2*tenantOutboxCursorFreshnessBound), "pod-stalled"); err != nil {
		t.Fatal(err)
	}
	n, err = s.DeleteTenantNotifyBefore(ctx, cutoff())
	if err != nil {
		t.Fatalf("DeleteTenantNotifyBefore case B: %v", err)
	}
	if n != 3 {
		t.Fatalf("case B deleted = %d, want 3 (stale cursor ignored, age-only prune)", n)
	}
	if ids = listTenantNotifyIDs(t, s); ids[id1] || ids[id2] || ids[id3] {
		t.Fatalf("case B: all old rows should be pruned: %v", ids)
	}

	// Case C: a stale cursor with a low last_id alongside a fresh cursor with
	// a high last_id — the floor comes from the fresh cursor only.
	testtidb.ResetMetaDB(t, s.DB())
	id1 = insertOldTenantNotify(t, s, "tenant-a", 2*time.Hour)
	id2 = insertOldTenantNotify(t, s, "tenant-a", 2*time.Hour)
	id3 = insertOldTenantNotify(t, s, "tenant-a", 2*time.Hour)
	if err := s.UpsertTenantOutboxCursor(ctx, "pod-stalled", id1); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx,
		`UPDATE tenant_outbox_cursor SET updated_at = ? WHERE pod_id = ?`,
		time.Now().Add(-2*tenantOutboxCursorFreshnessBound), "pod-stalled"); err != nil {
		t.Fatal(err)
	}
	if err := s.UpsertTenantOutboxCursor(ctx, "pod-fresh", id2); err != nil {
		t.Fatal(err)
	}
	n, err = s.DeleteTenantNotifyBefore(ctx, cutoff())
	if err != nil {
		t.Fatalf("DeleteTenantNotifyBefore case C: %v", err)
	}
	if n != 2 {
		t.Fatalf("case C deleted = %d, want 2 (floor from fresh cursor)", n)
	}
	if ids = listTenantNotifyIDs(t, s); ids[id1] || ids[id2] || !ids[id3] {
		t.Fatalf("case C: want id1,id2 pruned and id3 retained: %v", ids)
	}

	// Case D: no cursors at all — plain age-only pruning, fresh rows survive.
	testtidb.ResetMetaDB(t, s.DB())
	insertOldTenantNotify(t, s, "tenant-a", 2*time.Hour)
	if err := s.InsertTenantNotify(ctx, "tenant-a", 1); err != nil {
		t.Fatal(err)
	}
	n, err = s.DeleteTenantNotifyBefore(ctx, cutoff())
	if err != nil {
		t.Fatalf("DeleteTenantNotifyBefore case D: %v", err)
	}
	if n != 1 {
		t.Fatalf("case D deleted = %d, want 1", n)
	}
	if got := len(listTenantNotifyIDs(t, s)); got != 1 {
		t.Fatalf("case D: fresh row must survive, remaining = %d", got)
	}
}

// TestUpsertTenantOutboxCursorRefreshesUpdatedAt verifies the cursor's
// updated_at advances only when last_id advances — the freshness bound in
// DeleteTenantNotifyBefore depends on it: a pod whose poll is stuck (frozen
// last_id) must go stale even while its cursor flushes keep succeeding.
func TestUpsertTenantOutboxCursorRefreshesUpdatedAt(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	testtidb.ResetMetaDB(t, s.DB())
	stale := func() time.Time { return time.Now().Add(-2 * tenantOutboxCursorFreshnessBound) }
	age := func(podID string) {
		t.Helper()
		if _, err := s.DB().ExecContext(ctx,
			`UPDATE tenant_outbox_cursor SET updated_at = ? WHERE pod_id = ?`, stale(), podID); err != nil {
			t.Fatal(err)
		}
	}

	if err := s.UpsertTenantOutboxCursor(ctx, "pod-a", 10); err != nil {
		t.Fatal(err)
	}
	age("pod-a")

	// Advancing last_id refreshes updated_at.
	if err := s.UpsertTenantOutboxCursor(ctx, "pod-a", 20); err != nil {
		t.Fatal(err)
	}
	cursor, err := s.GetTenantOutboxCursor(ctx, "pod-a")
	if err != nil {
		t.Fatal(err)
	}
	if cursor.LastID != 20 {
		t.Fatalf("last_id = %d, want 20", cursor.LastID)
	}
	if time.Since(cursor.UpdatedAt) > tenantOutboxCursorFreshnessBound {
		t.Fatalf("updated_at = %v not refreshed by an advancing upsert", cursor.UpdatedAt)
	}

	// Re-flushing the SAME last_id (poller caught up, zero traffic — or poll
	// failing) must NOT refresh updated_at: the cursor goes stale.
	age("pod-a")
	if err := s.UpsertTenantOutboxCursor(ctx, "pod-a", 20); err != nil {
		t.Fatal(err)
	}
	cursor, err = s.GetTenantOutboxCursor(ctx, "pod-a")
	if err != nil {
		t.Fatal(err)
	}
	if time.Since(cursor.UpdatedAt) <= tenantOutboxCursorFreshnessBound {
		t.Fatalf("updated_at = %v refreshed by an unchanged last_id; cursor would never go stale", cursor.UpdatedAt)
	}

	// A regressing last_id must NOT refresh updated_at either (but last_id
	// itself follows the upsert, preserving prior upsert semantics).
	age("pod-a")
	if err := s.UpsertTenantOutboxCursor(ctx, "pod-a", 5); err != nil {
		t.Fatal(err)
	}
	cursor, err = s.GetTenantOutboxCursor(ctx, "pod-a")
	if err != nil {
		t.Fatal(err)
	}
	if cursor.LastID != 5 {
		t.Fatalf("last_id = %d, want 5", cursor.LastID)
	}
	if time.Since(cursor.UpdatedAt) <= tenantOutboxCursorFreshnessBound {
		t.Fatalf("updated_at = %v refreshed by a regressing last_id", cursor.UpdatedAt)
	}
}

// TestClaimSharedMaintenanceRun verifies the cluster-wide maintenance claim:
// one winner per interval across all "pods", re-claimable after the interval,
// independent per task name.
func TestClaimSharedMaintenanceRun(t *testing.T) {
	s := newControlStore(t) // ResetMetaDB covers shared_maintenance_state
	ctx := context.Background()

	// First claim wins (row created lazily at the epoch).
	claimed, err := s.ClaimSharedMaintenanceRun(ctx, "fs_events_sweep", 30*time.Minute)
	if err != nil {
		t.Fatalf("claim 1: %v", err)
	}
	if !claimed {
		t.Fatal("first claim should win")
	}

	// An immediate second claim (another pod) loses.
	claimed, err = s.ClaimSharedMaintenanceRun(ctx, "fs_events_sweep", 30*time.Minute)
	if err != nil {
		t.Fatalf("claim 2: %v", err)
	}
	if claimed {
		t.Fatal("second claim inside the interval should lose")
	}

	// A different task name claims independently.
	claimed, err = s.ClaimSharedMaintenanceRun(ctx, "other_task", 30*time.Minute)
	if err != nil {
		t.Fatalf("claim other task: %v", err)
	}
	if !claimed {
		t.Fatal("first claim of a different task should win")
	}

	// After last_run_at ages past the interval, the claim succeeds again.
	if _, err := s.DB().ExecContext(ctx,
		`UPDATE shared_maintenance_state SET last_run_at = DATE_SUB(NOW(3), INTERVAL 1 HOUR) WHERE name = ?`,
		"fs_events_sweep"); err != nil {
		t.Fatal(err)
	}
	claimed, err = s.ClaimSharedMaintenanceRun(ctx, "fs_events_sweep", 30*time.Minute)
	if err != nil {
		t.Fatalf("claim after aging: %v", err)
	}
	if !claimed {
		t.Fatal("claim after the interval should win again")
	}
}

// TestGetDeleteReconcilePending pins the point lookup used by the worker's
// piggyback reconcile gate: found=false with zero time for a missing marker,
// and the stored due time once a marker exists.
func TestGetDeleteReconcilePending(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()

	if _, found, err := s.GetDeleteReconcilePending(ctx, "tenant-none"); err != nil || found {
		t.Fatalf("missing marker: found=%v err=%v, want false/nil", found, err)
	}

	wantDue := now.Add(time.Minute).Truncate(time.Millisecond)
	if err := s.UpsertDeleteReconcilePending(ctx, "tenant-a", wantDue); err != nil {
		t.Fatal(err)
	}
	dueAt, found, err := s.GetDeleteReconcilePending(ctx, "tenant-a")
	if err != nil || !found {
		t.Fatalf("existing marker: found=%v err=%v, want true/nil", found, err)
	}
	if delta := dueAt.Sub(wantDue); delta < -time.Second || delta > time.Second {
		t.Fatalf("due_at = %s, want %s ±1s", dueAt, wantDue)
	}

	if err := s.DeleteDeleteReconcilePending(ctx, "tenant-a"); err != nil {
		t.Fatal(err)
	}
	if _, found, err := s.GetDeleteReconcilePending(ctx, "tenant-a"); err != nil || found {
		t.Fatalf("deleted marker: found=%v err=%v, want false/nil", found, err)
	}
}

// TestRearmDeleteReconcilePending pins the atomic follow-up upsert: a
// missing row is inserted, an owed row (due_at <= cutoff) moves to the new
// due time, and a row due after the cutoff — a newer delete's marker —
// keeps its own due time.
func TestRearmDeleteReconcilePending(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC().Truncate(time.Millisecond)
	cutoff := now
	followUp := now.Add(10 * time.Minute)

	readDue := func(tenantID string) time.Time {
		t.Helper()
		dueAt, found, err := s.GetDeleteReconcilePending(ctx, tenantID)
		if err != nil || !found {
			t.Fatalf("marker for %s: found=%v err=%v", tenantID, found, err)
		}
		return dueAt
	}

	// Absent → inserted with the follow-up due time.
	if err := s.RearmDeleteReconcilePending(ctx, "tenant-absent", followUp, cutoff); err != nil {
		t.Fatal(err)
	}
	if got := readDue("tenant-absent"); !got.Equal(followUp) {
		t.Fatalf("inserted due_at = %s, want %s", got, followUp)
	}

	// Owed (due before the sweep started) → moved to the follow-up time.
	if err := s.UpsertDeleteReconcilePending(ctx, "tenant-owed", now.Add(-time.Minute)); err != nil {
		t.Fatal(err)
	}
	if err := s.RearmDeleteReconcilePending(ctx, "tenant-owed", followUp, cutoff); err != nil {
		t.Fatal(err)
	}
	if got := readDue("tenant-owed"); !got.Equal(followUp) {
		t.Fatalf("owed due_at = %s, want %s", got, followUp)
	}

	// Newer delete's marker (due after the sweep started) → untouched.
	newer := now.Add(time.Hour)
	if err := s.UpsertDeleteReconcilePending(ctx, "tenant-newer", newer); err != nil {
		t.Fatal(err)
	}
	if err := s.RearmDeleteReconcilePending(ctx, "tenant-newer", followUp, cutoff); err != nil {
		t.Fatal(err)
	}
	if got := readDue("tenant-newer"); !got.Equal(newer) {
		t.Fatalf("newer due_at = %s, want %s (untouched)", got, newer)
	}
}

// TestDeleteReconcilePendingUpsertListDelete pins the durable
// delete-reconcile pending marker: one row per tenant, due_at controls when
// ListDeleteReconcilePendingDue returns it, a later upsert moves the due time
// forward, and DeleteDeleteReconcilePending clears it.
func TestDeleteReconcilePendingUpsertListDelete(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now()

	due, err := s.ListDeleteReconcilePendingDue(ctx, now, 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(due) != 0 {
		t.Fatalf("fresh pending list = %v, want empty", due)
	}

	if err := s.UpsertDeleteReconcilePending(ctx, "tenant-a", now.Add(time.Minute)); err != nil {
		t.Fatal(err)
	}
	// Not due yet at the original now.
	due, err = s.ListDeleteReconcilePendingDue(ctx, now, 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(due) != 0 {
		t.Fatalf("pending list before due = %v, want empty", due)
	}
	// Due once the deadline passes.
	due, err = s.ListDeleteReconcilePendingDue(ctx, now.Add(2*time.Minute), 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(due) != 1 || due[0] != "tenant-a" {
		t.Fatalf("pending list after due = %v, want [tenant-a]", due)
	}

	// A newer upsert moves the due time forward (throttle extension).
	if err := s.UpsertDeleteReconcilePending(ctx, "tenant-a", now.Add(2*time.Minute)); err != nil {
		t.Fatal(err)
	}
	// The atomic extension path does the same for an existing row.
	extended, err := s.ExtendDeleteReconcilePending(ctx, "tenant-a", now.Add(3*time.Minute))
	if err != nil {
		t.Fatal(err)
	}
	if !extended {
		t.Fatal("extend on an existing pending row should report true")
	}
	due, err = s.ListDeleteReconcilePendingDue(ctx, now.Add(90*time.Second), 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(due) != 0 {
		t.Fatalf("pending list after extension = %v, want empty", due)
	}
	// Upsert is monotonic too: writing an earlier due time must not shorten an
	// in-flight crash-recovery protection.
	if err := s.UpsertDeleteReconcilePending(ctx, "tenant-a", now.Add(time.Minute)); err != nil {
		t.Fatal(err)
	}
	due, err = s.ListDeleteReconcilePendingDue(ctx, now.Add(2*time.Minute), 10)
	if err != nil {
		t.Fatal(err)
	}
	for _, id := range due {
		if id == "tenant-a" {
			t.Fatalf("upsert pulled a future-dated row earlier: %v", due)
		}
	}
	// Extend is monotonic: an earlier due time must never pull a future-dated
	// crash-recovery marker earlier (another in-flight delete's protection).
	extended, err = s.ExtendDeleteReconcilePending(ctx, "tenant-a", now.Add(time.Minute))
	if err != nil {
		t.Fatal(err)
	}
	if extended {
		t.Fatal("extend with an earlier due time should report false")
	}
	due, err = s.ListDeleteReconcilePendingDue(ctx, now.Add(2*time.Minute), 10)
	if err != nil {
		t.Fatal(err)
	}
	for _, id := range due {
		if id == "tenant-a" {
			t.Fatalf("extend pulled a future-dated row earlier: %v", due)
		}
	}
	// Extend never inserts: a missing row reports false and stays missing.
	extended, err = s.ExtendDeleteReconcilePending(ctx, "never-existed", now.Add(-time.Minute))
	if err != nil {
		t.Fatal(err)
	}
	if extended {
		t.Fatal("extend on a missing pending row should report false")
	}
	due, err = s.ListDeleteReconcilePendingDue(ctx, now, 10)
	if err != nil {
		t.Fatal(err)
	}
	for _, id := range due {
		if id == "never-existed" {
			t.Fatalf("extend inserted a row for a missing tenant: %v", due)
		}
	}

	// InsertIfAbsent creates a missing row but never overwrites an existing
	// due time (a throttled worker's extension must survive a late dispatch).
	if err := s.InsertDeleteReconcilePendingIfAbsent(ctx, "tenant-a", now.Add(-time.Minute)); err != nil {
		t.Fatal(err)
	}
	due, err = s.ListDeleteReconcilePendingDue(ctx, now, 10)
	if err != nil {
		t.Fatal(err)
	}
	foundA := false
	for _, id := range due {
		if id == "tenant-a" {
			foundA = true
		}
	}
	if foundA {
		t.Fatalf("insert-if-absent overwrote an existing due time: %v", due)
	}
	if err := s.InsertDeleteReconcilePendingIfAbsent(ctx, "tenant-d", now.Add(time.Minute)); err != nil {
		t.Fatal(err)
	}
	due, err = s.ListDeleteReconcilePendingDue(ctx, now.Add(2*time.Minute), 10)
	if err != nil {
		t.Fatal(err)
	}
	foundD := false
	for _, id := range due {
		if id == "tenant-d" {
			foundD = true
		}
	}
	if !foundD {
		t.Fatalf("insert-if-absent did not create a missing row: %v", due)
	}

	// Refresh creates a missing row with the given due time, moves an EXPIRED
	// row's due time forward, and never pulls a future-dated row earlier (it
	// must not strip another in-flight delete's protection).
	if err := s.RefreshDeleteReconcilePending(ctx, "tenant-new", now); err != nil {
		t.Fatal(err)
	}
	// DATETIME(3) rounds to the millisecond, so use a small query window past
	// `now` when asserting a row refreshed to `now` is due.
	due, err = s.ListDeleteReconcilePendingDue(ctx, now.Add(2*time.Millisecond), 10)
	if err != nil {
		t.Fatal(err)
	}
	foundNew := false
	for _, id := range due {
		if id == "tenant-new" {
			foundNew = true
		}
	}
	if !foundNew {
		t.Fatalf("refresh did not create a missing row: %v", due)
	}
	// tenant-a's row is future-dated (+3min): refresh with now must NOT pull it
	// earlier.
	if err := s.RefreshDeleteReconcilePending(ctx, "tenant-a", now); err != nil {
		t.Fatal(err)
	}
	due, err = s.ListDeleteReconcilePendingDue(ctx, now.Add(2*time.Millisecond), 10)
	if err != nil {
		t.Fatal(err)
	}
	for _, id := range due {
		if id == "tenant-a" {
			t.Fatalf("refresh pulled a future-dated row earlier: %v", due)
		}
	}
	// An expired row is refreshed to the new due time.
	if err := s.UpsertDeleteReconcilePending(ctx, "tenant-e", now.Add(-time.Minute)); err != nil {
		t.Fatal(err)
	}
	if err := s.RefreshDeleteReconcilePending(ctx, "tenant-e", now); err != nil {
		t.Fatal(err)
	}
	due, err = s.ListDeleteReconcilePendingDue(ctx, now.Add(2*time.Millisecond), 10)
	if err != nil {
		t.Fatal(err)
	}
	foundE := false
	for _, id := range due {
		if id == "tenant-e" {
			foundE = true
		}
	}
	if !foundE {
		t.Fatalf("refresh did not move the expired row to the new due time: %v", due)
	}

	// Successful sweep clears it.
	if err := s.DeleteDeleteReconcilePending(ctx, "tenant-a"); err != nil {
		t.Fatal(err)
	}
	due, err = s.ListDeleteReconcilePendingDue(ctx, now.Add(time.Hour), 10)
	if err != nil {
		t.Fatal(err)
	}
	for _, id := range due {
		if id == "tenant-a" {
			t.Fatalf("pending marker for tenant-a remains after delete: %v", due)
		}
	}

	// Two tenants are independent.
	if err := s.UpsertDeleteReconcilePending(ctx, "tenant-b", now.Add(-time.Minute)); err != nil {
		t.Fatal(err)
	}
	due, err = s.ListDeleteReconcilePendingDue(ctx, now, 10)
	if err != nil {
		t.Fatal(err)
	}
	foundB := false
	for _, id := range due {
		if id == "tenant-b" {
			foundB = true
		}
	}
	if !foundB {
		t.Fatalf("tenant-b missing from pending list: %v", due)
	}

	// The cutoff clear only removes markers due no later than the cutoff: a
	// row written mid-sweep by a newer delete must survive.
	if err := s.UpsertDeleteReconcilePending(ctx, "tenant-c", now.Add(time.Minute)); err != nil {
		t.Fatal(err)
	}
	if err := s.DeleteDeleteReconcilePendingBefore(ctx, "tenant-c", now); err != nil {
		t.Fatal(err)
	}
	due, err = s.ListDeleteReconcilePendingDue(ctx, now.Add(2*time.Minute), 10)
	if err != nil {
		t.Fatal(err)
	}
	foundC := false
	for _, id := range due {
		if id == "tenant-c" {
			foundC = true
		}
	}
	if !foundC {
		t.Fatal("cutoff clear removed a future-dated pending row")
	}
	if err := s.DeleteDeleteReconcilePendingBefore(ctx, "tenant-c", now.Add(2*time.Minute)); err != nil {
		t.Fatal(err)
	}
	due, err = s.ListDeleteReconcilePendingDue(ctx, now.Add(2*time.Minute), 10)
	if err != nil {
		t.Fatal(err)
	}
	for _, id := range due {
		if id == "tenant-c" {
			t.Fatalf("cutoff clear did not remove the due pending row: %v", due)
		}
	}
}