package server

import (
	"context"
	"database/sql"
	"errors"
	"testing"
	"time"

	"go.uber.org/zap"

	"github.com/mem9-ai/drive9/pkg/backend"
	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/tenant/schema"
	"github.com/mem9-ai/drive9/pkg/testtidb"
)

// initWorkerReconcileSchema initializes the split tenant schema (inodes,
// contents, file_nodes, file_gc_tasks, ...) on the shared test TiDB and
// resets all tables, so a real datastore + backend can be exercised.
func initWorkerReconcileSchema(t *testing.T) *sql.DB {
	t.Helper()
	db := testtidb.OpenDB(t, testDSN)
	stmts := schema.TiDBAppEmbeddingTenantSchemaStatements()
	if err := schema.ExecSchemaStatements(db, stmts); err != nil {
		t.Fatalf("init tenant schema: %v", err)
	}
	testtidb.ResetDB(t, db)
	return db
}

// plantBrokenDentry plants the exact residual state the reconciliation sweep
// must repair: a file dentry whose parent directory row does not exist.
func plantBrokenDentry(t *testing.T, store *datastore.Store) {
	t.Helper()
	now := time.Now().UTC()
	if err := store.InsertFile(context.Background(), &datastore.File{
		FileID: "f-orphan", StorageType: datastore.StorageDB9, StorageRef: "/blobs/f-orphan",
		SizeBytes: 1, Revision: 1, Status: datastore.StatusConfirmed, CreatedAt: now, ConfirmedAt: &now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := store.InsertNode(context.Background(), &datastore.FileNode{
		NodeID: "n-orphan", Path: "/ghost/a.txt", ParentPath: "/ghost/",
		Name: "a.txt", FileID: "f-orphan", CreatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
}

// newWorkerReconcileManager builds a fallback-only tenant worker manager whose
// piggyback maintenance is throttled to an hour and whose maintenance slot is
// pre-seeded to "just ran", so any repair observed in these tests MUST have
// come from the WorkDeleteReconcile path (requestDeleteReconcile), not from
// the maintenance cadence.
func newWorkerReconcileManager(t *testing.T, b *backend.Dat9Backend) *tenantWorkerManager {
	t.Helper()
	m := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{MaintenanceInterval: time.Hour}, 0)
	if m == nil {
		t.Fatal("tenant worker manager not viable with a fallback backend")
	}
	m.mu.Lock()
	m.lastMaintenance[tenantLocalID] = time.Now()
	m.mu.Unlock()
	return m
}

func waitWorkerRepair(t *testing.T, store *datastore.Store, timeout time.Duration) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		ctx := context.Background()
		if _, err := store.GetNode(ctx, "/ghost/a.txt"); errors.Is(err, datastore.ErrNotFound) {
			if _, err := store.GetFileGCTaskByFileID(ctx, "f-orphan"); err == nil {
				return
			}
		}
		time.Sleep(20 * time.Millisecond)
	}
	t.Fatal("residual orphan was not repaired within timeout")
}

// TestWorkerDeleteReconcileKickRunsRepair is the consumer-side regression for
// gate 4: a WorkDeleteReconcile kick must run the delete-orphan
// reconciliation sweep through the real worker path (processKicked →
// requestDeleteReconcile → runDeleteReconcile), even when piggyback
// maintenance is throttled, and without the test calling
// ReconcileDeleteOrphans directly.
func TestWorkerDeleteReconcileKickRunsRepair(t *testing.T) {
	oldInterval := tenantDeleteReconcileInterval
	tenantDeleteReconcileInterval = time.Nanosecond
	t.Cleanup(func() { tenantDeleteReconcileInterval = oldInterval })

	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	plantBrokenDentry(t, store)

	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}
	m := newWorkerReconcileManager(t, b)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	m.Start(ctx)
	m.KickWithOrg(tenantLocalID, "", WorkDeleteReconcile)

	waitWorkerRepair(t, store, 10*time.Second)
	m.Stop()
}

// TestWorkerDeleteReconcileKickReKicksAfterThrottle pins the delayed re-kick
// path: when an explicit WorkDeleteReconcile kick arrives inside the
// per-tenant reconcile throttle window, the manager must NOT silently drop
// the request — it arms one delayed re-kick that repairs the residual orphan
// after the window opens, even though the tenant has no other activity.
func TestWorkerDeleteReconcileKickReKicksAfterThrottle(t *testing.T) {
	oldInterval := tenantDeleteReconcileInterval
	tenantDeleteReconcileInterval = 150 * time.Millisecond
	t.Cleanup(func() { tenantDeleteReconcileInterval = oldInterval })

	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	plantBrokenDentry(t, store)

	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}
	m := newWorkerReconcileManager(t, b)
	// Reconcile "just ran": the explicit kick must be throttled and re-kicked.
	m.mu.Lock()
	m.lastReconcile[tenantLocalID] = time.Now()
	m.mu.Unlock()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	m.Start(ctx)
	m.KickWithOrg(tenantLocalID, "", WorkDeleteReconcile)

	// The throttle must hold: no immediate repair.
	time.Sleep(50 * time.Millisecond)
	if _, err := store.GetNode(context.Background(), "/ghost/a.txt"); err != nil {
		t.Fatalf("orphan repaired before the throttle window elapsed: %v", err)
	}

	// The armed delayed re-kick must repair it once the window opens.
	waitWorkerRepair(t, store, 10*time.Second)
	m.Stop()
}

// TestWorkerDeleteReconcileDurableAcrossRestart is the gate-4 restart
// regression: the durable WorkDeleteReconcile outbox row is consumed once, the
// worker's reconcile is throttled (so the request lives only in the durable
// pending marker, not the process-local timer), the worker/poller are then
// recreated before any in-process timer fires, and the broken-chain orphan is
// still repaired automatically — without any unrelated tenant work — by the
// new poller's pending scan.
func TestWorkerDeleteReconcileDurableAcrossRestart(t *testing.T) {
	oldInterval := tenantDeleteReconcileInterval
	tenantDeleteReconcileInterval = time.Second
	t.Cleanup(func() { tenantDeleteReconcileInterval = oldInterval })

	initWorkerReconcileSchema(t)
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())

	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	plantBrokenDentry(t, store)
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}

	ctx := context.Background()

	// Producer: the durable outbox signal.
	if err := metaStore.InsertTenantNotify(ctx, tenantLocalID, WorkDeleteReconcile); err != nil {
		t.Fatal(err)
	}
	rows, err := metaStore.ListTenantNotifySince(ctx, 0, 10)
	if err != nil {
		t.Fatal(err)
	}
	var signal *meta.TenantNotifyRow
	for i := range rows {
		if rows[i].TenantID == tenantLocalID {
			signal = &rows[i]
			break
		}
	}
	if signal == nil {
		t.Fatal("WorkDeleteReconcile outbox row not found")
	}

	// First "process": poller consumes the durable signal and makes it durable
	// in the pending table, then the worker is throttled and extends the due
	// time. The in-process timer is armed but this process is then abandoned.
	m1 := newTenantWorkerManager(b, metaStore, nil, nil, TenantWorkerOptions{MaintenanceInterval: time.Hour}, 0)
	m1.mu.Lock()
	m1.lastMaintenance[tenantLocalID] = time.Now()
	m1.lastReconcile[tenantLocalID] = time.Now()
	m1.mu.Unlock()
	p1 := newTenantOutboxPoller(metaStore, newEventBuses(), &mockKicker{}, nil, "pod1", 0, 0)
	if err := p1.dispatch(ctx, *signal); err != nil {
		t.Fatalf("dispatch durable signal: %v", err)
	}
	m1.processKicked(ctx, tenantLocalID, "", WorkDeleteReconcile)

	// The pending marker must exist and must NOT be due yet (throttle window).
	due, err := metaStore.ListDeleteReconcilePendingDue(ctx, time.Now(), 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(due) != 0 {
		t.Fatalf("pending reconcile unexpectedly due after throttle: %v", due)
	}
	var pendingCount int
	if err := metaStore.DB().QueryRow(`SELECT COUNT(*) FROM delete_reconcile_pending WHERE tenant_id = ?`,
		tenantLocalID).Scan(&pendingCount); err != nil {
		t.Fatal(err)
	}
	if pendingCount != 1 {
		t.Fatalf("pending reconcile rows = %d, want 1", pendingCount)
	}

	// "Restart": abandon m1/p1 (its in-process timer is gone) and start a fresh
	// worker + poller. No unrelated tenant work is injected.
	m2 := newTenantWorkerManager(b, metaStore, nil, nil, TenantWorkerOptions{MaintenanceInterval: time.Hour}, 0)
	m2Ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	m2.Start(m2Ctx)
	p2 := newTenantOutboxPoller(metaStore, newEventBuses(), m2, nil, "pod2", 0, 0)

	// Wait for the durable due time to pass, then the new poller's pending
	// scan re-dispatches the request to the new worker.
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		due, err = metaStore.ListDeleteReconcilePendingDue(ctx, time.Now(), 10)
		if err != nil {
			t.Fatal(err)
		}
		if len(due) != 0 {
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	if len(due) == 0 {
		t.Fatal("pending reconcile never became due")
	}
	p2.scanDeleteReconcilePending(ctx)

	waitWorkerRepair(t, store, 10*time.Second)
	// The successful sweep must also clear the durable marker. Poll instead of
	// stopping immediately: the marker is deleted after the repair commits, so
	// stopping the worker mid-clear would cancel the delete (safe — the row
	// would be re-kicked later — but not what this test asserts).
	clearDeadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(clearDeadline) {
		due, err = metaStore.ListDeleteReconcilePendingDue(ctx, time.Now().Add(time.Hour), 10)
		if err != nil {
			t.Fatal(err)
		}
		if len(due) == 0 {
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	m2.Stop()
	if len(due) != 0 {
		t.Fatalf("pending reconcile remains after successful repair: %v", due)
	}
}

// TestWorkerDeleteReconcileKickBypassesFutureDatedMarker pins kick
// immediacy with a meta store wired: the marker left by RemoveAllCtx's
// fail-closed pre-write is future-dated (deleteReconcilePersistLeadTime)
// and RefreshDeleteReconcilePending never pulls it earlier, so post-delete
// repair depends on the WorkDeleteReconcile kick path
// (requestDeleteReconcile → runDeleteReconcile) bypassing the piggyback
// marker gate. A kick arriving while the durable marker is still
// future-dated must repair the orphan immediately.
func TestWorkerDeleteReconcileKickBypassesFutureDatedMarker(t *testing.T) {
	initWorkerReconcileSchema(t)
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())

	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	plantBrokenDentry(t, store)
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}

	// Future-dated crash-recovery marker: the piggyback gate must skip it,
	// the kick path must not wait for it.
	if err := metaStore.UpsertDeleteReconcilePending(context.Background(), tenantLocalID, time.Now().Add(time.Hour)); err != nil {
		t.Fatal(err)
	}
	m := newTenantWorkerManager(b, metaStore, nil, nil, TenantWorkerOptions{MaintenanceInterval: time.Hour}, 0)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	m.Start(ctx)
	m.KickWithOrg(tenantLocalID, "", WorkDeleteReconcile)

	waitWorkerRepair(t, store, 10*time.Second)
	m.Stop()
}

// TestWorkerDeleteReconcilePiggybackRequiresPendingMarker pins the piggyback
// backstop gate: with a meta store wired, the periodic reconcile round must
// skip tenants with no durable pending marker (a full-table scan for a
// tenant with no outstanding delete is pure cost), must skip a future-dated
// marker (an in-flight delete's crash-recovery protection — the poller
// re-dispatches it at due time), and must run once the marker is due.
func TestWorkerDeleteReconcilePiggybackRequiresPendingMarker(t *testing.T) {
	initWorkerReconcileSchema(t)
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())

	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	plantBrokenDentry(t, store)
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}

	m := newTenantWorkerManager(b, metaStore, nil, nil, TenantWorkerOptions{MaintenanceInterval: time.Hour}, 0)
	ctx := context.Background()
	target := &tenantTarget{tenantID: tenantLocalID, backend: b, store: store, release: func() {}}

	orphanPresent := func() bool {
		_, err := store.GetNode(ctx, "/ghost/a.txt")
		return err == nil
	}

	// No marker: the piggyback round must not scan, even though the throttle
	// has never run (lastReconcile zero → due).
	m.reconcileDeleteOrphans(ctx, target)
	if !orphanPresent() {
		t.Fatal("piggyback reconcile ran without a pending marker")
	}

	// Future-dated marker (in-flight delete protection): still skipped.
	if err := metaStore.UpsertDeleteReconcilePending(ctx, tenantLocalID, time.Now().Add(time.Hour)); err != nil {
		t.Fatal(err)
	}
	m.reconcileDeleteOrphans(ctx, target)
	if !orphanPresent() {
		t.Fatal("piggyback reconcile ran on a future-dated pending marker")
	}

	// Due marker: the backstop runs and repairs the orphan.
	if err := metaStore.DeleteDeleteReconcilePending(ctx, tenantLocalID); err != nil {
		t.Fatal(err)
	}
	if err := metaStore.UpsertDeleteReconcilePending(ctx, tenantLocalID, time.Now()); err != nil {
		t.Fatal(err)
	}
	m.reconcileDeleteOrphans(ctx, target)
	if orphanPresent() {
		t.Fatal("piggyback reconcile did not run on a due pending marker")
	}
	if _, err := store.GetFileGCTaskByFileID(ctx, "f-orphan"); err != nil {
		t.Fatalf("gc task for f-orphan: %v", err)
	}
}

// TestWorkerDeleteReconcileProducerPersistsBeforeOutboxFlush is the
// producer-side durability regression: a recursive delete must durably record
// the pending reconcile BEFORE any notify-coalescer flush or in-process kick
// handling, so a crash between the delete commit and the outbox flush cannot
// lose the request. It then simulates a restart with a fresh worker + poller
// (no unrelated work) and proves the orphan is repaired from the producer-side
// marker alone.
func TestWorkerDeleteReconcileProducerPersistsBeforeOutboxFlush(t *testing.T) {
	oldInterval := tenantDeleteReconcileInterval
	tenantDeleteReconcileInterval = time.Second
	t.Cleanup(func() { tenantDeleteReconcileInterval = oldInterval })

	initWorkerReconcileSchema(t)
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())

	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	plantBrokenDentry(t, store)
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}

	// The notifier is wired exactly like Server.notifyWorkEnqueued (post-delete
	// refresh: with no pre-existing row it creates the marker at due=now). m1 is
	// never started, so no worker handling and no timer processing can be the
	// source of durability. The fail-closed PRE-delete guarantee itself is
	// covered by TestRemoveAllCtxDeleteReconcilePersistFailClosed (backend) and
	// TestWorkerDeleteReconcileMarkerSurvivesCrashBeforeNotifier (server).
	m1 := newTenantWorkerManager(b, metaStore, nil, nil, TenantWorkerOptions{MaintenanceInterval: time.Hour}, 0)
	m1.mu.Lock()
	m1.lastMaintenance[tenantLocalID] = time.Now()
	m1.lastReconcile[tenantLocalID] = time.Now()
	m1.mu.Unlock()
	srv := &Server{meta: metaStore, tenantWorker: m1}
	b.SetWorkEnqueuedNotifier(func(workMask int, tidbCloudOrgID string) {
		srv.notifyWorkEnqueued(tenantLocalID, workMask, tidbCloudOrgID)
	})

	ctx := context.Background()
	if err := b.Mkdir("/del/", 0o755); err != nil {
		t.Fatal(err)
	}
	if err := b.RemoveAll("/del/"); err != nil {
		t.Fatal(err)
	}

	// The durable marker exists immediately after the delete, before any
	// outbox flush or worker processing.
	var pendingCount int
	if err := metaStore.DB().QueryRow(`SELECT COUNT(*) FROM delete_reconcile_pending WHERE tenant_id = ?`,
		tenantLocalID).Scan(&pendingCount); err != nil {
		t.Fatal(err)
	}
	if pendingCount != 1 {
		t.Fatalf("producer-side pending rows = %d, want 1 before any outbox flush", pendingCount)
	}

	// "Crash": abandon m1/srv (no coalescer flush, no timer handling) and
	// restart with a fresh worker + poller.
	m2 := newTenantWorkerManager(b, metaStore, nil, nil, TenantWorkerOptions{MaintenanceInterval: time.Hour}, 0)
	m2Ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	m2.Start(m2Ctx)
	p2 := newTenantOutboxPoller(metaStore, newEventBuses(), m2, nil, "pod2", 0, 0)
	p2.scanDeleteReconcilePending(ctx)

	waitWorkerRepair(t, store, 10*time.Second)
	m2.Stop()
}

// TestWorkerDeleteReconcilePendingPersistsOnNonOwnerDispatch is the shard
// failover regression: a NON-owner pod that reads a WorkDeleteReconcile row
// must still record the durable pending marker (shard ownership gates the
// kick, never whether durability is recorded). After ownership moves to a
// fresh owner, the pending scan repairs the orphan without the original owner
// or any unrelated work.
func TestWorkerDeleteReconcilePendingPersistsOnNonOwnerDispatch(t *testing.T) {
	initWorkerReconcileSchema(t)
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())

	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	plantBrokenDentry(t, store)
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}

	ctx := context.Background()
	if err := metaStore.InsertTenantNotify(ctx, tenantLocalID, WorkDeleteReconcile); err != nil {
		t.Fatal(err)
	}

	// Non-owner pod: reads and dispatches the row, records durability, does
	// NOT kick.
	k := &mockKicker{}
	pNonOwner := newTenantOutboxPoller(metaStore, newEventBuses(), k, func(string) bool { return false }, "pod-a", 0, 0)
	pNonOwner.pollOnce(ctx)
	if len(k.kicks) != 0 {
		t.Fatalf("non-owner pod must not kick, got %d kicks", len(k.kicks))
	}
	var pendingCount int
	if err := metaStore.DB().QueryRow(`SELECT COUNT(*) FROM delete_reconcile_pending WHERE tenant_id = ?`,
		tenantLocalID).Scan(&pendingCount); err != nil {
		t.Fatal(err)
	}
	if pendingCount != 1 {
		t.Fatalf("non-owner dispatch did not persist the pending marker: %d rows", pendingCount)
	}

	// Ownership moves to a fresh pod; its pending scan recovers and repairs.
	m2 := newTenantWorkerManager(b, metaStore, nil, nil, TenantWorkerOptions{MaintenanceInterval: time.Hour}, 0)
	m2Ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	m2.Start(m2Ctx)
	p2 := newTenantOutboxPoller(metaStore, newEventBuses(), m2, nil, "pod-b", 0, 0)
	p2.scanDeleteReconcilePending(ctx)

	waitWorkerRepair(t, store, 10*time.Second)
	m2.Stop()
}

// TestWorkerDeleteReconcileMarkerSurvivesCrashBeforeNotifier is the
// fail-closed producer regression: RemoveAllCtx's durable pre-write (backend
// persister) runs BEFORE the delete, and the notifier is a no-op simulating a
// crash after the delete commits but before notifier/coalescer/worker
// handling. The future-dated marker alone must survive the restart and drive
// the repair — no outbox row and no in-process processing exists.
func TestWorkerDeleteReconcileMarkerSurvivesCrashBeforeNotifier(t *testing.T) {
	oldLead := deleteReconcilePersistLeadTime
	deleteReconcilePersistLeadTime = 150 * time.Millisecond
	t.Cleanup(func() { deleteReconcilePersistLeadTime = oldLead })

	initWorkerReconcileSchema(t)
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())

	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	plantBrokenDentry(t, store)
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}

	ctx := context.Background()
	// Fail-closed durable pre-write wired exactly like the server does, and a
	// NO-OP notifier: the crash happens before any outbox/coalescer/worker
	// state exists.
	var persistedAt time.Time
	b.SetDeleteReconcilePersister(func() error {
		persistedAt = time.Now()
		return metaStore.UpsertDeleteReconcilePending(ctx, tenantLocalID,
			persistedAt.Add(deleteReconcilePersistLeadTime))
	})
	b.SetWorkEnqueuedNotifier(func(workMask int, tidbCloudOrgID string) {})

	if err := b.Mkdir("/del/", 0o755); err != nil {
		t.Fatal(err)
	}
	if err := b.RemoveAll("/del/"); err != nil {
		t.Fatal(err)
	}

	// The marker was written as future-dated crash-recovery (no outbox).
	// Do not use ListDue(now) here: RemoveAll after the persist can take
	// longer than the lead on a loaded runner, so the row may already be
	// due by the time we observe it.
	if persistedAt.IsZero() {
		t.Fatal("persister did not run")
	}
	var dueAt time.Time
	if err := metaStore.DB().QueryRow(`SELECT due_at FROM delete_reconcile_pending WHERE tenant_id = ?`,
		tenantLocalID).Scan(&dueAt); err != nil {
		t.Fatal(err)
	}
	wantDue := persistedAt.UTC().Add(deleteReconcilePersistLeadTime)
	if delta := dueAt.Sub(wantDue); delta < -50*time.Millisecond || delta > 50*time.Millisecond {
		t.Fatalf("pre-delete due_at=%s, want %s ±50ms", dueAt.UTC(), wantDue)
	}
	var pendingCount int
	if err := metaStore.DB().QueryRow(`SELECT COUNT(*) FROM delete_reconcile_pending WHERE tenant_id = ?`,
		tenantLocalID).Scan(&pendingCount); err != nil {
		t.Fatal(err)
	}
	if pendingCount != 1 {
		t.Fatalf("pre-delete pending rows = %d, want 1", pendingCount)
	}

	// "Restart": fresh worker + poller; wait for the lead to pass so the
	// marker becomes due, then the pending scan repairs without unrelated work.
	m2 := newTenantWorkerManager(b, metaStore, nil, nil, TenantWorkerOptions{MaintenanceInterval: time.Hour}, 0)
	m2Ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	m2.Start(m2Ctx)
	p2 := newTenantOutboxPoller(metaStore, newEventBuses(), m2, nil, "pod2", 0, 0)

	deadline := time.Now().Add(5 * time.Second)
	var due []string
	for time.Now().Before(deadline) {
		var err error
		due, err = metaStore.ListDeleteReconcilePendingDue(ctx, time.Now(), 10)
		if err != nil {
			t.Fatal(err)
		}
		if len(due) != 0 {
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	if len(due) == 0 {
		t.Fatal("pre-delete marker never became due")
	}
	p2.scanDeleteReconcilePending(ctx)

	waitWorkerRepair(t, store, 10*time.Second)
	m2.Stop()
}

// TestServerSingleTenantDeleteReconcileCrashRecovery is the drive9-server-local
// regression: a real Server with Backend + Meta (no Pool) wires the fail-closed
// delete-reconcile persister, commits a recursive delete, then crashes before
// the post-delete notifier/coalescer/worker can process anything. A fresh Server
// restart with no unrelated traffic discovers the durable pending marker through
// the outbox poller's scan and repairs the broken-chain orphan.
func TestServerSingleTenantDeleteReconcileCrashRecovery(t *testing.T) {
	oldLead := deleteReconcilePersistLeadTime
	deleteReconcilePersistLeadTime = 150 * time.Millisecond
	t.Cleanup(func() { deleteReconcilePersistLeadTime = oldLead })
	// Shrink the reconcile throttle so the post-repair follow-up round (the
	// convergence re-arm re-run that finds nothing and clears the marker it
	// re-created) completes within the marker-clear assertion window.
	oldInterval := tenantDeleteReconcileInterval
	tenantDeleteReconcileInterval = time.Second
	t.Cleanup(func() { tenantDeleteReconcileInterval = oldInterval })

	initWorkerReconcileSchema(t)
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())

	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	// Stop the process-global mutation dispatcher before the DB handles close
	// so its workers don't use a closed store.
	t.Cleanup(func() { backend.StopMutationDispatcher() })
	plantBrokenDentry(t, store)

	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}

	ctx := context.Background()

	// Start a server exactly like drive9-server-local does: Backend + Meta,
	// no Pool. This wires the real persister and notifier/coalescer.
	srv1 := NewWithConfig(Config{
		Backend: b,
		Meta:    metaStore,
		Logger:  zap.NewNop(),
		// Use a very long cursor flush so no periodic scanDeleteReconcilePending
		// can fire between RemoveAll and srv1.Close. The crash window must not
		// repair the orphan in-process.
		TenantOutboxCursorFlushInterval: time.Hour,
	})

	// Simulate the crash: after the delete commits but before the post-delete
	// notifier, coalescer, or worker can run, the work notifier is gone. The
	// fail-closed pre-delete persister has already written the durable marker.
	b.SetWorkEnqueuedNotifier(func(workMask int, tidbCloudOrgID string) {})

	if err := b.Mkdir("/del/", 0o755); err != nil {
		t.Fatal(err)
	}
	markerWriteTime := time.Now()
	if err := b.RemoveAll("/del/"); err != nil {
		t.Fatal(err)
	}

	// The marker exists and is future-dated; no outbox row was produced.
	var pendingCount int
	if err := metaStore.DB().QueryRow(`SELECT COUNT(*) FROM delete_reconcile_pending WHERE tenant_id = ?`,
		tenantLocalID).Scan(&pendingCount); err != nil {
		t.Fatal(err)
	}
	if pendingCount != 1 {
		t.Fatalf("pre-delete pending rows = %d, want 1", pendingCount)
	}
	var dueAt time.Time
	if err := metaStore.DB().QueryRow(`SELECT due_at FROM delete_reconcile_pending WHERE tenant_id = ?`,
		tenantLocalID).Scan(&dueAt); err != nil {
		t.Fatal(err)
	}
	if !dueAt.After(markerWriteTime) {
		t.Fatalf("pre-delete marker due_at %v not after write time %v", dueAt, markerWriteTime)
	}
	rows, err := metaStore.ListTenantNotifySince(ctx, 0, 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(rows) != 0 {
		t.Fatalf("unexpected outbox rows before crash: %v", rows)
	}

	// Crash: tear down the first server without letting it process the kick.
	srv1.Close()
	backend.StopMutationDispatcher()

	// The orphan must still be present after the crash; otherwise the repair
	// happened in-process rather than on restart.
	if _, err := store.GetNode(ctx, "/ghost/a.txt"); errors.Is(err, datastore.ErrNotFound) {
		t.Fatal("orphan /ghost/a.txt was repaired before restart")
	} else if err != nil {
		t.Fatalf("get orphan after crash: %v", err)
	}

	// Restart: a brand-new Server with the same backend/meta. No unrelated
	// tenant work is injected; the only durable hint is the pending marker.
	srv2 := NewWithConfig(Config{
		Backend:                         b,
		Meta:                            metaStore,
		Logger:                          zap.NewNop(),
		TenantOutboxCursorFlushInterval: 200 * time.Millisecond,
	})
	defer srv2.Close()

	waitWorkerRepair(t, store, 10*time.Second)

	// The successful sweep must also clear the durable marker.
	var due []string
	clearDeadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(clearDeadline) {
		due, err = metaStore.ListDeleteReconcilePendingDue(ctx, time.Now().Add(time.Hour), 10)
		if err != nil {
			t.Fatal(err)
		}
		if len(due) == 0 {
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	if len(due) != 0 {
		t.Fatalf("pending reconcile remains after successful repair: %v", due)
	}
}