package migration

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	pathpkg "path"
	"sort"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/contentpayload"
	contentservice "github.com/tidbcloud/AgentTeam/internal/service/content"
	contentstore "github.com/tidbcloud/AgentTeam/internal/store/content"
	migrationstore "github.com/tidbcloud/AgentTeam/internal/store/content/migration"
)

const (
	MigrationMaxFiles         = 2_000
	MigrationMaxFileBytes     = int64(16 << 20)
	MigrationMaxTotalBytes    = int64(64 << 20)
	MigrationMaxManifestBytes = int64(8 << 20)
	MigrationMaxRunDuration   = 2 * time.Minute
)

type MigrationStore interface {
	AcquireExactJob(context.Context, migrationstore.AcquireParams) (migrationstore.Job, error)
	CheckpointPublished(context.Context, migrationstore.CheckpointPublishedParams) (migrationstore.Job, error)
	VerifyMapping(context.Context, migrationstore.VerifyMappingParams) (migrationstore.Job, error)
	CompleteJob(context.Context, string, string, time.Time) (migrationstore.Job, error)
	ReleaseFailure(context.Context, string, string, error, time.Time) error
	GetMapping(context.Context, string, string) (migrationstore.Mapping, error)
}

type ContentStore interface {
	FindContentByBinding(context.Context, string, string, string) (contentservice.Content, bool, error)
	CreateContent(context.Context, contentservice.CreateContentRequest) (contentservice.Content, error)
	FindVersionOperation(context.Context, string, string, string) (contentservice.VersionOperationResult, bool, error)
	CommitVersion(context.Context, contentservice.CommitVersionRequest) (contentservice.CommitVersionResult, error)
	GetVersion(context.Context, string, string, string) (contentservice.Version, error)
	ListVersionPaths(context.Context, string, string, string, int) ([]contentservice.Path, error)
}

type DetachedContentStore interface {
	FindDetachedMigrationVersionOperation(context.Context, string, string, string) (contentservice.VersionOperationResult, bool, error)
	PublishDetachedMigrationVersion(context.Context, contentservice.PublishDetachedMigrationVersionRequest) (contentservice.CommitVersionResult, error)
}

type Worker struct {
	store         MigrationStore
	content       ContentStore
	newID         contentservice.IDGenerator
	domain        string
	leaseOwner    string
	leaseDuration time.Duration
	now           func() time.Time
	semaphore     chan struct{}
}

func NewWorker(migrationStore MigrationStore, content ContentStore, newID contentservice.IDGenerator, domain, leaseOwner string, leaseDuration time.Duration) (*Worker, error) {
	domain = strings.TrimSpace(domain)
	leaseOwner = strings.TrimSpace(leaseOwner)
	if migrationStore == nil || content == nil || newID == nil || domain == "" || leaseOwner == "" || leaseDuration < MigrationMaxRunDuration || leaseDuration > migrationstore.MaxMigrationLeaseDuration {
		return nil, fmt.Errorf("exact content migration worker dependencies and bounded lease are required")
	}
	return &Worker{
		store: migrationStore, content: content, newID: newID, domain: domain, leaseOwner: leaseOwner, leaseDuration: leaseDuration,
		now: func() time.Time { return time.Now().UTC() }, semaphore: make(chan struct{}, 1),
	}, nil
}

type Request struct {
	WorkspaceID           string
	BindingKey            string
	SourceCursor          string
	RepositoryID          string
	CommitSHA             string
	ExpectedHeadVersionID string
	Actor                 string
	Limits                contentpayload.Limits
	AuthorityFence        *contentstore.LegacyAuthorityFence
}

// DetachedRequest selects the migration-only publication path. The live head
// remains an exact fence, while ParentVersionID and ExpectedGeneration identify
// the imported lineage position. Quota selects the atomic create-time profile
// and must match an already-created lineage on retry; it is either fully
// defaulted or fully explicit.
type DetachedRequest struct {
	Request
	ParentVersionID    string
	ExpectedGeneration uint64
	Quota              contentservice.QuotaProfile
}

type File struct {
	Path       string
	Data       []byte
	Executable bool
}

type Export struct{ Files []File }

type Result struct {
	JobID      string
	ContentID  string
	VersionID  string
	RootDigest string
	FileCount  int
	SizeBytes  int64
	Paths      []contentservice.Path
	Cursor     string
	Replayed   bool
}

type Exporter func(context.Context, contentpayload.Limits) (Export, error)
type Projector func(context.Context, Result) error

// RunExact processes exactly one caller-selected product/source tuple. There
// is no pending-job list operation; the durable cursor is resumed only when
// the same tuple is presented again. One worker instance admits one run at a
// time, while the store lease fences the same job across processes.
func (worker *Worker) RunExact(ctx context.Context, request Request, export Exporter, project Projector) (result Result, err error) {
	if worker == nil || export == nil || project == nil {
		return Result{}, fmt.Errorf("exact content migration worker callbacks are required")
	}
	request = normalizeRequest(request)
	if err := validateRequest(request); err != nil {
		return Result{}, err
	}
	return worker.runExact(ctx, request, nil, export, project)
}

// RunDetachedExact imports one exact source revision without changing the
// live head. It shares the bounded job cursor but has a distinct publication
// and completion contract from the existing live-head migration.
func (worker *Worker) RunDetachedExact(ctx context.Context, request DetachedRequest, export Exporter, project Projector) (Result, error) {
	if worker == nil || export == nil || project == nil {
		return Result{}, fmt.Errorf("exact content migration worker callbacks are required")
	}
	request = normalizeDetachedRequest(request)
	if err := validateDetachedRequest(request); err != nil {
		return Result{}, err
	}
	if _, ok := worker.content.(DetachedContentStore); !ok {
		return Result{}, fmt.Errorf("detached migration content store boundary is required")
	}
	return worker.runExact(ctx, request.Request, &request, export, project)
}

func (worker *Worker) runExact(ctx context.Context, request Request, detached *DetachedRequest, export Exporter, project Projector) (result Result, err error) {
	select {
	case worker.semaphore <- struct{}{}:
		defer func() { <-worker.semaphore }()
	default:
		return Result{}, fmt.Errorf("content migration worker concurrency limit reached")
	}
	runCtx, cancel := context.WithTimeout(ctx, MigrationMaxRunDuration)
	defer cancel()
	now := worker.now()
	jobID, err := worker.newID("cmj")
	if err != nil {
		return Result{}, fmt.Errorf("generate content migration job id: %w", err)
	}
	leaseToken := migrationLeaseToken(worker.leaseOwner, jobID)
	job, err := worker.store.AcquireExactJob(runCtx, migrationstore.AcquireParams{
		ID: jobID, WorkspaceID: request.WorkspaceID, Domain: worker.domain, BindingKey: request.BindingKey,
		SourceCursor: request.SourceCursor, RepositoryID: request.RepositoryID, CommitSHA: request.CommitSHA,
		ExpectedHeadVersionID: request.ExpectedHeadVersionID, LeaseOwner: leaseToken,
		LeaseUntil: now.Add(worker.leaseDuration), Now: now,
	})
	if err != nil {
		return Result{}, err
	}
	leasedJobID := job.ID
	result = migrationResultFromJob(job)
	if job.Cursor == migrationstore.MigrationCursorCompleted {
		if detached != nil {
			if _, err := worker.validateDetachedResult(runCtx, request, *detached, job); err != nil {
				return result, err
			}
		} else if err := worker.requireMigrationHead(runCtx, request, job); err != nil {
			return result, err
		}
		result.Replayed = true
		return result, nil
	}
	defer func() {
		if err == nil {
			return
		}
		releaseCtx, releaseCancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
		defer releaseCancel()
		_ = worker.store.ReleaseFailure(releaseCtx, leasedJobID, leaseToken, err, worker.now())
	}()

	replayed := job.Cursor != migrationstore.MigrationCursorPending
	job, result, replayed, err = worker.publishPending(runCtx, request, detached, job, leaseToken, replayed, export)
	if err != nil {
		return result, err
	}
	return worker.finishMapped(runCtx, request, detached, job, leaseToken, replayed, project)
}

func (worker *Worker) publishPending(ctx context.Context, request Request, detached *DetachedRequest, job migrationstore.Job, leaseToken string, replayed bool, export Exporter) (migrationstore.Job, Result, bool, error) {
	result := migrationResultFromJob(job)
	if job.Cursor != migrationstore.MigrationCursorPending {
		result.Replayed = replayed
		return job, result, replayed, nil
	}
	if err := ctx.Err(); err != nil {
		return job, result, replayed, err
	}
	exported, err := export(ctx, request.Limits)
	if err != nil {
		return job, result, replayed, err
	}
	files, digest, sizeBytes, err := migrationPayload(exported.Files, request.Limits)
	if err != nil {
		return job, result, replayed, err
	}
	if err := ctx.Err(); err != nil {
		return job, result, replayed, err
	}
	quota := contentservice.QuotaProfile{}
	requireQuotaMatch := false
	expectedGeneration := uint64(0)
	if detached != nil {
		quota = detached.Quota
		requireQuotaMatch = true
		expectedGeneration = detached.ExpectedGeneration
	}
	content, err := worker.findOrCreateMigrationContent(ctx, request, quota, requireQuotaMatch, expectedGeneration)
	if err != nil {
		return job, result, replayed, err
	}
	var published contentservice.Version
	var receiptReplay bool
	if detached != nil {
		published, receiptReplay, err = worker.publishDetachedContent(ctx, request, *detached, content, files, digest, sizeBytes)
	} else {
		published, receiptReplay, err = worker.publishContent(ctx, request, content, files, digest, sizeBytes)
	}
	if err != nil {
		return job, result, replayed, err
	}
	replayed = replayed || receiptReplay
	job, err = worker.store.CheckpointPublished(ctx, migrationstore.CheckpointPublishedParams{
		JobID: job.ID, LeaseOwner: leaseToken, ContentID: content.ID, ContentVersionID: published.ID,
		RootDigest: published.RootDigest, FileCount: published.FileCount, SizeBytes: published.SizeBytes, Now: worker.now(),
	})
	if err != nil {
		return job, result, replayed, err
	}
	result = migrationResultFromJob(job)
	result.Replayed = replayed
	return job, result, replayed, nil
}

func (worker *Worker) publishContent(ctx context.Context, request Request, content contentservice.Content, files []contentpayload.File, digest string, sizeBytes int64) (contentservice.Version, bool, error) {
	operationKey := operationKey(worker.domain, request)
	publishCtx := migrationContentContext(ctx, worker.domain, request, content.ID, content.Generation, contentservice.AccessPublish)
	receipt, found, err := worker.content.FindVersionOperation(publishCtx, request.WorkspaceID, content.ID, operationKey)
	if err != nil {
		return contentservice.Version{}, false, err
	}
	if found {
		if receipt.OperationNamespace != "content.migration" || receipt.OperationVersion != 1 ||
			receipt.ExpectedHeadVersionID != request.ExpectedHeadVersionID || receipt.ExpectedGeneration != content.Generation ||
			receipt.RootDigest != digest || receipt.FileCount != len(files) || receipt.SizeBytes != sizeBytes {
			return contentservice.Version{}, false, migrationstore.ErrMigrationConflict
		}
		published, err := worker.content.GetVersion(migrationContentContext(ctx, worker.domain, request, content.ID, content.Generation, contentservice.AccessReadMetadata), request.WorkspaceID, content.ID, receipt.VersionID)
		if err != nil {
			return contentservice.Version{}, false, err
		}
		if published.RootDigest != digest || published.FileCount != len(files) || published.SizeBytes != sizeBytes {
			return contentservice.Version{}, false, migrationstore.ErrMigrationConflict
		}
		return published, true, nil
	}
	if content.HeadVersionID != request.ExpectedHeadVersionID {
		return contentservice.Version{}, false, contentstore.ErrHeadConflict
	}
	committed, err := worker.content.CommitVersion(publishCtx, contentservice.CommitVersionRequest{
		WorkspaceID: request.WorkspaceID, ContentID: content.ID, ExpectedHeadVersionID: request.ExpectedHeadVersionID,
		ExpectedGeneration: content.Generation, Actor: request.Actor, Model: contentservice.TreeModel, Checkpoint: true,
		Files: files, TreeLimits: request.Limits, OperationKey: operationKey,
		OperationNamespace: "content.migration", OperationVersion: 1,
		AuthorityFence: request.AuthorityFence,
	})
	if err != nil {
		return contentservice.Version{}, false, err
	}
	if committed.Version.RootDigest != digest || committed.Version.FileCount != len(files) || committed.Version.SizeBytes != sizeBytes {
		return contentservice.Version{}, false, migrationstore.ErrMigrationConflict
	}
	return committed.Version, false, nil
}

func (worker *Worker) publishDetachedContent(ctx context.Context, request Request, detached DetachedRequest, content contentservice.Content, files []contentpayload.File, digest string, sizeBytes int64) (contentservice.Version, bool, error) {
	detachedContent := worker.content.(DetachedContentStore)
	if content.HeadVersionID != request.ExpectedHeadVersionID {
		return contentservice.Version{}, false, contentstore.ErrHeadConflict
	}
	if content.Generation != detached.ExpectedGeneration {
		return contentservice.Version{}, false, contentstore.ErrGenerationConflict
	}
	key := operationKey(worker.domain, request)
	publishCtx := migrationContentContext(ctx, worker.domain, request, content.ID, detached.ExpectedGeneration, contentservice.AccessPublishDetachedMigration)
	receipt, found, err := detachedContent.FindDetachedMigrationVersionOperation(publishCtx, request.WorkspaceID, content.ID, key)
	if err != nil {
		return contentservice.Version{}, false, err
	}
	if found {
		if receipt.OperationNamespace != "content.migration" || receipt.OperationVersion != 1 ||
			receipt.ExpectedHeadVersionID != request.ExpectedHeadVersionID || receipt.ExpectedGeneration != detached.ExpectedGeneration ||
			receipt.RootDigest != digest || receipt.FileCount != len(files) || receipt.SizeBytes != sizeBytes {
			return contentservice.Version{}, false, migrationstore.ErrMigrationConflict
		}
		published, err := worker.content.GetVersion(migrationContentContext(ctx, worker.domain, request, content.ID, detached.ExpectedGeneration, contentservice.AccessReadMetadata), request.WorkspaceID, content.ID, receipt.VersionID)
		if err != nil {
			return contentservice.Version{}, false, err
		}
		if !matchesDetachedVersion(published, detached, digest, len(files), sizeBytes) {
			return contentservice.Version{}, false, migrationstore.ErrMigrationConflict
		}
		return published, true, nil
	}
	committed, err := detachedContent.PublishDetachedMigrationVersion(publishCtx, contentservice.PublishDetachedMigrationVersionRequest{
		WorkspaceID: request.WorkspaceID, ContentID: content.ID,
		ExpectedHeadVersionID: request.ExpectedHeadVersionID, ParentVersionID: detached.ParentVersionID,
		ExpectedGeneration: detached.ExpectedGeneration, Actor: request.Actor, Model: contentservice.TreeModel,
		Checkpoint: true, Files: files, TreeLimits: request.Limits, OperationKey: key,
	})
	if err != nil {
		return contentservice.Version{}, false, err
	}
	if !matchesDetachedVersion(committed.Version, detached, digest, len(files), sizeBytes) {
		return contentservice.Version{}, false, migrationstore.ErrMigrationConflict
	}
	return committed.Version, false, nil
}

func matchesDetachedVersion(version contentservice.Version, request DetachedRequest, digest string, fileCount int, sizeBytes int64) bool {
	return version.ParentVersionID == request.ParentVersionID && version.Generation == request.ExpectedGeneration &&
		version.RootDigest == digest && version.FileCount == fileCount && version.SizeBytes == sizeBytes
}

func (worker *Worker) finishMapped(ctx context.Context, request Request, detached *DetachedRequest, job migrationstore.Job, leaseToken string, replayed bool, project Projector) (Result, error) {
	result := migrationResultFromJob(job)
	result.Replayed = replayed
	if job.Cursor == migrationstore.MigrationCursorPublished {
		mapID, err := worker.newID("cmm")
		if err != nil {
			return result, fmt.Errorf("generate content migration map id: %w", err)
		}
		job, err = worker.store.VerifyMapping(ctx, migrationstore.VerifyMappingParams{
			MapID: mapID, JobID: job.ID, LeaseOwner: leaseToken, Now: worker.now(),
		})
		if err != nil {
			return result, err
		}
	}
	var version contentservice.Version
	var err error
	if detached != nil {
		version, err = worker.validateDetachedResult(ctx, request, *detached, job)
	} else {
		err = worker.requireMigrationHead(ctx, request, job)
		if err == nil {
			version, err = worker.content.GetVersion(migrationContentContext(ctx, worker.domain, request, job.ContentID, 0, contentservice.AccessReadMetadata), request.WorkspaceID, job.ContentID, job.ContentVersionID)
		}
	}
	if err != nil {
		return result, err
	}
	if version.RootDigest != job.RootDigest || version.FileCount != job.FileCount || version.SizeBytes != job.SizeBytes {
		return result, migrationstore.ErrMigrationConflict
	}
	if detached != nil && (version.ParentVersionID != detached.ParentVersionID || version.Generation != detached.ExpectedGeneration) {
		return result, migrationstore.ErrMigrationConflict
	}
	paths, err := worker.content.ListVersionPaths(migrationContentContext(ctx, worker.domain, request, job.ContentID, 0, contentservice.AccessReadPayload), request.WorkspaceID, job.ContentID, job.ContentVersionID, job.FileCount)
	if err != nil {
		return result, err
	}
	if len(paths) != job.FileCount {
		return result, migrationstore.ErrMigrationConflict
	}
	result = migrationResultFromJob(job)
	result.Replayed = replayed
	result.Paths = paths
	if err := ctx.Err(); err != nil {
		return result, err
	}
	if err := project(ctx, result); err != nil {
		return result, err
	}
	if detached != nil {
		if _, err := worker.validateDetachedResult(ctx, request, *detached, job); err != nil {
			return result, err
		}
	}
	job, err = worker.store.CompleteJob(ctx, job.ID, leaseToken, worker.now())
	if err != nil {
		return result, err
	}
	result = migrationResultFromJob(job)
	result.Replayed = replayed
	result.Paths = paths
	return result, nil
}

func (worker *Worker) findOrCreateMigrationContent(ctx context.Context, request Request, quota contentservice.QuotaProfile, requireQuotaMatch bool, expectedGeneration uint64) (contentservice.Content, error) {
	readCtx := migrationContentContext(ctx, worker.domain, request, "", 0, contentservice.AccessReadMetadata)
	content, found, err := worker.content.FindContentByBinding(readCtx, request.WorkspaceID, worker.domain, request.BindingKey)
	if err != nil {
		return contentservice.Content{}, err
	}
	if found {
		if content.Model != contentservice.TreeModel {
			return contentservice.Content{}, contentstore.ErrModelConflict
		}
		if requireQuotaMatch && !contentMatchesQuota(content, quota) {
			return contentservice.Content{}, migrationstore.ErrMigrationConflict
		}
		return content, nil
	}
	if request.ExpectedHeadVersionID != "" {
		return contentservice.Content{}, contentstore.ErrHeadConflict
	}
	if requireQuotaMatch && expectedGeneration != 1 {
		return contentservice.Content{}, contentstore.ErrGenerationConflict
	}
	createCtx := migrationContentContext(ctx, worker.domain, request, "", 1, contentservice.AccessCreateLineage)
	content, err = worker.content.CreateContent(createCtx, contentservice.CreateContentRequest{
		WorkspaceID: request.WorkspaceID, Domain: worker.domain, BindingKey: request.BindingKey, Model: contentservice.TreeModel, Actor: request.Actor,
		Quota: quota,
	})
	if err == nil {
		return content, nil
	}
	if !contentservice.IsBindingConflict(err) {
		return contentservice.Content{}, err
	}
	content, found, findErr := worker.content.FindContentByBinding(readCtx, request.WorkspaceID, worker.domain, request.BindingKey)
	if findErr != nil {
		return contentservice.Content{}, findErr
	}
	if !found || content.Model != contentservice.TreeModel || (requireQuotaMatch && !contentMatchesQuota(content, quota)) {
		return contentservice.Content{}, contentstore.ErrBindingConflict
	}
	return content, nil
}

func contentMatchesQuota(content contentservice.Content, quota contentservice.QuotaProfile) bool {
	if quota == (contentservice.QuotaProfile{}) {
		return content.MaxRetainedVersions == contentstore.DefaultTreeVersions &&
			content.MaxStoredPayloadBytes == contentstore.DefaultTreeBytes && content.MaxPins == contentstore.DefaultTreePins
	}
	return content.MaxRetainedVersions == quota.MaxRetainedVersions &&
		content.MaxStoredPayloadBytes == quota.MaxStoredPayloadBytes && content.MaxPins == quota.MaxPins
}

func (worker *Worker) validateDetachedResult(ctx context.Context, request Request, detached DetachedRequest, job migrationstore.Job) (contentservice.Version, error) {
	readCtx := migrationContentContext(ctx, worker.domain, request, "", 0, contentservice.AccessReadMetadata)
	content, found, err := worker.content.FindContentByBinding(readCtx, request.WorkspaceID, worker.domain, request.BindingKey)
	if err != nil {
		return contentservice.Version{}, err
	}
	if !found || content.ID != job.ContentID || content.HeadVersionID != request.ExpectedHeadVersionID {
		return contentservice.Version{}, contentstore.ErrHeadConflict
	}
	if content.Generation != detached.ExpectedGeneration {
		return contentservice.Version{}, contentstore.ErrGenerationConflict
	}
	version, err := worker.content.GetVersion(migrationContentContext(ctx, worker.domain, request, job.ContentID, detached.ExpectedGeneration, contentservice.AccessReadMetadata), request.WorkspaceID, job.ContentID, job.ContentVersionID)
	if err != nil {
		return contentservice.Version{}, err
	}
	if version.ParentVersionID != detached.ParentVersionID || version.Generation != detached.ExpectedGeneration ||
		version.RootDigest != job.RootDigest || version.FileCount != job.FileCount || version.SizeBytes != job.SizeBytes {
		return contentservice.Version{}, migrationstore.ErrMigrationConflict
	}
	mapping, err := worker.store.GetMapping(ctx, request.RepositoryID, request.CommitSHA)
	if err != nil {
		return contentservice.Version{}, err
	}
	if mapping.WorkspaceID != request.WorkspaceID || mapping.ContentID != job.ContentID || mapping.ContentVersionID != job.ContentVersionID ||
		mapping.Model != contentservice.TreeModel || mapping.RootDigest != job.RootDigest || mapping.FileCount != job.FileCount ||
		mapping.SizeBytes != job.SizeBytes || mapping.VerifiedAt == nil {
		return contentservice.Version{}, migrationstore.ErrMigrationConflict
	}
	return version, nil
}

func (worker *Worker) requireMigrationHead(ctx context.Context, request Request, job migrationstore.Job) error {
	readCtx := migrationContentContext(ctx, worker.domain, request, "", 0, contentservice.AccessReadMetadata)
	content, found, err := worker.content.FindContentByBinding(readCtx, request.WorkspaceID, worker.domain, request.BindingKey)
	if err != nil {
		return err
	}
	if !found || content.ID != job.ContentID || content.HeadVersionID != job.ContentVersionID {
		return contentstore.ErrHeadConflict
	}
	return nil
}

func normalizeRequest(request Request) Request {
	request.WorkspaceID = strings.TrimSpace(request.WorkspaceID)
	request.BindingKey = strings.TrimSpace(request.BindingKey)
	request.SourceCursor = strings.TrimSpace(request.SourceCursor)
	request.RepositoryID = strings.TrimSpace(request.RepositoryID)
	request.CommitSHA = strings.ToLower(strings.TrimSpace(request.CommitSHA))
	request.ExpectedHeadVersionID = strings.TrimSpace(request.ExpectedHeadVersionID)
	request.Actor = strings.TrimSpace(request.Actor)
	return request
}

func normalizeDetachedRequest(request DetachedRequest) DetachedRequest {
	request.Request = normalizeRequest(request.Request)
	request.ParentVersionID = strings.TrimSpace(request.ParentVersionID)
	return request
}

func validateDetachedRequest(request DetachedRequest) error {
	if err := validateRequest(request.Request); err != nil {
		return err
	}
	quota := request.Quota
	allDefault := quota == (contentservice.QuotaProfile{})
	allExplicit := quota.MaxRetainedVersions > 0 && quota.MaxStoredPayloadBytes > 0 && quota.MaxPins > 0
	if request.ExpectedGeneration == 0 || (!allDefault && !allExplicit) {
		return fmt.Errorf("detached content migration generation and quota are invalid")
	}
	return nil
}

func validateRequest(request Request) error {
	if request.WorkspaceID == "" || request.BindingKey == "" || request.SourceCursor == "" || request.RepositoryID == "" || request.Actor == "" || len(request.CommitSHA) != 40 {
		return fmt.Errorf("exact content migration target is invalid")
	}
	if _, err := hex.DecodeString(request.CommitSHA); err != nil {
		return fmt.Errorf("exact content migration commit is invalid")
	}
	limits, err := contentpayload.NormalizeLimits(request.Limits)
	if err != nil {
		return err
	}
	if limits != request.Limits || limits.MaxFiles > MigrationMaxFiles || limits.MaxFileBytes > MigrationMaxFileBytes || limits.MaxTotalBytes > MigrationMaxTotalBytes || limits.MaxManifestBytes > MigrationMaxManifestBytes {
		return fmt.Errorf("exact content migration limits exceed worker bounds")
	}
	return nil
}

func migrationPayload(exported []File, limits contentpayload.Limits) ([]contentpayload.File, string, int64, error) {
	if len(exported) == 0 || len(exported) > limits.MaxFiles || len(exported) > MigrationMaxFiles {
		return nil, "", 0, fmt.Errorf("exact content migration file count exceeds bounds")
	}
	exported = append([]File(nil), exported...)
	sort.Slice(exported, func(i, j int) bool { return strings.TrimSpace(exported[i].Path) < strings.TrimSpace(exported[j].Path) })
	entries := make([]contentpayload.ManifestEntry, 0, len(exported))
	files := make([]contentpayload.File, 0, len(exported))
	seen := make(map[string]struct{}, len(exported))
	var total int64
	for _, file := range exported {
		path := strings.TrimSpace(file.Path)
		size := int64(len(file.Data))
		cleaned := pathpkg.Clean(path)
		if path == "" || cleaned != path || pathpkg.IsAbs(path) || path == "." || strings.HasPrefix(path, "../") || strings.Contains(path, "\\") || strings.ContainsRune(path, '\x00') || size > limits.MaxFileBytes || total > limits.MaxTotalBytes-size {
			return nil, "", 0, fmt.Errorf("exact content migration file exceeds path or byte bounds")
		}
		if _, duplicate := seen[path]; duplicate {
			return nil, "", 0, fmt.Errorf("exact content migration contains duplicate path %q", path)
		}
		seen[path] = struct{}{}
		sum := sha256.Sum256(file.Data)
		entries = append(entries, contentpayload.ManifestEntry{Path: path, Digest: contentpayload.Digest(hex.EncodeToString(sum[:])), SizeBytes: size, Executable: file.Executable})
		files = append(files, contentpayload.File{Path: path, Size: size, Executable: file.Executable, Content: bytes.NewReader(file.Data)})
		total += size
	}
	encoded, err := json.Marshal(entries)
	if err != nil {
		return nil, "", 0, err
	}
	sum := sha256.Sum256(encoded)
	manifest, err := json.Marshal(contentpayload.Manifest{
		WorkspaceID: strings.Repeat("w", 64), ContentID: strings.Repeat("c", 64), VersionID: strings.Repeat("v", 64),
		RootDigest: contentpayload.Digest(hex.EncodeToString(sum[:])), SizeBytes: total, Files: entries,
	})
	if err != nil {
		return nil, "", 0, err
	}
	if int64(len(manifest)) > limits.MaxManifestBytes {
		return nil, "", 0, fmt.Errorf("exact content migration manifest exceeds bounds")
	}
	return files, hex.EncodeToString(sum[:]), total, nil
}

func operationKey(domain string, request Request) string {
	sum := sha256.Sum256([]byte(strings.Join([]string{request.WorkspaceID, domain, request.BindingKey, request.SourceCursor, request.RepositoryID, request.CommitSHA}, "\x00")))
	return "content.migration:" + hex.EncodeToString(sum[:])
}

func migrationLeaseToken(workerID, attemptID string) string {
	sum := sha256.Sum256([]byte(workerID + "\x00" + attemptID))
	return "cml_" + hex.EncodeToString(sum[:])
}

func migrationContentContext(ctx context.Context, domain string, request Request, contentID string, generation uint64, action contentservice.AccessAction) context.Context {
	return contentservice.WithAccessPrincipal(ctx, contentservice.AccessPrincipal{
		Subject: request.Actor, WorkspaceID: request.WorkspaceID, Domain: domain, BindingKey: request.BindingKey,
		ContentID: contentID, Generation: generation, Actions: []contentservice.AccessAction{action},
	})
}

func migrationResultFromJob(job migrationstore.Job) Result {
	return Result{
		JobID: job.ID, ContentID: job.ContentID, VersionID: job.ContentVersionID, RootDigest: job.RootDigest,
		FileCount: job.FileCount, SizeBytes: job.SizeBytes, Cursor: job.Cursor,
	}
}