package server

import (
	"bytes"
	"context"
	"crypto/rand"
	"database/sql"
	"errors"
	"fmt"
	"io"
	"strings"
	"testing"
	"time"

	"github.com/c4pt0r/agfs/agfs-server/pkg/filesystem"
	"github.com/go-sql-driver/mysql"
	backendpkg "github.com/mem9-ai/drive9/pkg/backend"
	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/encrypt"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/s3client"
	"github.com/mem9-ai/drive9/pkg/tenant"
)

type objectGCTestRuntime struct {
	meta   *meta.Store
	pool   *tenant.Pool
	dbHost string
	dbPort int
	dbUser string
	dbPass string
	dbName string
}

func newObjectGCTestRuntime(t *testing.T) *objectGCTestRuntime {
	t.Helper()
	if testDSN == "" {
		t.Skip("no test database available")
	}
	initServerTenantSchema(t, testDSN)
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	cleanupForkTestTables(t, metaStore)

	parsed, err := mysql.ParseDSN(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	host, port := "127.0.0.1", 3306
	if parsed.Addr != "" {
		h, p, _ := strings.Cut(parsed.Addr, ":")
		if h != "" {
			host = h
		}
		if p != "" {
			_, _ = fmt.Sscanf(p, "%d", &port)
		}
	}
	master := make([]byte, 32)
	if _, err := rand.Read(master); err != nil {
		t.Fatal(err)
	}
	enc, err := encrypt.NewLocalAESEncryptor(master)
	if err != nil {
		t.Fatal(err)
	}
	pool := tenant.NewPool(tenant.PoolConfig{S3Dir: mustTempDir(t), PublicURL: "http://localhost", SkipTiDBSchemaCheck: true}, enc)
	pool.SetMetaStore(metaStore)
	t.Cleanup(pool.Close)
	return &objectGCTestRuntime{meta: metaStore, pool: pool, dbHost: host, dbPort: port, dbUser: parsed.User, dbPass: parsed.Passwd, dbName: parsed.DBName}
}

func (rt *objectGCTestRuntime) insertTenant(t *testing.T, id string, status meta.TenantStatus, kind meta.TenantKind, namespaceID string) {
	t.Helper()
	ctx := context.Background()
	passCipher, err := rt.pool.Encrypt(ctx, []byte(rt.dbPass))
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC()
	if err := rt.meta.InsertTenant(ctx, &meta.Tenant{
		ID:                 id,
		Status:             status,
		Kind:               kind,
		ParentTenantID:     "",
		StorageNamespaceID: namespaceID,
		DBHost:             rt.dbHost,
		DBPort:             rt.dbPort,
		DBUser:             rt.dbUser,
		DBPasswordCipher:   passCipher,
		DBName:             rt.dbName,
		DBTLS:              false,
		Provider:           tenant.ProviderTiDBCloudNative,
		ClusterID:          "cluster-a",
		BranchID:           "branch-a",
		SchemaVersion:      1,
		CreatedAt:          now,
		UpdatedAt:          now,
	}); err != nil {
		t.Fatal(err)
	}
}

func (rt *objectGCTestRuntime) upsertNamespace(t *testing.T, id, ownerID string) {
	t.Helper()
	if err := rt.meta.UpsertStorageNamespace(context.Background(), &meta.StorageNamespace{
		ID:            id,
		OwnerTenantID: ownerID,
		Backend:       "local",
		Prefix:        ownerID + "/",
		State:         meta.StorageNamespaceActive,
	}); err != nil {
		t.Fatal(err)
	}
}

func (rt *objectGCTestRuntime) enqueueCandidate(t *testing.T, namespaceID, storageRef string) meta.ObjectGCCandidate {
	t.Helper()
	ctx := context.Background()
	if err := rt.meta.EnqueueObjectGCCandidate(ctx, &meta.ObjectGCCandidateInput{
		NamespaceID:    namespaceID,
		StorageRef:     storageRef,
		StorageRefHash: datastore.StorageRefHash(storageRef),
		Reason:         meta.ObjectGCReasonOverwrite,
		NotBefore:      time.Now().UTC().Add(-time.Minute),
	}); err != nil {
		t.Fatal(err)
	}
	candidates, err := rt.meta.ListDueObjectGCCandidates(ctx, time.Now().UTC(), 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(candidates) != 1 {
		t.Fatalf("candidate count = %d, want 1: %+v", len(candidates), candidates)
	}
	return candidates[0]
}

func objectGCCandidateRow(t *testing.T, db *sql.DB, namespaceID, storageRef string) (meta.ObjectGCCandidateState, string) {
	t.Helper()
	var state meta.ObjectGCCandidateState
	var lastError sql.NullString
	if err := db.QueryRow(`SELECT state, last_error FROM object_gc_candidates WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?`,
		namespaceID, datastore.StorageRefHash(storageRef), storageRef).Scan(&state, &lastError); err != nil {
		t.Fatal(err)
	}
	return state, lastError.String
}

func TestObjectGCSkipsSharedBlobWhenForkExists(t *testing.T) {
	rt := newObjectGCTestRuntime(t)
	rt.upsertNamespace(t, "ns-a", "owner-a")
	rt.insertTenant(t, "fork-a", meta.TenantProvisioning, meta.TenantKindFork, "ns-a")
	candidate := rt.enqueueCandidate(t, "ns-a", "blobs/shared")

	worker := &objectGCWorker{meta: rt.meta, pool: rt.pool}
	if err := worker.processCandidate(context.Background(), candidate); err != nil {
		t.Fatal(err)
	}

	state, lastError := objectGCCandidateRow(t, rt.meta.DB(), "ns-a", "blobs/shared")
	if state != meta.ObjectGCCandidatePending || lastError != "namespace has active fork" {
		t.Fatalf("candidate state=%s last_error=%q", state, lastError)
	}
}

func TestObjectGCSkipsSharedLayerWhenForkExists(t *testing.T) {
	rt := newObjectGCTestRuntime(t)
	rt.upsertNamespace(t, "ns-layer-fork", "owner-layer-fork")
	rt.insertTenant(t, "fork-layer-fork", meta.TenantProvisioning, meta.TenantKindFork, "ns-layer-fork")
	candidate := rt.enqueueCandidate(t, "ns-layer-fork", "layers/layer-fork/object-1")

	worker := &objectGCWorker{meta: rt.meta, pool: rt.pool}
	if err := worker.processCandidate(context.Background(), candidate); err != nil {
		t.Fatal(err)
	}
	state, lastError := objectGCCandidateRow(t, rt.meta.DB(), "ns-layer-fork", candidate.StorageRef)
	if state != meta.ObjectGCCandidatePending || lastError != "namespace has active fork" {
		t.Fatalf("candidate state=%s last_error=%q", state, lastError)
	}
}

func TestObjectGCProcessOnceHonorsConfiguredBatchSize(t *testing.T) {
	rt := newObjectGCTestRuntime(t)
	rt.upsertNamespace(t, "ns-batch", "owner-batch")
	rt.insertTenant(t, "owner-batch", meta.TenantProvisioning, meta.TenantKindLive, "ns-batch")
	for _, storageRef := range []string{"blobs/batch-1", "blobs/batch-2", "blobs/batch-3"} {
		if err := rt.meta.EnqueueObjectGCCandidate(context.Background(), &meta.ObjectGCCandidateInput{
			NamespaceID:    "ns-batch",
			StorageRef:     storageRef,
			StorageRefHash: datastore.StorageRefHash(storageRef),
			Reason:         meta.ObjectGCReasonOverwrite,
			NotBefore:      time.Now().UTC().Add(-time.Minute),
		}); err != nil {
			t.Fatal(err)
		}
	}

	worker := newObjectGCWorker(rt.meta, rt.pool, ObjectGCWorkerOptions{BatchSize: 2})
	if fatal := worker.processOnce(context.Background()); fatal {
		t.Fatal("processOnce returned fatal")
	}

	var processed int
	if err := rt.meta.DB().QueryRow(`SELECT COUNT(*) FROM object_gc_candidates
		WHERE namespace_id = ? AND last_error = ?`, "ns-batch", "namespace owner is not active").Scan(&processed); err != nil {
		t.Fatal(err)
	}
	if processed != 2 {
		t.Fatalf("processed candidates = %d, want 2", processed)
	}
}

func TestObjectGCWorkerOptionsCapOversizedBatch(t *testing.T) {
	maxOpts := (ObjectGCWorkerOptions{BatchSize: MaxObjectGCBatchSize}).normalized()
	if maxOpts.BatchSize != MaxObjectGCBatchSize {
		t.Fatalf("normalized maximum batch size = %d, want %d", maxOpts.BatchSize, MaxObjectGCBatchSize)
	}

	opts := (ObjectGCWorkerOptions{BatchSize: MaxObjectGCBatchSize + 1}).normalized()
	if opts.BatchSize != DefaultObjectGCBatchSize {
		t.Fatalf("normalized batch size = %d, want default %d", opts.BatchSize, DefaultObjectGCBatchSize)
	}
}

func TestObjectGCProcessOnceRetriesAfterTransientListError(t *testing.T) {
	calls := 0
	worker := &objectGCWorker{
		batchSize: 1,
		listDue: func(context.Context, time.Time, int) ([]meta.ObjectGCCandidate, error) {
			calls++
			if calls == 1 {
				return nil, errors.New("dial tcp: connection refused")
			}
			return nil, nil
		},
	}
	if fatal := worker.processOnce(context.Background()); fatal {
		t.Fatal("transient list error stopped object GC worker")
	}
	if fatal := worker.processOnce(context.Background()); fatal {
		t.Fatal("later object GC poll stopped worker")
	}
	if calls != 2 {
		t.Fatalf("list calls = %d, want 2", calls)
	}
}

func TestObjectGCPostponesWhenNamespaceOwnerInactive(t *testing.T) {
	rt := newObjectGCTestRuntime(t)
	rt.upsertNamespace(t, "ns-a", "owner-a")
	rt.insertTenant(t, "owner-a", meta.TenantProvisioning, meta.TenantKindLive, "ns-a")
	candidate := rt.enqueueCandidate(t, "ns-a", "blobs/inactive-owner")

	worker := &objectGCWorker{meta: rt.meta, pool: rt.pool}
	if err := worker.processCandidate(context.Background(), candidate); err != nil {
		t.Fatal(err)
	}

	state, lastError := objectGCCandidateRow(t, rt.meta.DB(), "ns-a", "blobs/inactive-owner")
	if state != meta.ObjectGCCandidatePending || lastError != "namespace owner is not active" {
		t.Fatalf("candidate state=%s last_error=%q", state, lastError)
	}
}

func TestObjectGCDeletesUnreachableBlob(t *testing.T) {
	rt := newObjectGCTestRuntime(t)
	rt.upsertNamespace(t, "ns-a", "owner-a")
	rt.insertTenant(t, "owner-a", meta.TenantActive, meta.TenantKindLive, "ns-a")
	candidate := rt.enqueueCandidate(t, "ns-a", "blobs/unreachable")

	worker := &objectGCWorker{meta: rt.meta, pool: rt.pool}
	if err := worker.processCandidate(context.Background(), candidate); err != nil {
		t.Fatal(err)
	}

	state, lastError := objectGCCandidateRow(t, rt.meta.DB(), "ns-a", "blobs/unreachable")
	if state != meta.ObjectGCCandidateDeleted || lastError != "" {
		t.Fatalf("candidate state=%s last_error=%q", state, lastError)
	}
}

func TestObjectGCRecoversKnownFailedPublication(t *testing.T) {
	rt := newObjectGCTestRuntime(t)
	ctx := context.Background()
	namespaceID := "ns-known-failed-publication"
	ownerID := "owner-known-failed-publication"
	storageRef := "blobs/known-failed-publication"
	payload := []byte("orphaned publication")
	rt.upsertNamespace(t, namespaceID, ownerID)
	rt.insertTenant(t, ownerID, meta.TenantActive, meta.TenantKindLive, namespaceID)
	owner := objectGCTenant(t, rt, ownerID)
	ownerBackend, release, err := rt.pool.Acquire(ctx, owner)
	if err != nil {
		t.Fatal(err)
	}
	if err := ownerBackend.S3().PutObject(ctx, storageRef, bytes.NewReader(payload), int64(len(payload)), s3client.EncryptionOpts{}); err != nil {
		release()
		t.Fatal(err)
	}
	release()

	notBefore := time.Now().UTC().Add(time.Hour)
	input := &meta.ObjectGCCandidateInput{
		NamespaceID:    namespaceID,
		StorageRef:     storageRef,
		StorageRefHash: datastore.StorageRefHash(storageRef),
		Reason:         meta.ObjectGCReasonFailedWrite,
		NotBefore:      notBefore,
	}
	if err := rt.meta.PrepareFailedWriteObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	if begun, err := rt.meta.BeginFailedWriteObjectGCCandidatePublish(ctx, namespaceID, input.StorageRefHash, storageRef); err != nil || !begun {
		t.Fatalf("begin publish: begun=%v err=%v", begun, err)
	}
	if err := rt.meta.RetryFailedWriteObjectGCCandidate(ctx, namespaceID, input.StorageRefHash, storageRef, notBefore, "synchronous delete failed"); err != nil {
		t.Fatal(err)
	}
	candidates, err := rt.meta.ListDueObjectGCCandidates(ctx, notBefore.Add(time.Second), 10)
	if err != nil || len(candidates) != 1 {
		t.Fatalf("recoverable candidates=%+v err=%v", candidates, err)
	}
	if candidates[0].State != meta.ObjectGCCandidatePending || candidates[0].LastError == "" {
		t.Fatalf("recoverable candidate=%+v, want pending with cleanup error", candidates[0])
	}
	worker := &objectGCWorker{meta: rt.meta, pool: rt.pool}
	if err := worker.processCandidate(ctx, candidates[0]); err != nil {
		t.Fatal(err)
	}
	ownerBackend, release, err = rt.pool.Acquire(ctx, owner)
	if err != nil {
		t.Fatal(err)
	}
	assertObjectGCS3ObjectMissing(t, ownerBackend, storageRef)
	release()

	if err := rt.meta.PrepareFailedWriteObjectGCCandidate(ctx, input); err != nil {
		t.Fatalf("prepare same storage ref after recovery: %v", err)
	}
	state, _ := objectGCCandidateRow(t, rt.meta.DB(), namespaceID, storageRef)
	if state != meta.ObjectGCCandidatePreparing {
		t.Fatalf("re-prepared candidate state=%s, want preparing", state)
	}
}

func TestObjectGCSkipsStaleSnapshotAfterFailedWriteRenewal(t *testing.T) {
	rt := newObjectGCTestRuntime(t)
	ctx := context.Background()
	namespaceID := "ns-stale-failed-write"
	ownerID := "owner-stale-failed-write"
	storageRef := "blobs/stale-failed-write"
	payload := []byte("stale snapshot must not delete")
	rt.upsertNamespace(t, namespaceID, ownerID)
	rt.insertTenant(t, ownerID, meta.TenantActive, meta.TenantKindLive, namespaceID)
	owner := objectGCTenant(t, rt, ownerID)
	ownerBackend, release, err := rt.pool.Acquire(ctx, owner)
	if err != nil {
		t.Fatal(err)
	}
	if err := ownerBackend.S3().PutObject(ctx, storageRef, bytes.NewReader(payload), int64(len(payload)), s3client.EncryptionOpts{}); err != nil {
		release()
		t.Fatal(err)
	}
	release()

	input := &meta.ObjectGCCandidateInput{
		NamespaceID:    namespaceID,
		StorageRef:     storageRef,
		StorageRefHash: datastore.StorageRefHash(storageRef),
		Reason:         meta.ObjectGCReasonFailedWrite,
		NotBefore:      time.Now().UTC().Add(-time.Minute),
	}
	if err := rt.meta.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	candidates, err := rt.meta.ListDueObjectGCCandidates(ctx, time.Now().UTC(), 100)
	if err != nil {
		t.Fatal(err)
	}
	var stale meta.ObjectGCCandidate
	for _, candidate := range candidates {
		if candidate.NamespaceID == namespaceID && candidate.StorageRef == storageRef {
			stale = candidate
			break
		}
	}
	if stale.StorageRef == "" {
		t.Fatalf("did not list stale candidate: %+v", candidates)
	}
	input.NotBefore = time.Now().UTC().Add(time.Hour)
	if err := rt.meta.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	worker := &objectGCWorker{meta: rt.meta, pool: rt.pool}
	if err := worker.processCandidate(ctx, stale); err != nil {
		t.Fatal(err)
	}
	state, lastError := objectGCCandidateRow(t, rt.meta.DB(), namespaceID, storageRef)
	if state != meta.ObjectGCCandidatePending || lastError != "" {
		t.Fatalf("refreshed candidate state=%s last_error=%q", state, lastError)
	}
	ownerBackend, release, err = rt.pool.Acquire(ctx, owner)
	if err != nil {
		t.Fatal(err)
	}
	defer release()
	assertObjectGCS3ObjectBytes(t, ownerBackend, storageRef, payload)
}

func TestObjectGCPostponesReachableBlob(t *testing.T) {
	rt := newObjectGCTestRuntime(t)
	ctx := context.Background()
	namespaceID := "ns-reachable"
	ownerID := "owner-reachable"
	path := "/reachable.bin"
	payload := deterministicObjectGCPayload(8*1024*1024, 0x61)

	rt.upsertNamespace(t, namespaceID, ownerID)
	rt.insertTenant(t, ownerID, meta.TenantActive, meta.TenantKindLive, namespaceID)
	owner := objectGCTenant(t, rt, ownerID)
	ownerBackend, release, err := rt.pool.Acquire(ctx, owner)
	if err != nil {
		t.Fatalf("acquire owner backend: %v", err)
	}
	if _, err := ownerBackend.Write(path, payload, 0, filesystem.WriteFlagCreate); err != nil {
		release()
		t.Fatalf("create s3 file: %v", err)
	}
	storageRef := objectGCFileStorageRef(t, ownerBackend, path)
	assertObjectGCS3ObjectBytes(t, ownerBackend, storageRef, payload)
	release()

	candidate := rt.enqueueCandidate(t, namespaceID, storageRef)
	worker := &objectGCWorker{meta: rt.meta, pool: rt.pool}
	if err := worker.processCandidate(ctx, candidate); err != nil {
		t.Fatal(err)
	}

	state, lastError := objectGCCandidateRow(t, rt.meta.DB(), namespaceID, storageRef)
	if state != meta.ObjectGCCandidatePending || lastError != "storage ref still reachable" {
		t.Fatalf("candidate state=%s last_error=%q", state, lastError)
	}
	ownerBackend, release, err = rt.pool.Acquire(ctx, owner)
	if err != nil {
		t.Fatalf("reacquire owner backend: %v", err)
	}
	defer release()
	assertObjectGCS3ObjectBytes(t, ownerBackend, storageRef, payload)
	assertObjectGCBackendVisibleBytes(t, ownerBackend, path, payload)
}

func TestObjectGCCompletesReachableFailedWriteCandidate(t *testing.T) {
	rt := newObjectGCTestRuntime(t)
	ctx := context.Background()
	namespaceID := "ns-reachable-failed-write"
	ownerID := "owner-reachable-failed-write"
	path := "/reachable-failed-write.bin"
	payload := deterministicObjectGCPayload(8*1024*1024, 0x62)

	rt.upsertNamespace(t, namespaceID, ownerID)
	rt.insertTenant(t, ownerID, meta.TenantActive, meta.TenantKindLive, namespaceID)
	owner := objectGCTenant(t, rt, ownerID)
	ownerBackend, release, err := rt.pool.Acquire(ctx, owner)
	if err != nil {
		t.Fatalf("acquire owner backend: %v", err)
	}
	if _, err := ownerBackend.Write(path, payload, 0, filesystem.WriteFlagCreate); err != nil {
		release()
		t.Fatalf("create s3 file: %v", err)
	}
	storageRef := objectGCFileStorageRef(t, ownerBackend, path)
	release()

	if err := rt.meta.EnqueueObjectGCCandidate(ctx, &meta.ObjectGCCandidateInput{
		NamespaceID:    namespaceID,
		StorageRef:     storageRef,
		StorageRefHash: datastore.StorageRefHash(storageRef),
		Reason:         meta.ObjectGCReasonFailedWrite,
		NotBefore:      time.Now().UTC().Add(-time.Minute),
	}); err != nil {
		t.Fatal(err)
	}
	candidates, err := rt.meta.ListDueObjectGCCandidates(ctx, time.Now().UTC(), 10)
	if err != nil || len(candidates) != 1 {
		t.Fatalf("failed-write candidates = %+v, err=%v", candidates, err)
	}
	worker := &objectGCWorker{meta: rt.meta, pool: rt.pool}
	if err := worker.processCandidate(ctx, candidates[0]); err != nil {
		t.Fatal(err)
	}

	var candidateRows int
	if err := rt.meta.DB().QueryRow(`SELECT COUNT(*) FROM object_gc_candidates
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?`,
		namespaceID, datastore.StorageRefHash(storageRef), storageRef).Scan(&candidateRows); err != nil {
		t.Fatal(err)
	}
	if candidateRows != 0 {
		t.Fatalf("reachable failed-write candidate rows = %d, want 0", candidateRows)
	}
	ownerBackend, release, err = rt.pool.Acquire(ctx, owner)
	if err != nil {
		t.Fatalf("reacquire owner backend: %v", err)
	}
	defer release()
	assertObjectGCS3ObjectBytes(t, ownerBackend, storageRef, payload)
	if err := ownerBackend.Remove(path); err != nil {
		t.Fatalf("delete committed file: %v", err)
	}
	if err := rt.meta.EnqueueObjectGCCandidate(ctx, &meta.ObjectGCCandidateInput{
		NamespaceID:    namespaceID,
		StorageRef:     storageRef,
		StorageRefHash: datastore.StorageRefHash(storageRef),
		Reason:         meta.ObjectGCReasonFileDelete,
		NotBefore:      time.Now().UTC().Add(-time.Minute),
	}); err != nil {
		t.Fatal(err)
	}
	candidates, err = rt.meta.ListDueObjectGCCandidates(ctx, time.Now().UTC().Add(48*time.Hour), 10)
	if err != nil || len(candidates) != 1 {
		t.Fatalf("post-delete candidates = %+v, err=%v", candidates, err)
	}
	if candidates[0].StorageRef != storageRef || candidates[0].Reason != meta.ObjectGCReasonFileDelete {
		t.Fatalf("post-delete candidate = %+v", candidates[0])
	}
	worker = &objectGCWorker{meta: rt.meta, pool: rt.pool}
	if err := worker.processCandidate(ctx, candidates[0]); err != nil {
		t.Fatal(err)
	}
	assertObjectGCS3ObjectMissing(t, ownerBackend, storageRef)
}

func TestObjectGCPostponesLayerObjectRetainedByCheckpoint(t *testing.T) {
	rt := newObjectGCTestRuntime(t)
	ctx := context.Background()
	namespaceID := "ns-layer-reachable"
	ownerID := "owner-layer-reachable"
	path := "/layer-reachable.bin"
	layerID := "layer/reachable"
	payload := []byte("checkpoint payload")

	rt.upsertNamespace(t, namespaceID, ownerID)
	rt.insertTenant(t, ownerID, meta.TenantActive, meta.TenantKindLive, namespaceID)
	owner := objectGCTenant(t, rt, ownerID)
	ownerBackend, release, err := rt.pool.Acquire(ctx, owner)
	if err != nil {
		t.Fatalf("acquire owner backend: %v", err)
	}
	defer release()

	if err := ownerBackend.Store().CreateFSLayer(ctx, &datastore.FSLayer{LayerID: layerID, BaseRootPath: "/"}); err != nil {
		t.Fatalf("create layer: %v", err)
	}
	entry, err := ownerBackend.PutFSLayerObject(ctx, layerID, path, bytes.NewReader(payload), int64(len(payload)))
	if err != nil {
		t.Fatalf("put layer object: %v", err)
	}
	entry.LayerID = layerID
	if err := ownerBackend.Store().UpsertFSLayerEntry(ctx, entry); err != nil {
		t.Fatalf("upsert layer entry: %v", err)
	}
	ownerBackend.CompleteFailedWriteObjectGC(ctx, entry.StorageRef)
	if err := ownerBackend.Store().CreateFSLayerCheckpoint(ctx, &datastore.FSLayerCheckpoint{
		CheckpointID: "checkpoint-reachable",
		LayerID:      layerID,
	}); err != nil {
		t.Fatalf("create checkpoint: %v", err)
	}
	if _, err := ownerBackend.WriteStoredObjectCtxIfRevision(ctx, path, entry); err != nil {
		t.Fatalf("commit layer object to base: %v", err)
	}
	if _, err := ownerBackend.Write(path, []byte("replacement"), 0, filesystem.WriteFlagTruncate); err != nil {
		t.Fatalf("overwrite base file: %v", err)
	}

	candidates, err := rt.meta.ListDueObjectGCCandidates(ctx, time.Now().UTC().Add(48*time.Hour), 10)
	if err != nil {
		t.Fatalf("list layer object GC candidate: %v", err)
	}
	if len(candidates) != 1 || candidates[0].StorageRef != entry.StorageRef {
		t.Fatalf("candidates = %+v, want layer ref %q", candidates, entry.StorageRef)
	}
	worker := &objectGCWorker{meta: rt.meta, pool: rt.pool}
	if err := worker.processCandidate(ctx, candidates[0]); err != nil {
		t.Fatal(err)
	}

	state, lastError := objectGCCandidateRow(t, rt.meta.DB(), namespaceID, entry.StorageRef)
	if state != meta.ObjectGCCandidatePending || lastError != "storage ref still reachable" {
		t.Fatalf("candidate state=%s last_error=%q", state, lastError)
	}
	got, err := ownerBackend.ReadFSLayerEntryData(ctx, entry)
	if err != nil {
		t.Fatalf("read checkpoint layer object after GC: %v", err)
	}
	if !bytes.Equal(got, payload) {
		t.Fatalf("checkpoint payload = %q, want %q", got, payload)
	}
}

func TestObjectGCSweepsTierTransitionObsoleteBlobKeepsCurrentContent(t *testing.T) {
	rt := newObjectGCTestRuntime(t)
	ctx := context.Background()
	namespaceID := "ns-tier-sweep"
	ownerID := "owner-tier-sweep"
	path := "/tier-sweep.bin"
	initialInline := deterministicObjectGCPayload(10*1024, 0x12)
	largeS3 := deterministicObjectGCPayload(8*1024*1024, 0x34)
	finalInline := deterministicObjectGCPayload(10*1024, 0x56)

	rt.upsertNamespace(t, namespaceID, ownerID)
	rt.insertTenant(t, ownerID, meta.TenantActive, meta.TenantKindLive, namespaceID)
	owner := objectGCTenant(t, rt, ownerID)
	ownerBackend, release, err := rt.pool.Acquire(ctx, owner)
	if err != nil {
		t.Fatalf("acquire owner backend: %v", err)
	}
	if _, err := ownerBackend.Write(path, initialInline, 0, filesystem.WriteFlagCreate); err != nil {
		release()
		t.Fatalf("create inline file: %v", err)
	}
	if _, err := ownerBackend.Write(path, largeS3, 0, filesystem.WriteFlagTruncate); err != nil {
		release()
		t.Fatalf("overwrite inline -> s3: %v", err)
	}
	obsoleteRef := objectGCFileStorageRef(t, ownerBackend, path)
	assertObjectGCS3ObjectBytes(t, ownerBackend, obsoleteRef, largeS3)
	if _, err := ownerBackend.Write(path, finalInline, 0, filesystem.WriteFlagTruncate); err != nil {
		release()
		t.Fatalf("overwrite s3 -> inline: %v", err)
	}
	assertObjectGCBackendVisibleBytes(t, ownerBackend, path, finalInline)
	if reachable, err := ownerBackend.HasConfirmedS3StorageRef(ctx, datastore.StorageRefHash(obsoleteRef), obsoleteRef); err != nil {
		release()
		t.Fatalf("check obsolete ref reachability: %v", err)
	} else if reachable {
		release()
		t.Fatalf("obsolete ref %q is still confirmed after inline overwrite", obsoleteRef)
	}
	release()

	candidates, err := rt.meta.ListDueObjectGCCandidates(ctx, time.Now().UTC().Add(8*24*time.Hour), 10)
	if err != nil {
		t.Fatalf("list due object gc candidates: %v", err)
	}
	if len(candidates) != 1 {
		t.Fatalf("candidate count = %d, want 1: %+v", len(candidates), candidates)
	}
	candidate := candidates[0]
	if candidate.NamespaceID != namespaceID || candidate.StorageRef != obsoleteRef || candidate.Reason != meta.ObjectGCReasonOverwrite {
		t.Fatalf("candidate = %+v, want namespace=%q ref=%q reason=%q", candidate, namespaceID, obsoleteRef, meta.ObjectGCReasonOverwrite)
	}

	worker := &objectGCWorker{meta: rt.meta, pool: rt.pool}
	if err := worker.processCandidate(ctx, candidate); err != nil {
		t.Fatal(err)
	}

	state, lastError := objectGCCandidateRow(t, rt.meta.DB(), namespaceID, obsoleteRef)
	if state != meta.ObjectGCCandidateDeleted || lastError != "" {
		t.Fatalf("candidate state=%s last_error=%q", state, lastError)
	}
	ownerBackend, release, err = rt.pool.Acquire(ctx, owner)
	if err != nil {
		t.Fatalf("reacquire owner backend: %v", err)
	}
	defer release()
	assertObjectGCS3ObjectMissing(t, ownerBackend, obsoleteRef)
	assertObjectGCBackendVisibleBytes(t, ownerBackend, path, finalInline)
	currentRef := objectGCFileStorageRef(t, ownerBackend, path)
	if currentRef != "inline" {
		t.Fatalf("current storage ref = %q, want inline", currentRef)
	}
}

func objectGCTenant(t *testing.T, rt *objectGCTestRuntime, tenantID string) *meta.Tenant {
	t.Helper()
	tenant, err := rt.meta.GetTenant(context.Background(), tenantID)
	if err != nil {
		t.Fatalf("get tenant %s: %v", tenantID, err)
	}
	return tenant
}

func objectGCFileStorageRef(t *testing.T, backendFS *backendpkg.Dat9Backend, path string) string {
	t.Helper()
	node, err := backendFS.Store().Stat(context.Background(), path)
	if err != nil {
		t.Fatalf("stat %s: %v", path, err)
	}
	if node.File == nil {
		t.Fatalf("stat %s returned no file", path)
	}
	return node.File.StorageRef
}

func assertObjectGCBackendVisibleBytes(t *testing.T, backendFS *backendpkg.Dat9Backend, path string, want []byte) {
	t.Helper()
	info, err := backendFS.Stat(path)
	if err != nil {
		t.Fatalf("stat %s: %v", path, err)
	}
	if info.Size != int64(len(want)) {
		t.Fatalf("visible size for %s = %d, want %d", path, info.Size, len(want))
	}
	got, err := backendFS.Read(path, 0, int64(len(want)+1))
	if err != nil {
		t.Fatalf("read %s: %v", path, err)
	}
	if !bytes.Equal(got, want) {
		t.Fatalf("visible bytes for %s mismatch: got %d bytes want %d bytes", path, len(got), len(want))
	}
}

func assertObjectGCS3ObjectBytes(t *testing.T, backendFS *backendpkg.Dat9Backend, storageRef string, want []byte) {
	t.Helper()
	reader, err := backendFS.S3().GetObject(context.Background(), storageRef)
	if err != nil {
		t.Fatalf("get s3 object %q: %v", storageRef, err)
	}
	defer func() { _ = reader.Close() }()
	got, err := io.ReadAll(reader)
	if err != nil {
		t.Fatalf("read s3 object %q: %v", storageRef, err)
	}
	if !bytes.Equal(got, want) {
		t.Fatalf("s3 object %q mismatch: got %d bytes want %d bytes", storageRef, len(got), len(want))
	}
}

func assertObjectGCS3ObjectMissing(t *testing.T, backendFS *backendpkg.Dat9Backend, storageRef string) {
	t.Helper()
	reader, err := backendFS.S3().GetObject(context.Background(), storageRef)
	if err == nil {
		_ = reader.Close()
		t.Fatalf("s3 object %q still exists after object gc", storageRef)
	}
}

func deterministicObjectGCPayload(size int, seed byte) []byte {
	out := make([]byte, size)
	for idx := range out {
		out[idx] = byte((idx*29 + int(seed)) % 251)
	}
	return out
}