package content

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

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/AgentTeam/internal/contentpayload"
	store "github.com/tidbcloud/AgentTeam/internal/store/content"
)

const testOperationNamespace = "kb.source-mirror"

func TestCommitVersionOperationReplaysExactWinnerAfterHeadAndQuotaAdvance(t *testing.T) {
	service, database := newServiceTest(t)
	ctx := context.Background()
	content, err := service.CreateContent(ctx, CreateContentRequest{WorkspaceID: "wsp_1", Domain: "kb_source", BindingKey: "src_1", Model: store.ModelTree, Actor: "worker_1"})
	require.NoError(t, err)
	require.NoError(t, database.Model(&store.VersionedContent{}).Where("id = ?", content.ID).Updates(map[string]any{"max_retained_versions": 2, "max_stored_payload_bytes": 10}).Error)

	request := operationCommitRequest(content.ID, "hello", "")
	first, err := service.CommitVersion(ctx, request)
	require.NoError(t, err)
	require.Len(t, first.RequestDigest, 64)

	second, err := service.CommitVersion(ctx, CommitVersionRequest{
		WorkspaceID: "wsp_1", ContentID: content.ID, ExpectedHeadVersionID: first.Version.ID,
		ExpectedGeneration: 1, Actor: "worker_1", Model: store.ModelTree,
		Files: []contentpayload.File{{Path: "README.md", Size: 5, Content: bytes.NewBufferString("world")}},
	})
	require.NoError(t, err)
	require.NotEqual(t, first.Version.ID, second.Version.ID)

	replayed, err := service.CommitVersion(ctx, operationCommitRequest(content.ID, "hello", first.RequestDigest))
	require.NoError(t, err)
	require.Equal(t, first.Version.ID, replayed.Version.ID)
	require.Equal(t, first.Version.ID, replayed.Content.HeadVersionID, "replay returns the historical operation result, not mutable current head")
	require.Equal(t, uint64(1), replayed.Content.RetainedVersionCount)
	require.Equal(t, int64(5), replayed.Content.StoredPayloadBytes)
	current, err := service.GetContent(ctx, "wsp_1", content.ID)
	require.NoError(t, err)
	require.Equal(t, second.Version.ID, current.HeadVersionID)
	require.Equal(t, uint64(2), current.RetainedVersionCount)
	require.Equal(t, int64(10), current.StoredPayloadBytes)
	require.NoError(t, database.Model(&store.VersionedContent{}).Where("id = ?", content.ID).Update("max_stored_payload_bytes", 4).Error)
	afterCeiling, err := service.GetContent(ctx, "wsp_1", content.ID)
	require.NoError(t, err)

	replayed, err = service.CommitVersion(ctx, operationCommitRequest(content.ID, "hello", first.RequestDigest))
	require.NoError(t, err, "completed replay ignores a subsequently lowered dynamic lineage ceiling")
	require.Equal(t, first.Version.ID, replayed.Version.ID)

	newOperation := operationCommitRequest(content.ID, "x", "")
	newOperation.OperationKey = testOperationNamespace + ":job-2"
	newOperation.ExpectedHeadVersionID = second.Version.ID
	_, err = service.CommitVersion(ctx, newOperation)
	require.ErrorIs(t, err, store.ErrQuotaExceeded)
	_, err = service.CommitVersion(ctx, CommitVersionRequest{
		WorkspaceID: "wsp_1", ContentID: content.ID, ExpectedHeadVersionID: second.Version.ID,
		ExpectedGeneration: 1, Actor: "worker_1", Model: store.ModelTree,
		Files: []contentpayload.File{{Path: "README.md", Size: 1, Content: bytes.NewBufferString("x")}},
	})
	require.ErrorIs(t, err, store.ErrQuotaExceeded)
	afterRejectedWrites, err := service.GetContent(ctx, "wsp_1", content.ID)
	require.NoError(t, err)
	require.Equal(t, afterCeiling, afterRejectedWrites, "replay and quota failures do not mutate current lineage state")

	var versions, outbox, receipts int64
	require.NoError(t, database.Model(&store.ContentVersion{}).Count(&versions).Error)
	require.NoError(t, database.Model(&store.Outbox{}).Count(&outbox).Error)
	require.NoError(t, database.Model(&store.VersionOperation{}).Count(&receipts).Error)
	require.Equal(t, int64(2), versions)
	require.Equal(t, int64(2), outbox)
	require.Equal(t, int64(1), receipts)
	var pendingGC int64
	require.NoError(t, database.Model(&store.StagedPayloadGC{}).Where("status = ?", store.GCStatusPending).Count(&pendingGC).Error)
	require.Equal(t, int64(4), pendingGC, "every replay or rejected staged payload remains independently reclaimable")
}

func TestCommitVersionOperationNormalizesTreeBoundsAndRejectsPolicyConflict(t *testing.T) {
	service, _ := newServiceTest(t)
	ctx := context.Background()
	content, err := service.CreateContent(ctx, CreateContentRequest{WorkspaceID: "wsp_1", Domain: "kb_source", BindingKey: "src_1", Model: store.ModelTree, Actor: "worker_1"})
	require.NoError(t, err)

	request := operationCommitRequest(content.ID, "hello", "")
	request.Files = append(request.Files, contentpayload.File{Path: "SECOND.md", Size: 1, Content: bytes.NewBufferString("x")})
	first, err := service.CommitVersion(ctx, request)
	require.NoError(t, err)

	defaults := operationCommitRequest(content.ID, "hello", first.RequestDigest)
	defaults.Files = append(defaults.Files, contentpayload.File{Path: "SECOND.md", Size: 1, Content: bytes.NewBufferString("x")})
	defaults.TreeLimits = contentpayload.DefaultLimits
	replayed, err := service.CommitVersion(ctx, defaults)
	require.NoError(t, err, "explicit defaults and zero/defaulted bounds are the same policy")
	require.Equal(t, first.Version.ID, replayed.Version.ID)

	tooStrict := operationCommitRequest(content.ID, "hello", "")
	tooStrict.Files = append(tooStrict.Files, contentpayload.File{Path: "SECOND.md", Size: 1, Content: bytes.NewBufferString("x")})
	tooStrict.TreeLimits.MaxFiles = 1
	_, err = service.CommitVersion(ctx, tooStrict)
	require.Error(t, err, "a receipt cannot bypass current staging bounds")
	require.NotErrorIs(t, err, store.ErrOperationConflict)

	different := operationCommitRequest(content.ID, "hello", "")
	different.Files = append(different.Files, contentpayload.File{Path: "SECOND.md", Size: 1, Content: bytes.NewBufferString("x")})
	different.TreeLimits = contentpayload.DefaultLimits
	different.TreeLimits.MaxFiles--
	_, err = service.CommitVersion(ctx, different)
	require.ErrorIs(t, err, store.ErrOperationConflict)
}

func TestCommitVersionDocumentOperationUsesOwnerBoundsAndZeroReceipt(t *testing.T) {
	service, database := newServiceTest(t)
	ctx := context.Background()
	content, err := service.CreateContent(ctx, CreateContentRequest{WorkspaceID: "wsp_1", Domain: "memory", BindingKey: "mem_1", Model: store.ModelDocument, Actor: "worker_1"})
	require.NoError(t, err)
	request := CommitVersionRequest{
		WorkspaceID: "wsp_1", ContentID: content.ID, ExpectedGeneration: 1,
		Actor: "worker_1", Model: store.ModelDocument, Checkpoint: true,
		Files:        []contentpayload.File{{Path: "MEMORY.md", Size: 5, Content: bytes.NewBufferString("hello")}},
		OperationKey: testOperationNamespace + ":document-1", OperationNamespace: testOperationNamespace,
		OperationVersion: 1,
	}
	first, err := service.CommitVersion(ctx, request)
	require.NoError(t, err)
	require.NotEmpty(t, first.RequestDigest)
	request.ExpectedRequestDigest = first.RequestDigest
	request.Files = []contentpayload.File{{Path: "MEMORY.md", Size: 5, Content: bytes.NewBufferString("hello")}}
	replayed, err := service.CommitVersion(ctx, request)
	require.NoError(t, err)
	require.Equal(t, first.Version.ID, replayed.Version.ID)

	var receipt store.VersionOperation
	require.NoError(t, database.Where("workspace_id = ? AND content_id = ? AND operation_key = ?", "wsp_1", content.ID, request.OperationKey).First(&receipt).Error)
	require.Equal(t, documentBoundsPolicyNamespace, receipt.BoundsPolicyNamespace)
	require.Zero(t, receipt.MaxFiles)
	require.Zero(t, receipt.MaxFileBytes)
	require.Zero(t, receipt.MaxTotalBytes)
	require.Zero(t, receipt.MaxManifestBytes)
}

func TestCommitVersionActorLengthUsesSharedPersistenceBound(t *testing.T) {
	service, database := newServiceTest(t)
	ctx := context.Background()
	content, err := service.CreateContent(ctx, CreateContentRequest{WorkspaceID: "wsp_1", Domain: "kb_source", BindingKey: "src_1", Model: store.ModelTree, Actor: "worker_1"})
	require.NoError(t, err)

	allowed := operationCommitRequest(content.ID, "hello", "")
	allowed.Actor = string(bytes.Repeat([]byte{'a'}, 64))
	allowed.OperationKey = testOperationNamespace + ":actor-64"
	_, err = service.CommitVersion(ctx, allowed)
	require.NoError(t, err)

	other, err := service.CreateContent(ctx, CreateContentRequest{WorkspaceID: "wsp_1", Domain: "kb_source", BindingKey: "src_2", Model: store.ModelTree, Actor: "worker_1"})
	require.NoError(t, err)
	tooLong := operationCommitRequest(other.ID, "hello", "")
	tooLong.Actor = string(bytes.Repeat([]byte{'a'}, 65))
	tooLong.OperationKey = testOperationNamespace + ":actor-65"
	_, err = service.CommitVersion(ctx, tooLong)
	require.ErrorContains(t, err, "actor is invalid")

	for model, name := range map[any]string{
		&store.ContentVersion{}: "version", &store.Outbox{}: "outbox",
		&store.VersionOperation{}: "receipt", &store.StagedPayloadGC{}: "gc",
	} {
		var count int64
		require.NoError(t, database.Model(model).Where("content_id = ?", other.ID).Count(&count).Error)
		require.Zero(t, count, name)
	}
}

func TestCommitVersionRejectsTreeBoundsOutsideOwnerPolicyBeforeStaging(t *testing.T) {
	for _, testCase := range []struct {
		name   string
		limits contentpayload.Limits
	}{
		{name: "above hard cap", limits: contentpayload.Limits{MaxFileBytes: contentpayload.DefaultLimits.MaxFileBytes + 1}},
		{name: "negative", limits: contentpayload.Limits{MaxFiles: -1}},
	} {
		t.Run(testCase.name, func(t *testing.T) {
			service, database := newServiceTest(t)
			ctx := context.Background()
			content, err := service.CreateContent(ctx, CreateContentRequest{WorkspaceID: "wsp_1", Domain: "kb_source", BindingKey: "src_1", Model: store.ModelTree, Actor: "worker_1"})
			require.NoError(t, err)
			request := operationCommitRequest(content.ID, "x", "")
			request.TreeLimits = testCase.limits
			_, err = service.CommitVersion(ctx, request)
			require.ErrorContains(t, err, "outside owner bounds")

			for model, name := range map[any]string{
				&store.ContentVersion{}: "version", &store.Outbox{}: "outbox",
				&store.VersionOperation{}: "receipt", &store.StagedPayloadGC{}: "gc",
			} {
				var count int64
				require.NoError(t, database.Model(model).Where("content_id = ?", content.ID).Count(&count).Error)
				require.Zero(t, count, name)
			}
		})
	}
}

func TestCommitVersionOperationLostAcknowledgementRecoversExactReceipt(t *testing.T) {
	service, database := newServiceTest(t)
	ctx := context.Background()
	content, err := service.CreateContent(ctx, CreateContentRequest{WorkspaceID: "wsp_1", Domain: "kb_source", BindingKey: "src_1", Model: store.ModelTree, Actor: "worker_1"})
	require.NoError(t, err)
	service.store = &lostPublishAcknowledgementStore{Store: service.store}
	request := operationCommitRequest(content.ID, "hello", "")
	_, err = service.CommitVersion(ctx, request)
	require.ErrorContains(t, err, "lost publish acknowledgement")

	var receipt store.VersionOperation
	require.NoError(t, database.Where("workspace_id = ? AND content_id = ? AND operation_key = ?", "wsp_1", content.ID, request.OperationKey).First(&receipt).Error)
	request = operationCommitRequest(content.ID, "hello", receipt.RequestDigest)
	replayed, err := service.CommitVersion(ctx, request)
	require.NoError(t, err)
	require.Equal(t, receipt.VersionID, replayed.Version.ID)
}

func TestCommitVersionOperationRejectsEnvelopeConflictAndInvalidFirstFence(t *testing.T) {
	service, database := newServiceTest(t)
	ctx := context.Background()
	content, err := service.CreateContent(ctx, CreateContentRequest{WorkspaceID: "wsp_1", Domain: "kb_source", BindingKey: "src_1", Model: store.ModelTree, Actor: "worker_1"})
	require.NoError(t, err)
	request := operationCommitRequest(content.ID, "hello", "")
	first, err := service.CommitVersion(ctx, request)
	require.NoError(t, err)

	actorConflict := operationCommitRequest(content.ID, "hello", "")
	actorConflict.Actor = "worker_2"
	_, err = service.CommitVersion(ctx, actorConflict)
	require.ErrorIs(t, err, store.ErrOperationConflict)
	fenceConflict := operationCommitRequest(content.ID, "hello", "")
	fenceConflict.ExpectedHeadVersionID = first.Version.ID
	_, err = service.CommitVersion(ctx, fenceConflict)
	require.ErrorIs(t, err, store.ErrOperationConflict)

	invalid := operationCommitRequest(content.ID, "other", "")
	invalid.OperationKey = testOperationNamespace + ":job-invalid"
	invalid.ExpectedHeadVersionID = "ctv_missing"
	_, err = service.CommitVersion(ctx, invalid)
	require.ErrorIs(t, err, store.ErrHeadConflict)
	var invalidReceipts int64
	require.NoError(t, database.Model(&store.VersionOperation{}).Where("operation_key = ?", invalid.OperationKey).Count(&invalidReceipts).Error)
	require.Zero(t, invalidReceipts)
	var pendingGC int64
	require.NoError(t, database.Model(&store.StagedPayloadGC{}).Where("status = ?", store.GCStatusPending).Count(&pendingGC).Error)
	require.GreaterOrEqual(t, pendingGC, int64(3))
}

func TestCommitVersionOperationRepeatedCallsReturnOneVersion(t *testing.T) {
	service, database := newServiceTest(t)
	ctx := context.Background()
	content, err := service.CreateContent(ctx, CreateContentRequest{WorkspaceID: "wsp_1", Domain: "kb_source", BindingKey: "src_1", Model: store.ModelTree, Actor: "worker_1"})
	require.NoError(t, err)

	first, err := service.CommitVersion(ctx, operationCommitRequest(content.ID, "hello", ""))
	require.NoError(t, err)
	second, err := service.CommitVersion(ctx, operationCommitRequest(content.ID, "hello", first.RequestDigest))
	require.NoError(t, err)
	require.Equal(t, first.Version.ID, second.Version.ID)
	var versions, receipts, outbox int64
	require.NoError(t, database.Model(&store.ContentVersion{}).Count(&versions).Error)
	require.NoError(t, database.Model(&store.VersionOperation{}).Count(&receipts).Error)
	require.NoError(t, database.Model(&store.Outbox{}).Count(&outbox).Error)
	require.Equal(t, int64(1), versions)
	require.Equal(t, int64(1), receipts)
	require.Equal(t, int64(1), outbox)
}

func TestPublishDetachedMigrationVersionUsesIndependentParentAndHeadFence(t *testing.T) {
	service, database := newServiceTest(t)
	ctx := context.Background()
	content, err := service.CreateContent(ctx, CreateContentRequest{WorkspaceID: "wsp_1", Domain: "memory", BindingKey: "mem_1", Model: store.ModelTree, Actor: "worker_1"})
	require.NoError(t, err)
	parent, err := service.CommitVersion(ctx, CommitVersionRequest{
		WorkspaceID: "wsp_1", ContentID: content.ID, ExpectedGeneration: 1, Actor: "worker_1", Model: store.ModelTree,
		Files: []contentpayload.File{{Path: "MEMORY.md", Size: 6, Content: bytes.NewBufferString("parent")}},
	})
	require.NoError(t, err)
	live, err := service.CommitVersion(ctx, CommitVersionRequest{
		WorkspaceID: "wsp_1", ContentID: content.ID, ExpectedHeadVersionID: parent.Version.ID, ExpectedGeneration: 1,
		Actor: "worker_1", Model: store.ModelTree,
		Files: []contentpayload.File{{Path: "MEMORY.md", Size: 4, Content: bytes.NewBufferString("live")}},
	})
	require.NoError(t, err)

	request := PublishDetachedMigrationVersionRequest{
		WorkspaceID: "wsp_1", ContentID: content.ID, ExpectedHeadVersionID: live.Version.ID,
		ParentVersionID: parent.Version.ID, ExpectedGeneration: 1, Actor: "worker_1", Model: store.ModelTree, Checkpoint: true,
		Files:        []contentpayload.File{{Path: "archive.md", Size: 8, Content: bytes.NewBufferString("detached")}},
		OperationKey: "content.migration:memory-detached",
	}
	detached, err := service.publishDetachedMigrationVersion(ctx, request)
	require.NoError(t, err)
	require.Equal(t, live.Version.ID, detached.Content.HeadVersionID)
	require.Equal(t, parent.Version.ID, detached.Version.ParentVersionID)
	require.Equal(t, uint64(1), detached.Version.Generation)
	require.Len(t, detached.RequestDigest, 64)

	request.ExpectedRequestDigest = detached.RequestDigest
	request.Files = []contentpayload.File{{Path: "archive.md", Size: 8, Content: bytes.NewBufferString("detached")}}
	replayed, err := service.publishDetachedMigrationVersion(ctx, request)
	require.NoError(t, err)
	require.Equal(t, detached.Version.ID, replayed.Version.ID)
	require.Equal(t, live.Version.ID, replayed.Content.HeadVersionID)
	_, err = service.CommitVersion(ctx, CommitVersionRequest{
		WorkspaceID: "wsp_1", ContentID: content.ID, ExpectedHeadVersionID: live.Version.ID, ExpectedGeneration: 1,
		Actor: "worker_1", Model: store.ModelTree, Checkpoint: true,
		Files:        []contentpayload.File{{Path: "archive.md", Size: 8, Content: bytes.NewBufferString("detached")}},
		OperationKey: request.OperationKey, OperationNamespace: "content.migration", OperationVersion: 1,
	})
	require.ErrorIs(t, err, store.ErrOperationConflict, "ordinary publication cannot reinterpret a detached receipt")
	var versions, receipts, outbox int64
	require.NoError(t, database.Model(&store.ContentVersion{}).Where("content_id = ?", content.ID).Count(&versions).Error)
	require.NoError(t, database.Model(&store.VersionOperation{}).Where("content_id = ?", content.ID).Count(&receipts).Error)
	require.NoError(t, database.Model(&store.Outbox{}).Where("content_id = ?", content.ID).Count(&outbox).Error)
	require.Equal(t, int64(3), versions)
	require.Equal(t, int64(1), receipts)
	require.Equal(t, int64(3), outbox)

	request.ExpectedHeadVersionID = parent.Version.ID
	request.ExpectedRequestDigest = ""
	request.Files = []contentpayload.File{{Path: "archive.md", Size: 8, Content: bytes.NewBufferString("detached")}}
	_, err = service.publishDetachedMigrationVersion(ctx, request)
	require.ErrorIs(t, err, store.ErrHeadConflict)
}

func TestAuthorizedCommitVersionOperationReplayReauthorizesBeforeStaging(t *testing.T) {
	core, database := newServiceTest(t)
	policy := &mutableContentPolicy{}
	authorized, err := NewAuthorizedService(core, policy)
	require.NoError(t, err)
	ctx := WithAccessPrincipal(context.Background(), AccessPrincipal{
		Subject: "worker_1", WorkspaceID: "wsp_1", Domain: "kb_source", BindingKey: "src_1",
		Actions: []AccessAction{AccessCreateLineage, AccessReadMetadata, AccessPublish},
	})
	content, err := authorized.CreateContent(ctx, CreateContentRequest{WorkspaceID: "wsp_1", Domain: "kb_source", BindingKey: "src_1", Model: store.ModelTree, Actor: "worker_1"})
	require.NoError(t, err)
	first, err := authorized.CommitVersion(ctx, operationCommitRequest(content.ID, "hello", ""))
	require.NoError(t, err)
	var gcBefore int64
	require.NoError(t, database.Model(&store.StagedPayloadGC{}).Count(&gcBefore).Error)
	policy.denied = true
	_, err = authorized.CommitVersion(ctx, operationCommitRequest(content.ID, "hello", first.RequestDigest))
	require.ErrorIs(t, err, ErrAccessDenied)
	var gcAfter int64
	require.NoError(t, database.Model(&store.StagedPayloadGC{}).Count(&gcAfter).Error)
	require.Equal(t, gcBefore, gcAfter, "denied replay must not stage a payload")
}

func operationCommitRequest(contentID, body, expectedDigest string) CommitVersionRequest {
	return CommitVersionRequest{
		WorkspaceID: "wsp_1", ContentID: contentID, ExpectedGeneration: 1,
		Actor: "worker_1", Model: store.ModelTree, Checkpoint: true,
		Files:        []contentpayload.File{{Path: "README.md", Size: int64(len(body)), Content: bytes.NewBufferString(body)}},
		OperationKey: testOperationNamespace + ":job-1", OperationNamespace: testOperationNamespace,
		OperationVersion: 1, ExpectedRequestDigest: expectedDigest,
	}
}

type lostPublishAcknowledgementStore struct {
	Store
	once sync.Once
}

type mutableContentPolicy struct{ denied bool }

func (policy *mutableContentPolicy) AuthorizeContentAccess(context.Context, AccessRequest) error {
	if policy.denied {
		return ErrAccessDenied
	}
	return nil
}

func (wrapper *lostPublishAcknowledgementStore) PublishVersion(ctx context.Context, params store.PublishVersionParams) (store.PublishVersionResult, error) {
	result, err := wrapper.Store.PublishVersion(ctx, params)
	if err != nil {
		return result, err
	}
	lost := false
	wrapper.once.Do(func() { lost = true })
	if lost {
		return store.PublishVersionResult{}, errors.New("lost publish acknowledgement")
	}
	return result, nil
}