//go:build failpoint

package datastore

import (
	"context"
	"errors"
	"runtime"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/pingcap/failpoint"
)

const fsLayerDeleteAfterPinsCheckFailpoint = "github.com/mem9-ai/drive9/pkg/datastore/fsLayerDeleteAfterPinsCheck"

func TestDeleteFSLayerSerializesConcurrentForkAfterPinsCheck(t *testing.T) {
	s := newTestStore(t)
	ctx := context.Background()
	if err := s.CreateFSLayer(ctx, &FSLayer{LayerID: "p-del-race", BaseRootPath: "/repo"}); err != nil {
		t.Fatalf("create: %v", err)
	}

	var (
		mu      sync.Mutex
		forkErr error
	)
	forkDone := make(chan struct{})
	if err := failpoint.EnableCall(fsLayerDeleteAfterPinsCheckFailpoint, func(layerID string) {
		if layerID != "p-del-race" {
			return
		}
		go func() {
			defer close(forkDone)
			_, err := s.ForkFSLayer(ctx, FSLayerForkOptions{
				ChildLayerID:  "c-del-race",
				ParentLayerID: "p-del-race",
			})
			mu.Lock()
			forkErr = err
			mu.Unlock()
		}()
	}); err != nil {
		t.Fatalf("enable failpoint: %v", err)
	}
	t.Cleanup(func() { _ = failpoint.Disable(fsLayerDeleteAfterPinsCheckFailpoint) })

	if err := s.DeleteFSLayer(ctx, "p-del-race", DeleteFSLayerOptions{}); err != nil {
		t.Fatalf("DeleteFSLayer: %v", err)
	}
	<-forkDone

	mu.Lock()
	gotForkErr := forkErr
	mu.Unlock()
	if !errors.Is(gotForkErr, ErrFSLayerStateConflict) {
		t.Fatalf("concurrent fork err=%v, want ErrFSLayerStateConflict", gotForkErr)
	}
	parent, err := s.GetFSLayer(ctx, "p-del-race")
	if err != nil {
		t.Fatalf("get parent: %v", err)
	}
	if parent.State != FSLayerStateAbandoned {
		t.Fatalf("parent state=%s, want abandoned", parent.State)
	}
	if _, err := s.GetFSLayer(ctx, "c-del-race"); !errors.Is(err, ErrNotFound) {
		t.Fatalf("child get err=%v, want ErrNotFound (fork must not pin an abandoned parent)", err)
	}
}

const (
	fsLayerUpsertBeforeCommitFailpoint = "github.com/mem9-ai/drive9/pkg/datastore/fsLayerUpsertBeforeCommit"
	fsLayerForkAfterBeginFailpoint     = "github.com/mem9-ai/drive9/pkg/datastore/fsLayerForkAfterBegin"
)

func TestForkPinsParentWriteThatCommitsDuringLockWait(t *testing.T) {
	s := newTestStore(t)
	ctx := context.Background()
	if err := s.CreateFSLayer(ctx, &FSLayer{LayerID: "p-fork-pin", BaseRootPath: "/repo"}); err != nil {
		t.Fatalf("create: %v", err)
	}

	var (
		mu      sync.Mutex
		forkErr error
		child   *FSLayer
	)
	upsertHolds := make(chan struct{})
	forkBegun := make(chan struct{})
	forkDone := make(chan struct{})

	if err := failpoint.EnableCall(fsLayerUpsertBeforeCommitFailpoint, func(layerID string) {
		if layerID != "p-fork-pin" {
			return
		}
		close(upsertHolds)
		go func() {
			defer close(forkDone)
			got, err := s.ForkFSLayer(ctx, FSLayerForkOptions{
				ChildLayerID:  "c-fork-pin",
				ParentLayerID: "p-fork-pin",
			})
			mu.Lock()
			child, forkErr = got, err
			mu.Unlock()
		}()
		<-forkBegun
		// AfterBegin returns and the next statement is parent FOR UPDATE.
		// Do not sleep: wait until that query is actually blocked.
		waitForFSLayerParentLockWait(t, 2*time.Second)
	}); err != nil {
		t.Fatalf("enable upsert failpoint: %v", err)
	}
	t.Cleanup(func() { _ = failpoint.Disable(fsLayerUpsertBeforeCommitFailpoint) })

	if err := failpoint.EnableCall(fsLayerForkAfterBeginFailpoint, func(parentID string) {
		if parentID != "p-fork-pin" {
			return
		}
		<-upsertHolds
		close(forkBegun)
	}); err != nil {
		t.Fatalf("enable fork failpoint: %v", err)
	}
	t.Cleanup(func() { _ = failpoint.Disable(fsLayerForkAfterBeginFailpoint) })

	if err := s.UpsertFSLayerEntry(ctx, &FSLayerEntry{
		LayerID: "p-fork-pin", Path: "/repo/during.txt",
		Op: FSLayerEntryOpUpsert, Kind: FSLayerEntryKindFile, ContentBlob: []byte("pin"),
	}); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	<-forkDone

	mu.Lock()
	gotChild, gotErr := child, forkErr
	mu.Unlock()
	if gotErr != nil {
		t.Fatalf("fork: %v", gotErr)
	}
	if gotChild == nil || gotChild.OriginSeq != 1 {
		t.Fatalf("origin_seq=%v, want 1 (parent write committed while fork waited)", gotChild)
	}
	_, hit, err := s.ResolveFSLayerPath(ctx, "c-fork-pin", "/repo/during.txt")
	if err != nil {
		t.Fatalf("resolve: %v", err)
	}
	if !hit {
		t.Fatal("child must see the parent write that committed during fork lock wait")
	}
}

// waitForFSLayerParentLockWait fails unless ForkFSLayer is blocked in the
// parent `SELECT … FOR UPDATE` (QueryRowContext on the network). That is
// the lock-wait the snapshot-MAX bug depends on; a timeout must not
// silently pass.
//
// Do not probe processlist / data_lock_waits: the testcontainers `dat9`
// user cannot see PFS lock tables (CI stayed empty for the full 2s while
// the FOR UPDATE itself logged duration_ms≈2000), and processlist INFO
// still shows the writer's already-finished parent FOR UPDATE.
func waitForFSLayerParentLockWait(t *testing.T, timeout time.Duration) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	buf := make([]byte, 1<<20)
	var last string
	for time.Now().Before(deadline) {
		n := runtime.Stack(buf, true)
		last = string(buf[:n])
		if forkBlockedOnParentLock(last) {
			return
		}
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatalf("fork never blocked on parent FOR UPDATE while the writer held the lock\n%s", last)
}

func forkBlockedOnParentLock(stacks string) bool {
	for _, g := range strings.Split(stacks, "\n\n") {
		if !strings.Contains(g, "(*Store).ForkFSLayer") {
			continue
		}
		// After AfterBegin the next statement is parent FOR UPDATE. The
		// fork goroutine sits in a socket read until the writer commits.
		if strings.Contains(g, "[IO wait]") ||
			strings.Contains(g, "[syscall]") ||
			strings.Contains(g, "internal/poll.runtime_pollWait") ||
			strings.Contains(g, "net.(*netFD).Read") {
			return true
		}
	}
	return false
}

func TestForkBlockedOnParentLockStackShape(t *testing.T) {
	blocked := `goroutine 42 [IO wait]:
internal/poll.runtime_pollWait(0x1, 0x72)
	runtime/netpoll.go:1
net.(*netFD).Read(0x1, {0x2, 0x3, 0x4})
	net/fd_posix.go:1
database/sql.(*Tx).QueryRowContext(0x1, 0x2, {0x3, 0x4}, {0x5, 0x6, 0x7})
	database/sql/sql.go:1
github.com/mem9-ai/drive9/pkg/datastore.(*Store).ForkFSLayer(0x1, 0x2, {0x3})
	pkg/datastore/fs_layer_chain.go:71 +0x1
`
	if !forkBlockedOnParentLock(blocked) {
		t.Fatal("expected IO-wait ForkFSLayer stack to count as blocked")
	}
	if forkBlockedOnParentLock(`goroutine 1 [running]:
github.com/mem9-ai/drive9/pkg/datastore.(*Store).ForkFSLayer(...)
	pkg/datastore/fs_layer_chain.go:70`) {
		t.Fatal("running AfterBegin must not count as the parent lock wait")
	}
}