package backend

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"io"
	"strings"
	"testing"

	"github.com/c4pt0r/agfs/agfs-server/pkg/filesystem"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest/observer"

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

func newAppendLogTestBackend(t *testing.T) (*Dat9Backend, *fakeMetaQuotaStore) {
	t.Helper()
	b := newTestBackend(t)
	appendClient, ok := b.S3().(s3client.AppendLogClient)
	if !ok {
		t.Fatalf("local S3 client does not implement AppendLogClient: %T", b.S3())
	}
	metaStore := newFakeMetaQuotaStore()
	b.appendLogS3 = appendClient
	b.metaStore = metaStore
	b.tenantID = "append-log-test-tenant"
	b.storageNamespaceID = "append-log-test-namespace"
	return b, metaStore
}

func readAppendLogRange(t *testing.T, b *Dat9Backend, path string, start, end int64) string {
	t.Helper()
	plan, err := b.ReadPlanCtx(context.Background(), path)
	if err != nil {
		t.Fatalf("ReadPlanCtx: %v", err)
	}
	rc, _, err := b.OpenAppendLogRange(context.Background(), plan.StorageRef, plan.Size, start, end)
	if err != nil {
		t.Fatalf("OpenAppendLogRange(%d, %d): %v", start, end, err)
	}
	defer func() { _ = rc.Close() }()
	data, err := io.ReadAll(rc)
	if err != nil {
		t.Fatalf("read append-log range: %v", err)
	}
	return string(data)
}

type appendLogPutHookClient struct {
	s3client.AppendLogClient
	beforePut func()
}

func (c *appendLogPutHookClient) PutAppendLogObject(ctx context.Context, key string, body io.Reader, size int64) error {
	hook := c.beforePut
	c.beforePut = nil
	if hook != nil {
		hook()
	}
	return c.AppendLogClient.PutAppendLogObject(ctx, key, body, size)
}

type appendLogAppendHookClient struct {
	s3client.AppendLogClient
	afterAppend func()
}

func (c *appendLogAppendHookClient) AppendObject(ctx context.Context, key string, body io.Reader, size, offset int64) error {
	if err := c.AppendLogClient.AppendObject(ctx, key, body, size, offset); err != nil {
		return err
	}
	hook := c.afterAppend
	c.afterAppend = nil
	if hook != nil {
		hook()
	}
	return nil
}

type appendLogPutCountClient struct {
	s3client.AppendLogClient
	puts int
}

func (c *appendLogPutCountClient) PutAppendLogObject(ctx context.Context, key string, body io.Reader, size int64) error {
	c.puts++
	return c.AppendLogClient.PutAppendLogObject(ctx, key, body, size)
}

func rebindAppendLogPath(t *testing.T, b *Dat9Backend, ctx context.Context, path, movedPath string) {
	t.Helper()
	if err := b.RenameCtx(ctx, path, movedPath); err != nil {
		t.Fatalf("rename %s to %s: %v", path, movedPath, err)
	}
	if _, err := b.AppendLog(ctx, path, bytes.NewReader([]byte("new")), 3, 0, 0); err != nil {
		t.Fatalf("recreate append-log at %s: %v", path, err)
	}
}

func assertAppendLogMetadata(t *testing.T, b *Dat9Backend, ctx context.Context, path string, wantRevision, wantSize int64) {
	t.Helper()
	nf, err := b.Store().Stat(ctx, path)
	if err != nil {
		t.Fatalf("stat %s: %v", path, err)
	}
	if nf.File == nil || nf.File.Revision != wantRevision || nf.File.SizeBytes != wantSize {
		t.Fatalf("metadata for %s = %+v, want revision=%d size=%d", path, nf.File, wantRevision, wantSize)
	}
}

func TestAppendLogRejectsPathRebindBeforeCAS(t *testing.T) {
	ctx := context.Background()
	b, _ := newAppendLogTestBackend(t)
	created, err := b.AppendLog(ctx, "/append", bytes.NewReader([]byte("seed")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	b.appendLogS3 = &appendLogAppendHookClient{
		AppendLogClient: b.appendLogS3,
		afterAppend: func() {
			rebindAppendLogPath(t, b, ctx, "/append", "/append-moved")
		},
	}
	if _, err := b.AppendLog(ctx, "/append", bytes.NewReader([]byte("tail")), 4, created.Revision, created.Size); !errors.Is(err, datastore.ErrRevisionConflict) {
		t.Fatalf("append after path rebind error = %v, want revision conflict", err)
	}
	assertAppendLogMetadata(t, b, ctx, "/append-moved", created.Revision, created.Size)
	assertAppendLogMetadata(t, b, ctx, "/append", 1, 3)
}

func TestRewriteAppendLogRejectsPathRebindBeforeCAS(t *testing.T) {
	ctx := context.Background()
	b, _ := newAppendLogTestBackend(t)
	created, err := b.AppendLog(ctx, "/rewrite", bytes.NewReader([]byte("seed")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	b.appendLogS3 = &appendLogPutHookClient{
		AppendLogClient: b.appendLogS3,
		beforePut: func() {
			rebindAppendLogPath(t, b, ctx, "/rewrite", "/rewrite-moved")
		},
	}
	if _, err := b.RewriteAppendLog(ctx, "/rewrite", bytes.NewReader([]byte("reset")), 5, created.Revision); !errors.Is(err, datastore.ErrRevisionConflict) {
		t.Fatalf("rewrite after path rebind error = %v, want revision conflict", err)
	}
	assertAppendLogMetadata(t, b, ctx, "/rewrite-moved", created.Revision, created.Size)
	assertAppendLogMetadata(t, b, ctx, "/rewrite", 1, 3)
}

func TestRebaseAppendLogRejectsPathRebindBeforeCAS(t *testing.T) {
	ctx := context.Background()
	b, _ := newAppendLogTestBackend(t)
	created, err := b.AppendLog(ctx, "/rebase", bytes.NewReader([]byte("seed")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	local, ok := b.appendLogS3.(*s3client.LocalS3Client)
	if !ok {
		t.Fatalf("append-log client = %T, want local client", b.appendLogS3)
	}
	local.SetAppendLogFaults(s3client.AppendLogFaults{AppendErr: s3client.ErrAppendOffsetMismatch})
	b.appendLogS3 = &appendLogPutHookClient{
		AppendLogClient: local,
		beforePut: func() {
			rebindAppendLogPath(t, b, ctx, "/rebase", "/rebase-moved")
		},
	}
	if _, err := b.AppendLog(ctx, "/rebase", bytes.NewReader([]byte("tail")), 4, created.Revision, created.Size); !errors.Is(err, datastore.ErrRevisionConflict) {
		t.Fatalf("rebase after path rebind error = %v, want revision conflict", err)
	}
	assertAppendLogMetadata(t, b, ctx, "/rebase-moved", created.Revision, created.Size)
	assertAppendLogMetadata(t, b, ctx, "/rebase", 1, 3)
}

func TestAppendLogCreateAppendRebaseAndRewrite(t *testing.T) {
	ctx := context.Background()
	b, metaStore := newAppendLogTestBackend(t)

	created, err := b.AppendLog(ctx, "/wal", bytes.NewReader([]byte("head")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	if created.Revision != 1 || created.Size != 4 {
		t.Fatalf("create result = %+v, want revision=1 size=4", created)
	}
	appended, err := b.AppendLog(ctx, "/wal", bytes.NewReader([]byte("tail")), 4, 1, 4)
	if err != nil {
		t.Fatalf("append tail: %v", err)
	}
	if appended.Revision != 2 || appended.Size != 8 {
		t.Fatalf("append result = %+v, want revision=2 size=8", appended)
	}

	nf, err := b.Store().Stat(ctx, "/wal")
	if err != nil {
		t.Fatalf("stat append-log: %v", err)
	}
	if nf.File == nil || nf.File.ContentLayout != datastore.ContentLayoutAppendLog || nf.File.StorageType != datastore.StorageS3 {
		t.Fatalf("append-log storage = %+v, want S3 append_log", nf.File)
	}
	if wantPrefix := "append-log/" + b.storageNamespaceID + "/"; !strings.HasPrefix(nf.File.StorageRef, wantPrefix) {
		t.Fatalf("append-log ref = %q, want namespace-scoped prefix %q", nf.File.StorageRef, wantPrefix)
	}
	if nf.File.StorageEncryptionMode != datastore.StorageEncryptionNone || nf.File.StorageEncryptionKeyID != "" {
		t.Fatalf("append-log encryption = (%q, %q), want (none, empty)", nf.File.StorageEncryptionMode, nf.File.StorageEncryptionKeyID)
	}
	content, err := b.Store().GetContent(ctx, nf.File.FileID)
	if err != nil {
		t.Fatalf("get append-log content: %v", err)
	}
	if content.ContentBlob != nil {
		t.Fatalf("append-log content_blob = %q, want nil", content.ContentBlob)
	}

	if err := b.appendLogS3.AppendObject(ctx, nf.File.StorageRef, bytes.NewReader([]byte("hidden")), 6, 8); err != nil {
		t.Fatalf("create uncommitted physical suffix: %v", err)
	}
	if got := readAppendLogRange(t, b, "/wal", 0, 7); got != "headtail" {
		t.Fatalf("bounded append-log read = %q, want committed prefix", got)
	}
	plan, err := b.ReadPlanCtx(ctx, "/wal")
	if err != nil {
		t.Fatalf("read plan: %v", err)
	}
	if _, _, err := b.OpenAppendLogRange(ctx, plan.StorageRef, plan.Size, 8, 8); !errors.Is(err, io.EOF) {
		t.Fatalf("suffix range error = %v, want io.EOF", err)
	}

	if _, err := b.AppendLog(ctx, "/wal", bytes.NewReader([]byte("x")), 1, 2, 8); !errors.Is(err, ErrAppendLogRebased) {
		t.Fatalf("append after physical divergence = %v, want ErrAppendLogRebased", err)
	}
	nf, err = b.Store().Stat(ctx, "/wal")
	if err != nil {
		t.Fatalf("stat rebased append-log: %v", err)
	}
	if nf.File.Revision != 3 || nf.File.SizeBytes != 8 {
		t.Fatalf("rebased metadata = revision=%d size=%d, want revision=3 size=8", nf.File.Revision, nf.File.SizeBytes)
	}
	if got := readAppendLogRange(t, b, "/wal", 0, 7); got != "headtail" {
		t.Fatalf("rebased data = %q, want committed prefix", got)
	}
	if len(metaStore.objectGCCandidates) != 1 || metaStore.objectGCCandidates[0].ContentLayout != string(datastore.ContentLayoutAppendLog) {
		t.Fatalf("old-ref candidates = %+v, want one append_log candidate", metaStore.objectGCCandidates)
	}

	rewritten, err := b.RewriteAppendLog(ctx, "/wal", bytes.NewReader([]byte("reset")), 5, 3)
	if err != nil {
		t.Fatalf("rewrite append-log: %v", err)
	}
	if rewritten.Revision != 4 || rewritten.Size != 5 {
		t.Fatalf("rewrite result = %+v, want revision=4 size=5", rewritten)
	}
	nf, err = b.Store().Stat(ctx, "/wal")
	if err != nil {
		t.Fatalf("stat rewritten append-log: %v", err)
	}
	if nf.File.ContentLayout != datastore.ContentLayoutAppendLog || nf.File.StorageType != datastore.StorageS3 {
		t.Fatalf("rewrite changed append-log layout: %+v", nf.File)
	}
	if got := readAppendLogRange(t, b, "/wal", 0, 4); got != "reset" {
		t.Fatalf("rewritten data = %q, want reset", got)
	}
	if _, err := b.RewriteAppendLog(ctx, "/wal", bytes.NewReader([]byte("stale")), 5, 3); !errors.Is(err, datastore.ErrRevisionConflict) {
		t.Fatalf("stale rewrite error = %v, want revision conflict", err)
	}
}

func TestAppendLogRejectsOverLimitBeforeObjectWrite(t *testing.T) {
	b, _ := newAppendLogTestBackend(t)
	_, err := b.AppendLog(context.Background(), "/too-large", bytes.NewReader(nil), maxAppendLogObjectBytes+1, 0, 0)
	if !errors.Is(err, ErrAppendLogTooLarge) {
		t.Fatalf("AppendLog over limit error = %v, want ErrAppendLogTooLarge", err)
	}
	if _, err := b.Store().Stat(context.Background(), "/too-large"); !errors.Is(err, datastore.ErrNotFound) {
		t.Fatalf("over-limit path stat error = %v, want not found", err)
	}
}

func TestAppendLogFileSizeQuotaPrecedesObjectWrite(t *testing.T) {
	ctx := context.Background()
	b, metaStore := newAppendLogTestBackend(t)
	metaStore.mu.Lock()
	metaStore.config[b.tenantID] = &QuotaConfigView{
		TenantID: b.tenantID, MaxStorageBytes: 100, MaxFileSizeBytes: 3,
	}
	metaStore.mu.Unlock()
	client := &appendLogPutCountClient{AppendLogClient: b.appendLogS3}
	b.appendLogS3 = client
	if _, err := b.AppendLog(ctx, "/too-big", bytes.NewReader([]byte("four")), 4, 0, 0); !errors.Is(err, ErrFileSizeQuotaExceeded) {
		t.Fatalf("append-log file-size quota error = %v, want ErrFileSizeQuotaExceeded", err)
	}
	if client.puts != 0 {
		t.Fatalf("append-log PutAppendLogObject calls = %d, want 0", client.puts)
	}
	if _, err := b.Store().Stat(ctx, "/too-big"); !errors.Is(err, datastore.ErrNotFound) {
		t.Fatalf("quota-rejected append-log path stat = %v, want not found", err)
	}
}

func TestAppendLogTooManyPartsRebasesAndPermitsRetry(t *testing.T) {
	ctx := context.Background()
	b, _ := newAppendLogTestBackend(t)
	created, err := b.AppendLog(ctx, "/parts", bytes.NewReader([]byte("seed")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	local, ok := b.appendLogS3.(*s3client.LocalS3Client)
	if !ok {
		t.Fatalf("append-log client = %T, want local client", b.appendLogS3)
	}
	local.SetAppendLogFaults(s3client.AppendLogFaults{AppendErr: s3client.ErrAppendTooManyParts})
	if _, err := b.AppendLog(ctx, "/parts", bytes.NewReader([]byte("x")), 1, created.Revision, created.Size); !errors.Is(err, ErrAppendLogRebased) {
		t.Fatalf("TooManyParts append error = %v, want ErrAppendLogRebased", err)
	}
	local.SetAppendLogFaults(s3client.AppendLogFaults{})
	nf, err := b.Store().Stat(ctx, "/parts")
	if err != nil {
		t.Fatalf("stat rebased append-log: %v", err)
	}
	if nf.File.Revision != 2 || nf.File.SizeBytes != 4 {
		t.Fatalf("rebased metadata = revision=%d size=%d, want revision=2 size=4", nf.File.Revision, nf.File.SizeBytes)
	}
	result, err := b.AppendLog(ctx, "/parts", bytes.NewReader([]byte("x")), 1, nf.File.Revision, nf.File.SizeBytes)
	if err != nil {
		t.Fatalf("append after rebase: %v", err)
	}
	if result.Revision != 3 || result.Size != 5 {
		t.Fatalf("retry result = %+v, want revision=3 size=5", result)
	}
}

func TestAppendLogZeroByteDivergenceRebases(t *testing.T) {
	ctx := context.Background()
	b, _ := newAppendLogTestBackend(t)
	created, err := b.AppendLog(ctx, "/empty", bytes.NewReader(nil), 0, 0, 0)
	if err != nil {
		t.Fatalf("create zero-byte append-log: %v", err)
	}
	nf, err := b.Store().Stat(ctx, "/empty")
	if err != nil {
		t.Fatalf("stat zero-byte append-log: %v", err)
	}
	if err := b.appendLogS3.AppendObject(ctx, nf.File.StorageRef, bytes.NewReader([]byte("orphan")), 6, 0); err != nil {
		t.Fatalf("write uncommitted zero-byte suffix: %v", err)
	}
	if _, err := b.AppendLog(ctx, "/empty", bytes.NewReader([]byte("x")), 1, created.Revision, created.Size); !errors.Is(err, ErrAppendLogRebased) {
		t.Fatalf("zero-byte divergence error = %v, want ErrAppendLogRebased", err)
	}
	nf, err = b.Store().Stat(ctx, "/empty")
	if err != nil {
		t.Fatalf("stat rebased zero-byte append-log: %v", err)
	}
	if nf.File.Revision != 2 || nf.File.SizeBytes != 0 {
		t.Fatalf("rebased zero-byte metadata = revision=%d size=%d, want revision=2 size=0", nf.File.Revision, nf.File.SizeBytes)
	}
}

func TestAppendLogKnownCASConflictRetainsOldRefCandidate(t *testing.T) {
	ctx := context.Background()
	b, metaStore := newAppendLogTestBackend(t)
	created, err := b.AppendLog(ctx, "/cas", bytes.NewReader([]byte("seed")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	nf, err := b.Store().Stat(ctx, "/cas")
	if err != nil {
		t.Fatalf("stat append-log: %v", err)
	}
	local, ok := b.appendLogS3.(*s3client.LocalS3Client)
	if !ok {
		t.Fatalf("append-log client = %T, want local client", b.appendLogS3)
	}
	local.SetAppendLogFaults(s3client.AppendLogFaults{AppendErr: s3client.ErrAppendOffsetMismatch})
	var mutationErr error
	b.appendLogS3 = &appendLogPutHookClient{
		AppendLogClient: local,
		beforePut: func() {
			mutationErr = b.Store().InTx(ctx, func(tx *sql.Tx) error {
				live, err := b.Store().StatForUpdateTx(ctx, tx, "/cas")
				if err != nil {
					return err
				}
				_, err = b.Store().AppendLogContentIfMatchTx(tx, live, created.Revision, created.Size, nf.File.StorageRef, 0)
				return err
			})
		},
	}
	if _, err := b.AppendLog(ctx, "/cas", bytes.NewReader([]byte("x")), 1, created.Revision, created.Size); !errors.Is(err, datastore.ErrRevisionConflict) {
		t.Fatalf("append after CAS mutation error = %v, want revision conflict", err)
	}
	if mutationErr != nil {
		t.Fatalf("mutate metadata before replacement CAS: %v", mutationErr)
	}
	if len(metaStore.objectGCCandidates) != 1 || metaStore.objectGCCandidates[0].StorageRef != nf.File.StorageRef || metaStore.objectGCCandidates[0].Reason != meta.ObjectGCReasonOverwrite {
		t.Fatalf("known CAS conflict candidates = %+v, want retained old-ref overwrite candidate", metaStore.objectGCCandidates)
	}
	nf, err = b.Store().Stat(ctx, "/cas")
	if err != nil {
		t.Fatalf("stat after CAS conflict: %v", err)
	}
	if nf.File.Revision != 2 || nf.File.StorageRef == "" {
		t.Fatalf("metadata after CAS conflict = %+v, want original ref at revision 2", nf.File)
	}
}

func TestAppendLogRejectsGenericOverwriteAndExistingZeroAppend(t *testing.T) {
	ctx := context.Background()
	b, _ := newAppendLogTestBackend(t)
	created, err := b.AppendLog(ctx, "/wal", bytes.NewReader([]byte("seed")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	if _, err := b.WriteCtx(ctx, "/wal", []byte("rewrite"), 0, filesystem.WriteFlagTruncate); !errors.Is(err, ErrAppendLogUnsupported) {
		t.Fatalf("generic overwrite error = %v, want ErrAppendLogUnsupported", err)
	}
	if _, err := b.InitiateUpload(ctx, "/wal", s3client.PartSize); !errors.Is(err, ErrAppendLogUnsupported) {
		t.Fatalf("V1 upload error = %v, want ErrAppendLogUnsupported", err)
	}
	if _, err := b.InitiateUploadV2(ctx, "/wal", s3client.PartSize); !errors.Is(err, ErrAppendLogUnsupported) {
		t.Fatalf("V2 upload error = %v, want ErrAppendLogUnsupported", err)
	}
	nf, err := b.Store().Stat(ctx, "/wal")
	if err != nil {
		t.Fatalf("stat append-log: %v", err)
	}
	if _, _, err := b.WriteStoredObjectCreatedCtxIfRevision(ctx, "/wal", &datastore.FSLayerEntry{
		Path:         "/wal",
		StorageType:  string(datastore.StorageS3),
		StorageRef:   "layers/replacement",
		SizeBytes:    1,
		BaseRevision: created.Revision,
		BaseInodeID:  nf.File.FileID,
	}); !errors.Is(err, ErrAppendLogUnsupported) {
		t.Fatalf("LayerFS overwrite error = %v, want ErrAppendLogUnsupported", err)
	}
	if _, err := b.AppendLog(ctx, "/wal", bytes.NewReader(nil), 0, created.Revision, created.Size); !errors.Is(err, ErrAppendLogInvalidRequest) {
		t.Fatalf("existing zero append error = %v, want ErrAppendLogInvalidRequest", err)
	}
	nf, err = b.Store().Stat(ctx, "/wal")
	if err != nil {
		t.Fatalf("stat append-log: %v", err)
	}
	if nf.File.Revision != created.Revision || nf.File.SizeBytes != created.Size {
		t.Fatalf("metadata after rejected writes = revision=%d size=%d, want revision=%d size=%d", nf.File.Revision, nf.File.SizeBytes, created.Revision, created.Size)
	}
	results, err := b.BatchWriteCtx(ctx, []BatchWriteItem{{Path: "/wal", Data: []byte("inline"), ExpectedRevision: created.Revision}})
	if err != nil {
		t.Fatalf("batch write returned transaction error: %v", err)
	}
	if len(results) != 1 || !errors.Is(results[0].Err, ErrAppendLogUnsupported) {
		t.Fatalf("batch write result = %+v, want append-log unsupported", results)
	}
}

func TestAppendLogRangeUsesPlannedGeneration(t *testing.T) {
	ctx := context.Background()
	b, _ := newAppendLogTestBackend(t)
	created, err := b.AppendLog(ctx, "/wal", bytes.NewReader([]byte("old")), 3, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	plan, err := b.ReadPlanCtx(ctx, "/wal")
	if err != nil {
		t.Fatalf("read plan: %v", err)
	}
	if _, err := b.RewriteAppendLog(ctx, "/wal", bytes.NewReader([]byte("new")), 3, created.Revision); err != nil {
		t.Fatalf("rewrite append-log: %v", err)
	}
	rc, _, err := b.OpenAppendLogRange(ctx, plan.StorageRef, plan.Size, 0, plan.Size-1)
	if err != nil {
		t.Fatalf("open planned generation: %v", err)
	}
	defer func() { _ = rc.Close() }()
	data, err := io.ReadAll(rc)
	if err != nil {
		t.Fatalf("read planned generation: %v", err)
	}
	if string(data) != "old" {
		t.Fatalf("planned generation body = %q, want old", data)
	}
}

func TestAppendLogEnabledRequiresClientAndDurableGCRoute(t *testing.T) {
	withoutAppendLog := newTestBackend(t)
	if withoutAppendLog.AppendLogEnabled() {
		t.Fatal("append-log reported enabled without an append-log client")
	}
	withAppendLog, _ := newAppendLogTestBackend(t)
	if !withAppendLog.AppendLogEnabled() {
		t.Fatal("append-log reported disabled with a client, MetaStore, and namespace")
	}
	withAppendLog.storageNamespaceID = ""
	if withAppendLog.AppendLogEnabled() {
		t.Fatal("append-log reported enabled without a durable GC namespace")
	}
}

func TestAppendLogForkWriteDisableKeepsCommittedReads(t *testing.T) {
	ctx := context.Background()
	b, _ := newAppendLogTestBackend(t)
	created, err := b.AppendLog(ctx, "/wal", bytes.NewReader([]byte("seed")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	b.appendLogWriteDisabled = true
	if b.AppendLogEnabled() {
		t.Fatal("fork backend advertised append-log writes")
	}
	if _, err := b.AppendLog(ctx, "/wal", bytes.NewReader([]byte("tail")), 4, created.Revision, created.Size); !errors.Is(err, ErrAppendLogUnsupported) {
		t.Fatalf("fork append error = %v, want ErrAppendLogUnsupported", err)
	}
	if _, err := b.RewriteAppendLog(ctx, "/wal", bytes.NewReader([]byte("reset")), 5, created.Revision); !errors.Is(err, ErrAppendLogUnsupported) {
		t.Fatalf("fork rewrite error = %v, want ErrAppendLogUnsupported", err)
	}
	plan, err := b.ReadPlanCtx(ctx, "/wal")
	if err != nil {
		t.Fatalf("fork read plan: %v", err)
	}
	if plan.ContentLayout != datastore.ContentLayoutAppendLog || plan.StorageRef == "" {
		t.Fatalf("fork read plan = %+v, want readable append-log generation", plan)
	}
}

func TestAppendLogWriteRequiresDurableGCRoute(t *testing.T) {
	ctx := context.Background()
	b := newTestBackend(t)
	appendClient, ok := b.S3().(s3client.AppendLogClient)
	if !ok {
		t.Fatalf("local S3 client does not implement AppendLogClient: %T", b.S3())
	}
	b.appendLogS3 = appendClient
	if _, err := b.AppendLog(ctx, "/no-gc", bytes.NewReader([]byte("x")), 1, 0, 0); !errors.Is(err, ErrAppendLogUnsupported) {
		t.Fatalf("append without durable GC route error = %v, want ErrAppendLogUnsupported", err)
	}
	if _, err := b.Store().Stat(ctx, "/no-gc"); !errors.Is(err, datastore.ErrNotFound) {
		t.Fatalf("append without durable GC route created metadata: %v", err)
	}

	ready, _ := newAppendLogTestBackend(t)
	if _, err := ready.AppendLog(ctx, "/rewrite-no-gc", bytes.NewReader([]byte("seed")), 4, 0, 0); err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	ready.metaStore = nil
	if _, err := ready.RewriteAppendLog(ctx, "/rewrite-no-gc", bytes.NewReader([]byte("reset")), 5, 1); !errors.Is(err, ErrAppendLogUnsupported) {
		t.Fatalf("rewrite without durable GC route error = %v, want ErrAppendLogUnsupported", err)
	}
}

// TestAppendLogUsesSingleInTxAndNoStorageMetaReread is the contract test for
// the collapsed append path: one transaction for the existing-object tail
// append and no redundant GetFileStorageMetaForUpdateTx re-read inside the
// CAS. It observes the instrumented DB operation log instead of injecting
// call-counting failpoints.
func TestAppendLogUsesSingleInTxAndNoStorageMetaReread(t *testing.T) {
	t.Setenv("DRIVE9_DB_TRACE_LOG_ENABLED", "true")
	logger.ResetDBTraceLogEnabledForTest()
	t.Cleanup(logger.ResetDBTraceLogEnabledForTest)
	t.Setenv("DRIVE9_DB_SLOW_TRACE_MS", "0")
	logger.ResetDBSlowTraceThresholdForTest()
	t.Cleanup(logger.ResetDBSlowTraceThresholdForTest)

	core, recorded := observer.New(zap.InfoLevel)
	prevLogger := logger.L()
	logger.Set(zap.New(core))
	t.Cleanup(func() { logger.Set(prevLogger) })

	ctx := context.Background()
	b, _ := newAppendLogTestBackend(t)
	created, err := b.AppendLog(ctx, "/contract", bytes.NewReader([]byte("seed")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	recorded.TakeAll() // discard create-path noise

	result, err := b.AppendLog(ctx, "/contract", bytes.NewReader([]byte("x")), 1, created.Revision, created.Size)
	if err != nil {
		t.Fatalf("append: %v", err)
	}
	if result.Revision != created.Revision+1 || result.Size != created.Size+1 {
		t.Fatalf("append result = %+v, want revision=%d size=%d", result, created.Revision+1, created.Size+1)
	}

	begins := 0
	rereads := 0
	for _, entry := range recorded.TakeAll() {
		if entry.Message != "db_operation_timing" {
			continue
		}
		fields := entry.ContextMap()
		if fields["operation"] == "begin" {
			begins++
		}
		if sql, ok := fields["sql"].(string); ok &&
			strings.Contains(sql, "c.storage_ref, i.revision, i.size_bytes") {
			rereads++
		}
	}
	if begins != 1 {
		t.Fatalf("append InTx begins = %d, want 1", begins)
	}
	if rereads != 0 {
		t.Fatalf("GetFileStorageMetaForUpdateTx queries = %d, want 0", rereads)
	}
}

// TestAppendLogStorageQuotaPrecheckPrecedesObjectWrite guards the read-only
// quota precheck ordering: an over-limit storage projection must reject the
// append before any S3 object write.
func TestAppendLogStorageQuotaPrecheckPrecedesObjectWrite(t *testing.T) {
	ctx := context.Background()
	b, metaStore := newAppendLogTestBackend(t)
	metaStore.mu.Lock()
	metaStore.config[b.tenantID] = &QuotaConfigView{
		TenantID: b.tenantID, MaxStorageBytes: 100,
	}
	metaStore.mu.Unlock()
	appends := 0
	b.appendLogS3 = &appendLogAppendHookClient{
		AppendLogClient: b.appendLogS3,
		afterAppend: func() {
			appends++
		},
	}

	created, err := b.AppendLog(ctx, "/quota", bytes.NewReader([]byte("ab")), 2, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	appends = 0
	metaStore.mu.Lock()
	metaStore.usage[b.tenantID] = &QuotaUsageView{TenantID: b.tenantID, StorageBytes: 99}
	metaStore.mu.Unlock()
	if _, err := b.AppendLog(ctx, "/quota", bytes.NewReader([]byte("xx")), 2, created.Revision, created.Size); !errors.Is(err, ErrStorageQuotaExceeded) {
		t.Fatalf("over-limit append error = %v, want ErrStorageQuotaExceeded", err)
	}
	if appends != 0 {
		t.Fatalf("append-log AppendObject calls = %d, want 0", appends)
	}
}

func TestAppendLogMissingContentsFailsClosedBeforeS3(t *testing.T) {
	ctx := context.Background()
	b, _ := newAppendLogTestBackend(t)
	appends := 0
	b.appendLogS3 = &appendLogAppendHookClient{
		AppendLogClient: b.appendLogS3,
		afterAppend: func() {
			appends++
		},
	}
	created, err := b.AppendLog(ctx, "/wal", bytes.NewReader([]byte("head")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	nf, err := b.Store().Stat(ctx, "/wal")
	if err != nil {
		t.Fatalf("stat append-log: %v", err)
	}
	if _, err := b.Store().DB().ExecContext(ctx, `DELETE FROM contents WHERE inode_id = ?`, nf.File.FileID); err != nil {
		t.Fatalf("delete contents row: %v", err)
	}
	if _, err := b.AppendLog(ctx, "/wal", bytes.NewReader([]byte("tail")), 4, created.Revision, created.Size); !errors.Is(err, datastore.ErrRevisionConflict) {
		t.Fatalf("append with missing contents error = %v, want ErrRevisionConflict", err)
	}
	if appends != 0 {
		t.Fatalf("AppendObject calls = %d, want 0", appends)
	}
}

func TestAppendLogMissingSemanticFailsClosedBeforeS3(t *testing.T) {
	ctx := context.Background()
	b, _ := newAppendLogTestBackend(t)
	appends := 0
	b.appendLogS3 = &appendLogAppendHookClient{
		AppendLogClient: b.appendLogS3,
		afterAppend: func() {
			appends++
		},
	}
	created, err := b.AppendLog(ctx, "/wal", bytes.NewReader([]byte("head")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	nf, err := b.Store().Stat(ctx, "/wal")
	if err != nil {
		t.Fatalf("stat append-log: %v", err)
	}
	if _, err := b.Store().DB().ExecContext(ctx, `DELETE FROM semantic WHERE inode_id = ?`, nf.File.FileID); err != nil {
		t.Fatalf("delete semantic row: %v", err)
	}
	if _, err := b.AppendLog(ctx, "/wal", bytes.NewReader([]byte("tail")), 4, created.Revision, created.Size); !errors.Is(err, datastore.ErrRevisionConflict) {
		t.Fatalf("append with missing semantic error = %v, want ErrRevisionConflict", err)
	}
	if appends != 0 {
		t.Fatalf("AppendObject calls = %d, want 0", appends)
	}
}

func TestAppendLogNonConfirmedInodeFailsClosedBeforeS3(t *testing.T) {
	for _, status := range []string{"PENDING", "DELETED"} {
		t.Run(status, func(t *testing.T) {
			ctx := context.Background()
			b, _ := newAppendLogTestBackend(t)
			appends := 0
			b.appendLogS3 = &appendLogAppendHookClient{
				AppendLogClient: b.appendLogS3,
				afterAppend: func() {
					appends++
				},
			}
			created, err := b.AppendLog(ctx, "/wal", bytes.NewReader([]byte("head")), 4, 0, 0)
			if err != nil {
				t.Fatalf("create append-log: %v", err)
			}
			nf, err := b.Store().Stat(ctx, "/wal")
			if err != nil {
				t.Fatalf("stat append-log: %v", err)
			}
			if _, err := b.Store().DB().ExecContext(ctx, `UPDATE inodes SET status = ? WHERE inode_id = ?`, status, nf.File.FileID); err != nil {
				t.Fatalf("set inode status: %v", err)
			}
			if _, err := b.AppendLog(ctx, "/wal", bytes.NewReader([]byte("tail")), 4, created.Revision, created.Size); !errors.Is(err, datastore.ErrRevisionConflict) {
				t.Fatalf("append with %s inode error = %v, want ErrRevisionConflict", status, err)
			}
			if appends != 0 {
				t.Fatalf("AppendObject calls = %d, want 0", appends)
			}
		})
	}
}