package content

import (
	"context"
	"errors"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

func newTestStore(t *testing.T) *Store {
	t.Helper()
	db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "contentstore.db")), &gorm.Config{})
	require.NoError(t, err)
	require.NoError(t, db.AutoMigrate(&Authority{}, &VersionedContent{}, &ContentVersion{}, &VersionOperation{}, &VersionPath{}, &VersionPin{}, &Outbox{}, &StagedPayloadGC{}))
	return New(db)
}

func TestPublishVersionRejectsStaleLegacyAuthorityFenceBeforeHeadAdvance(t *testing.T) {
	ctx := context.Background()
	store := newTestStore(t)
	now := time.Date(2026, 8, 23, 4, 0, 0, 0, time.UTC)
	_, err := store.CreateVersionedContent(ctx, VersionedContent{
		ID: "ctn_1", WorkspaceID: "wsp_1", Domain: "automation", BindingKey: "team/repo", Model: ModelTree,
		Generation: 1, MaxRetainedVersions: 4, MaxStoredPayloadBytes: 10, MaxPins: 2,
		CreatedBy: "usr_1", CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)
	stale, err := store.AcquireLegacyAuthorityFence(ctx, AcquireLegacyAuthorityFenceParams{
		WorkspaceID: "wsp_1", Domain: "automation", BindingKey: "team/repo",
		Token: "migration-1", Actor: "usr_1", Now: now, ExpiresAt: now.Add(time.Minute),
	})
	require.NoError(t, err)
	currentNow := now.Add(2 * time.Minute)
	current, err := store.AcquireLegacyAuthorityFence(ctx, AcquireLegacyAuthorityFenceParams{
		WorkspaceID: "wsp_1", Domain: "automation", BindingKey: "team/repo",
		Token: "migration-2", Actor: "usr_1", Now: currentNow, ExpiresAt: currentNow.Add(time.Minute),
	})
	require.NoError(t, err)
	require.NoError(t, store.RegisterStagedPayloadGC(ctx, StagedPayloadGC{
		ID: "ctg_1", WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: "ctv_1", Model: ModelTree,
		PayloadRef: "wsp_1/ctn_1/ctv_1", DeleteAfter: currentNow.Add(time.Hour),
	}))
	params := PublishVersionParams{
		ContentID: "ctn_1", WorkspaceID: "wsp_1", ExpectedGeneration: 1, Actor: "usr_1", PublishedAt: currentNow,
		Version: ContentVersion{ID: "ctv_1", Model: ModelTree, RootDigest: strings.Repeat("a", 64), PayloadRef: "wsp_1/ctn_1/ctv_1", SizeBytes: 1},
		Outbox:  Outbox{ID: "cto_1"}, GC: StagedPayloadGC{ID: "ctg_1"}, AuthorityFence: &stale,
	}
	_, err = store.PublishVersion(ctx, params)
	require.ErrorIs(t, err, ErrAuthorityConflict)
	content, err := store.GetVersionedContent(ctx, "wsp_1", "ctn_1")
	require.NoError(t, err)
	require.Empty(t, content.HeadVersionID)
	var versions int64
	require.NoError(t, store.db.Model(&ContentVersion{}).Where("content_id = ?", "ctn_1").Count(&versions).Error)
	require.Zero(t, versions)

	params.AuthorityFence = &current
	published, err := store.PublishVersion(ctx, params)
	require.NoError(t, err)
	require.Equal(t, "ctv_1", published.HeadVersionID)
}

func TestPublishDetachedVersionKeepsHeadAndReusesPublicationTransaction(t *testing.T) {
	ctx := context.Background()
	store := newTestStore(t)
	now := time.Date(2026, 8, 22, 1, 0, 0, 0, time.UTC)
	_, err := store.CreateVersionedContent(ctx, VersionedContent{
		ID: "ctn_1", WorkspaceID: "wsp_1", Domain: "memory", BindingKey: "mem_1", Model: ModelTree,
		HeadVersionID: "ctv_live", Revision: 2, Generation: 1, RetainedVersionCount: 2, StoredPayloadBytes: 2,
		MaxRetainedVersions: 4, MaxStoredPayloadBytes: 10, MaxPins: 2, CreatedBy: "usr_1", CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)
	require.NoError(t, store.db.Create(&[]ContentVersion{
		{ID: "ctv_parent", WorkspaceID: "wsp_1", ContentID: "ctn_1", Generation: 1, Model: ModelTree, PayloadRef: "wsp_1/ctn_1/ctv_parent", RootDigest: strings.Repeat("a", 64), SizeBytes: 1, CreatedBy: "usr_1", CreatedAt: now},
		{ID: "ctv_live", WorkspaceID: "wsp_1", ContentID: "ctn_1", ParentVersionID: "ctv_parent", Generation: 1, Model: ModelTree, PayloadRef: "wsp_1/ctn_1/ctv_live", RootDigest: strings.Repeat("b", 64), SizeBytes: 1, CreatedBy: "usr_1", CreatedAt: now},
	}).Error)
	require.NoError(t, store.RegisterStagedPayloadGC(ctx, StagedPayloadGC{ID: "ctg_detached", WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: "ctv_detached", Model: ModelTree, PayloadRef: "wsp_1/ctn_1/ctv_detached", DeleteAfter: now.Add(time.Hour)}))

	params := PublishDetachedVersionParams{
		PublishVersionParams: PublishVersionParams{
			ContentID: "ctn_1", WorkspaceID: "wsp_1", ExpectedHeadVersionID: "ctv_live", ExpectedGeneration: 1,
			Actor: "usr_1", PublishedAt: now.Add(time.Minute),
			Version: ContentVersion{ID: "ctv_detached", Model: ModelTree, RootDigest: strings.Repeat("c", 64), PayloadRef: "wsp_1/ctn_1/ctv_detached", SizeBytes: 3, FileCount: 1},
			Paths:   []VersionPath{{Path: "MEMORY.md", Digest: strings.Repeat("d", 64), Kind: "file", SizeBytes: 3}},
			Outbox:  Outbox{ID: "cto_detached"}, GC: StagedPayloadGC{ID: "ctg_detached"},
			Operation: &VersionOperation{
				ID: "ctop_detached", OperationKey: "content.migration:detached", RequestDigest: strings.Repeat("e", 64),
				OperationNamespace: "content.migration", OperationVersion: 1,
				BoundsPolicyNamespace: "content.tree", BoundsPolicyVersion: 1, MaxFiles: 10, MaxFileBytes: 10, MaxTotalBytes: 10, MaxManifestBytes: 10,
			},
		},
		ParentVersionID: "ctv_parent",
	}
	published, err := store.PublishDetachedVersion(ctx, params)
	require.NoError(t, err)
	require.Equal(t, "ctv_live", published.HeadVersionID)
	require.Equal(t, uint64(3), published.Revision)
	require.Equal(t, uint64(3), published.RetainedVersionCount)
	require.Equal(t, int64(5), published.StoredPayloadBytes)
	require.Equal(t, "ctv_parent", published.Version.ParentVersionID)

	var path VersionPath
	require.NoError(t, store.db.Where("version_id = ?", "ctv_detached").First(&path).Error)
	var outbox Outbox
	require.NoError(t, store.db.Where("id = ?", "cto_detached").First(&outbox).Error)
	require.Equal(t, OutboxVersionCommitted, outbox.EventType)
	var gc StagedPayloadGC
	require.NoError(t, store.db.Where("id = ?", "ctg_detached").First(&gc).Error)
	require.Equal(t, GCStatusRetained, gc.Status)
	var operation VersionOperation
	require.NoError(t, store.db.Where("operation_key = ?", "content.migration:detached").First(&operation).Error)
	require.Equal(t, uint64(3), operation.ContentRevision)

	replayed, err := store.PublishDetachedVersion(ctx, params)
	require.NoError(t, err)
	require.True(t, replayed.Replayed)
	require.Equal(t, "ctv_live", replayed.HeadVersionID)
	require.Equal(t, uint64(3), replayed.Revision)
	var versionCount int64
	require.NoError(t, store.db.Model(&ContentVersion{}).Where("content_id = ?", "ctn_1").Count(&versionCount).Error)
	require.Equal(t, int64(3), versionCount)
	require.NoError(t, store.db.Model(&ContentVersion{}).Where("id = ?", "ctv_parent").Update("generation", 2).Error)
	_, err = store.PublishDetachedVersion(ctx, params)
	require.ErrorIs(t, err, ErrParentConflict, "receipt replay must revalidate the exact parent")
	require.NoError(t, store.db.Model(&ContentVersion{}).Where("id = ?", "ctv_parent").Update("generation", 1).Error)

	require.NoError(t, store.db.Model(&VersionedContent{}).Where("id = ?", "ctn_1").Update("head_version_id", "ctv_detached").Error)
	_, err = store.PublishDetachedVersion(ctx, params)
	require.ErrorIs(t, err, ErrHeadConflict, "receipt replay must still honor the live-head fence")
}

func TestPublishDetachedVersionRejectsForeignParentAndQuotaWithoutMutation(t *testing.T) {
	ctx := context.Background()
	store := newTestStore(t)
	now := time.Now().UTC()
	for _, content := range []VersionedContent{
		{ID: "ctn_1", WorkspaceID: "wsp_1", Domain: "memory", BindingKey: "mem_1", Model: ModelTree, Generation: 1, MaxRetainedVersions: 1, MaxStoredPayloadBytes: 1, MaxPins: 1, CreatedBy: "usr_1", CreatedAt: now, UpdatedAt: now},
		{ID: "ctn_2", WorkspaceID: "wsp_2", Domain: "memory", BindingKey: "mem_2", Model: ModelTree, Generation: 1, CreatedBy: "usr_1", CreatedAt: now, UpdatedAt: now},
	} {
		_, err := store.CreateVersionedContent(ctx, content)
		require.NoError(t, err)
	}
	require.NoError(t, store.db.Create(&ContentVersion{ID: "ctv_foreign", WorkspaceID: "wsp_2", ContentID: "ctn_2", Generation: 1, Model: ModelTree, PayloadRef: "wsp_2/ctn_2/ctv_foreign", RootDigest: strings.Repeat("a", 64), CreatedBy: "usr_1", CreatedAt: now}).Error)
	require.NoError(t, store.db.Create(&ContentVersion{ID: "ctv_wrong_generation", WorkspaceID: "wsp_1", ContentID: "ctn_1", Generation: 2, Model: ModelTree, PayloadRef: "wsp_1/ctn_1/ctv_wrong_generation", RootDigest: strings.Repeat("c", 64), CreatedBy: "usr_1", CreatedAt: now}).Error)

	register := func(id string) {
		require.NoError(t, store.RegisterStagedPayloadGC(ctx, StagedPayloadGC{ID: "ctg_" + id, WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: id, Model: ModelTree, PayloadRef: "wsp_1/ctn_1/" + id, DeleteAfter: now.Add(time.Hour)}))
	}
	publish := func(id, parent string, size int64) error {
		register(id)
		_, err := store.PublishDetachedVersion(ctx, PublishDetachedVersionParams{PublishVersionParams: PublishVersionParams{
			ContentID: "ctn_1", WorkspaceID: "wsp_1", ExpectedGeneration: 1, Actor: "usr_1", PublishedAt: now,
			Version: ContentVersion{ID: id, Model: ModelTree, RootDigest: strings.Repeat("b", 64), PayloadRef: "wsp_1/ctn_1/" + id, SizeBytes: size},
			Outbox:  Outbox{ID: "cto_" + id}, GC: StagedPayloadGC{ID: "ctg_" + id},
		}, ParentVersionID: parent})
		return err
	}
	require.ErrorIs(t, publish("ctv_cross", "ctv_foreign", 0), ErrParentConflict)
	require.ErrorIs(t, publish("ctv_generation", "ctv_wrong_generation", 0), ErrParentConflict)
	require.ErrorIs(t, publish("ctv_quota", "", 2), ErrQuotaExceeded)

	content, err := store.GetVersionedContent(ctx, "wsp_1", "ctn_1")
	require.NoError(t, err)
	require.Empty(t, content.HeadVersionID)
	require.Zero(t, content.Revision)
	require.Zero(t, content.RetainedVersionCount)
	var count int64
	require.NoError(t, store.db.Model(&ContentVersion{}).Where("content_id = ?", "ctn_1").Count(&count).Error)
	require.Equal(t, int64(1), count, "only the pre-existing wrong-generation fixture remains")
}

func TestPublishVersionCommitsCASQuotaPathsAndOutbox(t *testing.T) {
	ctx := context.Background()
	store := newTestStore(t)
	now := time.Date(2026, 8, 10, 20, 0, 0, 0, time.UTC)
	content, err := store.CreateVersionedContent(ctx, VersionedContent{
		ID: "ctn_1", WorkspaceID: "wsp_1", Domain: "skill", BindingKey: "skill_1", Model: ModelTree,
		CreatedBy: "usr_1", CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)
	require.Equal(t, DefaultTreeVersions, content.MaxRetainedVersions)
	require.NoError(t, store.RegisterStagedPayloadGC(ctx, StagedPayloadGC{ID: "ctg_1", WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: "ctv_1", Model: ModelTree, PayloadRef: "wsp_1/ctn_1/ctv_1", DeleteAfter: now.Add(time.Hour)}))

	published, err := store.PublishVersion(ctx, PublishVersionParams{
		ContentID: "ctn_1", WorkspaceID: "wsp_1", ExpectedGeneration: 1, Actor: "usr_1", PublishedAt: now,
		Version: ContentVersion{ID: "ctv_1", Model: ModelTree, RootDigest: strings.Repeat("a", 64), PayloadRef: "wsp_1/ctn_1/ctv_1", SizeBytes: 12, FileCount: 1},
		Paths:   []VersionPath{{Path: "README.md", Digest: strings.Repeat("b", 64), Kind: "file", SizeBytes: 12}},
		Outbox:  Outbox{ID: "cto_1"}, GC: StagedPayloadGC{ID: "ctg_1"},
	})
	require.NoError(t, err)
	require.Equal(t, "ctv_1", published.HeadVersionID)
	require.Equal(t, uint64(1), published.RetainedVersionCount)
	require.Equal(t, int64(12), published.StoredPayloadBytes)

	history, err := store.PathHistory(ctx, "wsp_1", "ctn_1", "README.md", 10)
	require.NoError(t, err)
	require.Len(t, history, 1)
	require.Equal(t, "ctv_1", history[0].VersionID)

	var outbox Outbox
	require.NoError(t, store.db.Where("id = ?", "cto_1").First(&outbox).Error)
	require.Equal(t, OutboxVersionCommitted, outbox.EventType)

	_, err = store.PublishVersion(ctx, PublishVersionParams{
		ContentID: "ctn_1", WorkspaceID: "wsp_1", ExpectedGeneration: 1, Actor: "usr_2", PublishedAt: now.Add(time.Minute),
		Version: ContentVersion{ID: "ctv_stale", Model: ModelTree, RootDigest: strings.Repeat("c", 64), PayloadRef: "wsp_1/ctn_1/ctv_stale"},
		Outbox:  Outbox{ID: "cto_stale"}, GC: StagedPayloadGC{ID: "ctg_stale"},
	})
	require.ErrorIs(t, err, ErrHeadConflict)
}

func TestPublishVersionRejectsQuotaWithoutMutatingHead(t *testing.T) {
	ctx := context.Background()
	store := newTestStore(t)
	now := time.Now().UTC()
	_, err := store.CreateVersionedContent(ctx, VersionedContent{
		ID: "ctn_1", WorkspaceID: "wsp_1", Domain: "memory", BindingKey: "mem_1", Model: ModelDocument,
		MaxRetainedVersions: 1, MaxStoredPayloadBytes: 4, MaxPins: 1, CreatedBy: "usr_1", CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)
	require.NoError(t, store.RegisterStagedPayloadGC(ctx, StagedPayloadGC{ID: "ctg_1", WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: "ctv_1", Model: ModelDocument, PayloadRef: "wsp_1/ctn_1/ctv_1", DeleteAfter: now.Add(time.Hour)}))
	_, err = store.PublishVersion(ctx, PublishVersionParams{
		ContentID: "ctn_1", WorkspaceID: "wsp_1", ExpectedGeneration: 1, Actor: "usr_1", PublishedAt: now,
		Version: ContentVersion{ID: "ctv_1", Model: ModelDocument, RootDigest: strings.Repeat("a", 64), PayloadRef: "wsp_1/ctn_1/ctv_1", SizeBytes: 5},
		Outbox:  Outbox{ID: "cto_1"}, GC: StagedPayloadGC{ID: "ctg_1"},
	})
	require.ErrorIs(t, err, ErrQuotaExceeded)
	content, err := store.GetVersionedContent(ctx, "wsp_1", "ctn_1")
	require.NoError(t, err)
	require.Empty(t, content.HeadVersionID)
	require.Zero(t, content.RetainedVersionCount)
}

func TestContentOutboxLeaseIsFencedByOwner(t *testing.T) {
	ctx := context.Background()
	store := newTestStore(t)
	now := time.Now().UTC()
	require.NoError(t, store.db.Create(&Outbox{
		ID: "cto_1", WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: "ctv_1", EventType: OutboxVersionCommitted,
		Status: OutboxStatusPending, NextAttemptAt: now, CreatedAt: now, UpdatedAt: now,
	}).Error)

	leased, err := store.LeaseDueOutbox(ctx, now, "worker-1", now.Add(time.Minute), 10)
	require.NoError(t, err)
	require.Len(t, leased, 1)
	competing, err := store.LeaseDueOutbox(ctx, now, "worker-2", now.Add(time.Minute), 10)
	require.NoError(t, err)
	require.Empty(t, competing)

	updated, err := store.MarkOutboxDelivered(ctx, leased[0], "worker-2", now)
	require.NoError(t, err)
	require.False(t, updated)
	updated, err = store.MarkOutboxDelivered(ctx, leased[0], "worker-1", now)
	require.NoError(t, err)
	require.True(t, updated)
}

func TestVersionPinsEnforceExactVersionWorkspaceAndConfiguredLimit(t *testing.T) {
	ctx := context.Background()
	store := newTestStore(t)
	now := time.Now().UTC()
	_, err := store.CreateVersionedContent(ctx, VersionedContent{
		ID: "ctn_1", WorkspaceID: "wsp_1", Domain: "memory", BindingKey: "mem_1", Model: ModelDocument,
		MaxRetainedVersions: 2, MaxStoredPayloadBytes: 1024, MaxPins: 1, CreatedBy: "usr_1", CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)
	require.NoError(t, store.db.Create(&[]ContentVersion{
		{ID: "ctv_1", WorkspaceID: "wsp_1", ContentID: "ctn_1", Generation: 1, Model: ModelDocument, PayloadRef: "wsp_1/ctn_1/ctv_1", RootDigest: strings.Repeat("a", 64), CreatedBy: "usr_1", CreatedAt: now},
		{ID: "ctv_2", WorkspaceID: "wsp_1", ContentID: "ctn_1", Generation: 1, Model: ModelDocument, PayloadRef: "wsp_1/ctn_1/ctv_2", RootDigest: strings.Repeat("b", 64), CreatedBy: "usr_1", CreatedAt: now.Add(time.Minute)},
	}).Error)

	created, err := store.CreateVersionPin(ctx, VersionPin{ID: "ctp_1", WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: "ctv_1", Label: "backup", CreatedBy: "usr_1", CreatedAt: now})
	require.NoError(t, err)
	require.Equal(t, "ctv_1", created.VersionID)
	_, err = store.CreateVersionPin(ctx, VersionPin{ID: "ctp_duplicate", WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: "ctv_1", Label: "backup", CreatedBy: "usr_1", CreatedAt: now})
	require.ErrorIs(t, err, ErrPinConflict)
	_, err = store.CreateVersionPin(ctx, VersionPin{ID: "ctp_2", WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: "ctv_2", Label: "other", CreatedBy: "usr_1", CreatedAt: now})
	require.ErrorIs(t, err, ErrQuotaExceeded)
	_, err = store.CreateVersionPin(ctx, VersionPin{ID: "ctp_cross", WorkspaceID: "wsp_2", ContentID: "ctn_1", VersionID: "ctv_1", Label: "cross", CreatedBy: "usr_1", CreatedAt: now})
	require.ErrorIs(t, err, gorm.ErrRecordNotFound)

	pins, err := store.ListVersionPins(ctx, "wsp_1", "ctn_1", 10)
	require.NoError(t, err)
	require.Len(t, pins, 1)
	require.NoError(t, store.DeleteVersionPin(ctx, "wsp_1", "ctn_1", created.ID))
	require.ErrorIs(t, store.DeleteVersionPin(ctx, "wsp_1", "ctn_1", created.ID), gorm.ErrRecordNotFound)
}

func TestStagedPayloadGCDeletionFencePreventsLatePublish(t *testing.T) {
	ctx := context.Background()
	store := newTestStore(t)
	now := time.Now().UTC()
	_, err := store.CreateVersionedContent(ctx, VersionedContent{
		ID: "ctn_1", WorkspaceID: "wsp_1", Domain: "skill", BindingKey: "skill_1", Model: ModelTree,
		CreatedBy: "usr_1", CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)
	require.NoError(t, store.RegisterStagedPayloadGC(ctx, StagedPayloadGC{
		ID: "ctg_1", WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: "ctv_1", Model: ModelTree,
		PayloadRef: "wsp_1/ctn_1/ctv_1", DeleteAfter: now,
	}))
	leased, err := store.LeaseDueStagedPayloadGCForBinding(ctx, GCBindingScope{WorkspaceID: "wsp_1", Domain: "skill", BindingKey: "skill_1"}, now, "worker-1", now.Add(time.Minute), 10)
	require.NoError(t, err)
	require.Len(t, leased, 1)
	action, err := store.PrepareStagedPayloadGC(ctx, leased[0], "worker-1", now)
	require.NoError(t, err)
	require.Equal(t, GCActionDelete, action)

	_, err = store.PublishVersion(ctx, PublishVersionParams{
		ContentID: "ctn_1", WorkspaceID: "wsp_1", ExpectedGeneration: 1, Actor: "usr_1", PublishedAt: now,
		Version: ContentVersion{ID: "ctv_1", Model: ModelTree, RootDigest: strings.Repeat("a", 64), PayloadRef: "wsp_1/ctn_1/ctv_1"},
		Outbox:  Outbox{ID: "cto_1"}, GC: StagedPayloadGC{ID: "ctg_1"},
	})
	require.ErrorContains(t, err, "expected one pending row")
	_, err = store.GetContentVersion(ctx, "wsp_1", "ctn_1", "ctv_1")
	require.ErrorIs(t, err, gorm.ErrRecordNotFound)
	content, err := store.GetVersionedContent(ctx, "wsp_1", "ctn_1")
	require.NoError(t, err)
	require.Empty(t, content.HeadVersionID)
}

func TestStagedPayloadGCReconciliationRetainsReachableVersion(t *testing.T) {
	ctx := context.Background()
	store := newTestStore(t)
	now := time.Now().UTC()
	_, err := store.CreateVersionedContent(ctx, VersionedContent{
		ID: "ctn_1", WorkspaceID: "wsp_1", Domain: "skill", BindingKey: "skill_1", Model: ModelTree,
		CreatedBy: "usr_1", CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)
	require.NoError(t, store.db.Create(&ContentVersion{
		ID: "ctv_1", WorkspaceID: "wsp_1", ContentID: "ctn_1", Generation: 1, Model: ModelTree,
		PayloadRef: "wsp_1/ctn_1/ctv_1", RootDigest: strings.Repeat("a", 64), CreatedBy: "usr_1", CreatedAt: now,
	}).Error)
	require.NoError(t, store.RegisterStagedPayloadGC(ctx, StagedPayloadGC{
		ID: "ctg_1", WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: "ctv_1", Model: ModelTree,
		PayloadRef: "wsp_1/ctn_1/ctv_1", DeleteAfter: now,
	}))
	leased, err := store.LeaseDueStagedPayloadGCForBinding(ctx, GCBindingScope{WorkspaceID: "wsp_1", Domain: "skill", BindingKey: "skill_1"}, now, "worker-1", now.Add(time.Minute), 10)
	require.NoError(t, err)
	require.Len(t, leased, 1)
	action, err := store.PrepareStagedPayloadGC(ctx, leased[0], "worker-1", now)
	require.NoError(t, err)
	require.Equal(t, GCActionRetained, action)
	var row StagedPayloadGC
	require.NoError(t, store.db.Where("id = ?", "ctg_1").First(&row).Error)
	require.Equal(t, GCStatusRetained, row.Status)
	require.Empty(t, row.LeaseOwner)
}

func TestRetentionReviewCandidatesExcludeHeadPinsAndUndeliveredEvents(t *testing.T) {
	ctx := context.Background()
	store := newTestStore(t)
	now := time.Now().UTC()
	_, err := store.CreateVersionedContent(ctx, VersionedContent{
		ID: "ctn_1", WorkspaceID: "wsp_1", Domain: "memory", BindingKey: "mem_1", Model: ModelDocument,
		CreatedBy: "usr_1", CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)
	versions := make([]ContentVersion, 0, 4)
	outbox := make([]Outbox, 0, 4)
	for index, id := range []string{"ctv_1", "ctv_2", "ctv_3", "ctv_4"} {
		createdAt := now.Add(time.Duration(index) * time.Minute)
		versions = append(versions, ContentVersion{
			ID: id, WorkspaceID: "wsp_1", ContentID: "ctn_1", Generation: 1, Model: ModelDocument,
			PayloadRef: "wsp_1/ctn_1/" + id, RootDigest: strings.Repeat(string(rune('a'+index)), 64), CreatedBy: "usr_1", CreatedAt: createdAt,
		})
		status := OutboxStatusDelivered
		if id == "ctv_4" {
			status = OutboxStatusPending
		}
		outbox = append(outbox, Outbox{
			ID: "cto_" + id, WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: id,
			EventType: OutboxVersionCommitted, Status: status, NextAttemptAt: createdAt, CreatedAt: createdAt, UpdatedAt: createdAt,
		})
	}
	require.NoError(t, store.db.Create(&versions).Error)
	require.NoError(t, store.db.Create(&outbox).Error)
	require.NoError(t, store.db.Model(&VersionedContent{}).Where("id = ?", "ctn_1").Update("head_version_id", "ctv_3").Error)
	_, err = store.CreateVersionPin(ctx, VersionPin{
		ID: "ctp_1", WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: "ctv_1", Label: "backup", CreatedBy: "usr_1", CreatedAt: now,
	})
	require.NoError(t, err)

	candidates, err := store.ListRetentionReviewCandidates(ctx, "wsp_1", "ctn_1", 10)
	require.NoError(t, err)
	require.Len(t, candidates, 1)
	require.Equal(t, "ctv_2", candidates[0].ID)
}

func TestStagedPayloadGCFailureIsBoundedAndTerminal(t *testing.T) {
	ctx := context.Background()
	store := newTestStore(t)
	now := time.Now().UTC()
	_, err := store.CreateVersionedContent(ctx, VersionedContent{
		ID: "ctn_1", WorkspaceID: "wsp_1", Domain: "skill", BindingKey: "skill_1", Model: ModelTree,
		CreatedBy: "usr_1", CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)
	require.NoError(t, store.RegisterStagedPayloadGC(ctx, StagedPayloadGC{
		ID: "ctg_1", WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: "ctv_1", Model: ModelTree,
		PayloadRef: "wsp_1/ctn_1/ctv_1", DeleteAfter: now,
	}))
	require.NoError(t, store.db.Model(&StagedPayloadGC{}).Where("id = ?", "ctg_1").Update("attempts", MaxGCAttempts-1).Error)
	leased, err := store.LeaseDueStagedPayloadGCForBinding(ctx, GCBindingScope{WorkspaceID: "wsp_1", Domain: "skill", BindingKey: "skill_1"}, now, "worker-1", now.Add(time.Minute), 1)
	require.NoError(t, err)
	require.Len(t, leased, 1)
	action, err := store.PrepareStagedPayloadGC(ctx, leased[0], "worker-1", now)
	require.NoError(t, err)
	require.Equal(t, GCActionDelete, action)
	updated, err := store.MarkStagedPayloadGCAttemptFailed(ctx, leased[0], "worker-1", errors.New(strings.Repeat("failure", 1000)), now)
	require.NoError(t, err)
	require.True(t, updated)
	var row StagedPayloadGC
	require.NoError(t, store.db.Where("id = ?", "ctg_1").First(&row).Error)
	require.Equal(t, GCStatusFailed, row.Status)
	require.Equal(t, MaxGCAttempts, row.Attempts)
	require.LessOrEqual(t, len(row.LastError), maxDeliveryLastError)
	require.Empty(t, row.LeaseOwner)
	leased, err = store.LeaseDueStagedPayloadGCForBinding(ctx, GCBindingScope{WorkspaceID: "wsp_1", Domain: "skill", BindingKey: "skill_1"}, now.Add(time.Hour), "worker-2", now.Add(2*time.Hour), 1)
	require.NoError(t, err)
	require.Empty(t, leased)
}