//go:build failpoint

package backend

import (
	"bytes"
	"context"
	"errors"
	"testing"

	"github.com/pingcap/failpoint"
)

const (
	appendLogFinalTxFailFailpoint = "github.com/mem9-ai/drive9/pkg/backend/appendLogFinalTxFail"
)

// TestAppendLogFinalTxFailureSelfHeals covers the new failure semantics: the
// S3 append succeeds but the final transaction fails, leaving orphan tail
// bytes beyond the metadata size. The next append at the same offset hits
// InvalidWriteOffset, the rebase path rebuilds the ref from the metadata-size
// prefix, and a retry with the fresh revision succeeds.
func TestAppendLogFinalTxFailureSelfHeals(t *testing.T) {
	ctx := context.Background()
	b, _ := newAppendLogTestBackend(t)
	created, err := b.AppendLog(ctx, "/heal", bytes.NewReader([]byte("seed")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}

	injected := 0
	if err := failpoint.EnableCall(appendLogFinalTxFailFailpoint, func(path string, txErr *error) {
		if path != "/heal" {
			return
		}
		injected++
		if injected == 1 {
			*txErr = ErrStorageQuotaExceeded
		}
	}); err != nil {
		t.Fatalf("enable final tx fail failpoint: %v", err)
	}
	t.Cleanup(func() { _ = failpoint.Disable(appendLogFinalTxFailFailpoint) })

	if _, err := b.AppendLog(ctx, "/heal", bytes.NewReader([]byte("x")), 1, created.Revision, created.Size); !errors.Is(err, ErrStorageQuotaExceeded) {
		t.Fatalf("append error = %v, want ErrStorageQuotaExceeded (507)", err)
	}

	// The metadata size is unchanged, so the read path still truncates the
	// orphan tail byte.
	if got := readAppendLogRange(t, b, "/heal", 0, created.Size-1); got != "seed" {
		t.Fatalf("read after failed append = %q, want %q", got, "seed")
	}

	// The stale-offset append must rebase and return ErrAppendLogRebased.
	if _, err := b.AppendLog(ctx, "/heal", bytes.NewReader([]byte("x")), 1, created.Revision, created.Size); !errors.Is(err, ErrAppendLogRebased) {
		t.Fatalf("stale-offset append error = %v, want ErrAppendLogRebased", err)
	}

	nf, err := b.Store().Stat(ctx, "/heal")
	if err != nil {
		t.Fatalf("stat rebased append-log: %v", err)
	}
	if nf.File == nil {
		t.Fatal("rebased append-log has no file metadata")
	}
	if got := readAppendLogRange(t, b, "/heal", 0, nf.File.SizeBytes-1); got != "seed" {
		t.Fatalf("read after rebase = %q, want %q", got, "seed")
	}

	result, err := b.AppendLog(ctx, "/heal", bytes.NewReader([]byte("x")), 1, nf.File.Revision, nf.File.SizeBytes)
	if err != nil {
		t.Fatalf("append after rebase: %v", err)
	}
	if result.Revision != nf.File.Revision+1 || result.Size != nf.File.SizeBytes+1 {
		t.Fatalf("retry result = %+v, want revision=%d size=%d", result, nf.File.Revision+1, nf.File.SizeBytes+1)
	}
	if got := readAppendLogRange(t, b, "/heal", 0, result.Size-1); got != "seedx" {
		t.Fatalf("read after retry = %q, want %q", got, "seedx")
	}
}