//go:build failpoint

package backend

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/pingcap/failpoint"

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

// TestRemoveAllZeroOrphanResidualSchedulesReconcile is the producer-side
// regression for the residual-window gate: a successful recursive delete of a
// tree with no files leaves a broken-chain child via the §4.2 commit-boundary
// window while OrphansEnqueued == 0, and the backend still emits the
// delete-reconcile work kick. The consumer side (kick → processKicked →
// requestDeleteReconcile → repair) is covered by
// pkg/server/delete_reconcile_worker_test.go; this test additionally proves
// the residual state is repairable through ReconcileDeleteOrphans on the same
// store.
func TestRemoveAllZeroOrphanResidualSchedulesReconcile(t *testing.T) {
	b := newTestBackend(t)
	ctx := context.Background()
	now := time.Now()

	// 1025 empty subdirectories force the sweep (default sweep batch is 1000;
	// /z/d1024/ sorts last and is deleted in the final drain batch of /z/).
	insertBulkDirFixture(t, b, 1025)

	var reconcileKicks atomic.Int32
	b.SetWorkEnqueuedNotifier(func(workMask int, _ string) {
		if workMask&BackendWorkDeleteReconcile != 0 {
			reconcileKicks.Add(1)
		}
	})

	hit := make(chan struct{})
	release := make(chan struct{})
	var fired atomic.Bool
	if err := failpoint.EnableCall("github.com/mem9-ai/drive9/pkg/datastore/drainEmptyDirBeforeRecheck", func(dirPath string) {
		if dirPath != "/z/d1024/" || !fired.CompareAndSwap(false, true) {
			return
		}
		close(hit)
		<-release
	}); err != nil {
		t.Fatalf("enable failpoint: %v", err)
	}
	t.Cleanup(func() {
		_ = failpoint.Disable("github.com/mem9-ai/drive9/pkg/datastore/drainEmptyDirBeforeRecheck")
	})

	deleteDone := make(chan error, 1)
	go func() { deleteDone <- b.RemoveAll("/z/") }()

	select {
	case <-hit:
	case <-time.After(30 * time.Second):
		close(release)
		t.Fatal("empty-dir recheck failpoint never fired")
	}

	// Real create path: the file identity, then EnsureParentDirs sees /z/d1024/
	// alive (uncommitted delete), then the dentry INSERT blocks on the paused
	// batch's child-range gap lock and commits after it — the residual window.
	if err := b.store.InsertFile(ctx, &datastore.File{
		FileID: "f-race3", StorageType: datastore.StorageDB9, StorageRef: "/blobs/f-race3",
		SizeBytes: 1, Revision: 1, Status: datastore.StatusConfirmed, CreatedAt: now, ConfirmedAt: &now,
	}); err != nil {
		close(release)
		t.Fatalf("insert racing inode: %v", err)
	}
	if err := b.store.EnsureParentDirs(ctx, "/z/d1024/zz-new.txt", b.genID); err != nil {
		close(release)
		t.Fatalf("ensure parents: %v", err)
	}
	insertDone := make(chan error, 1)
	go func() {
		insertDone <- b.store.InsertNode(ctx, &datastore.FileNode{
			NodeID: "n-race3", Path: "/z/d1024/zz-new.txt", ParentPath: "/z/d1024/", Name: "zz-new.txt",
			FileID: "f-race3", CreatedAt: now,
		})
	}()
	close(release)

	if err := <-deleteDone; err != nil {
		t.Fatalf("RemoveAll: %v", err)
	}
	select {
	case err := <-insertDone:
		if err != nil {
			t.Fatalf("racing insert: %v", err)
		}
	case <-time.After(30 * time.Second):
		t.Fatal("racing insert still blocked after RemoveAll returned")
	}

	// The delete itself enqueued zero GC tasks, yet the scheduling path still
	// emitted the reconcile kick (the repair pass must not depend on
	// OrphansEnqueued > 0 or unrelated future work).
	if got := reconcileKicks.Load(); got != 1 {
		t.Fatalf("delete-reconcile kicks = %d, want 1", got)
	}
	if _, err := b.store.GetNode(ctx, "/z/"); !errors.Is(err, datastore.ErrNotFound) {
		t.Fatalf("root /z/ still resolves after RemoveAll: %v", err)
	}
	if _, err := b.store.GetNode(ctx, "/z/d1024/zz-new.txt"); err != nil {
		t.Fatalf("broken-chain child should be alive before reconciliation: %v", err)
	}
	if _, err := b.store.GetFileGCTaskByFileID(ctx, "f-race3"); !errors.Is(err, datastore.ErrNotFound) {
		t.Fatalf("gc task for f-race3 before reconciliation: err = %v, want ErrNotFound", err)
	}

	// The scheduled repair pass actually fixes the residual orphan.
	report, err := b.store.ReconcileDeleteOrphans(ctx, false, 100, nil)
	if err != nil {
		t.Fatalf("ReconcileDeleteOrphans: %v", err)
	}
	if report.BrokenDentries != 1 || report.Repaired != 1 {
		t.Fatalf("reconcile report = %+v, want exactly 1 broken dentry repaired", report)
	}
	if _, err := b.store.GetNode(ctx, "/z/d1024/zz-new.txt"); !errors.Is(err, datastore.ErrNotFound) {
		t.Fatalf("broken-chain child still present after repair: %v", err)
	}
	if _, err := b.store.GetFileGCTaskByFileID(ctx, "f-race3"); err != nil {
		t.Fatalf("gc task for f-race3 after repair: %v", err)
	}
}

// insertBulkDirFixture bulk-inserts /z/ plus count empty subdirectories
// /z/dNNNN/ (all carrying inode_id so the racing create's EnsureParentDirs
// sees them as fully formed directories). Direct bulk SQL keeps the fixture
// fast enough to exceed the 1000-row sweep batch.
func insertBulkDirFixture(t *testing.T, b *Dat9Backend, count int) {
	t.Helper()
	now := time.Now().UTC()
	nodes := make([]*datastore.FileNode, 0, count+1)
	root := &datastore.FileNode{
		NodeID: "inode-z", Path: "/z/", ParentPath: "/", Name: "z",
		IsDirectory: true, InodeID: "inode-z", CreatedAt: now,
	}
	nodes = append(nodes, root)
	for i := 0; i < count; i++ {
		path := fmt.Sprintf("/z/d%04d/", i)
		nodes = append(nodes, &datastore.FileNode{
			NodeID: "n-" + path, Path: path, ParentPath: "/z/", Name: fmt.Sprintf("d%04d", i),
			IsDirectory: true, InodeID: "inode-" + path, CreatedAt: now,
		})
	}
	const cols = `node_id, path, path_hash, parent_path, parent_path_hash, name, is_directory, file_id, inode_id, created_at`
	for start := 0; start < len(nodes); start += 200 {
		end := min(start+200, len(nodes))
		var sb strings.Builder
		sb.WriteString(`INSERT INTO file_nodes (` + cols + `) VALUES `)
		var args []any
		for i, n := range nodes[start:end] {
			if i > 0 {
				sb.WriteString(", ")
			}
			sb.WriteString(`(?, ?, ?, ?, ?, ?, ?, NULL, ?, ?)`)
			args = append(args, n.NodeID, n.Path, datastore.StorageRefHash(n.Path), n.ParentPath,
				datastore.StorageRefHash(n.ParentPath), n.Name, n.IsDirectory, n.InodeID, n.CreatedAt)
		}
		if _, err := b.store.DB().Exec(sb.String(), args...); err != nil {
			t.Fatalf("bulk insert dir fixture: %v", err)
		}
	}
}