package activity

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"sort"
	"strings"
	"time"

	"golang.org/x/sync/singleflight"
	"gorm.io/gorm"

	activitystore "github.com/tidbcloud/AgentTeam/internal/store/resource/activity"
	reposstore "github.com/tidbcloud/AgentTeam/internal/store/resource/repos"
)

// DefaultResourceActivityRefreshAfter is the read-through cache TTL for
// managed resource activity data-plane projections.
const DefaultResourceActivityRefreshAfter = 15 * time.Minute

var (
	ErrResourceActivityProjectorRateLimited = errors.New("resource activity projector rate limited")
	ErrResourceActivityProjectorNotFound    = errors.New("resource activity projector repository not found")
)

// ResourceActivityRepositoryLister is the narrow repository-store port used to discover bounded projector inputs.
type ResourceActivityRepositoryLister interface {
	ListActivityProjectorRepositories(ctx context.Context, opts reposstore.ListActivityProjectorRepositoriesOptions) ([]reposstore.ResourceRepository, error)
}

// ResourceActivityProjectorConfig bounds repository discovery, per-kind reads, backfill, and the service-owned projection rules.
type ResourceActivityProjectorConfig struct {
	RepoLimit       int
	PerKindLimit    int
	BackfillWindow  time.Duration
	ProjectionRules []ResourceActivityProjectionRule
}

// ResourceActivityProjector coordinates bounded AGS data-plane reads into the resource activity persistence owner; it is not a generic AgentTeam activity worker.
type ResourceActivityProjector struct {
	Repositories ResourceActivityRepositoryLister
	Activity     *activitystore.Store
	Source       ResourceActivityDataPlaneSource
	Config       ResourceActivityProjectorConfig
	Now          func() time.Time

	refreshFlights singleflight.Group
}

type resourceActivityProjectorTargetSpec struct {
	AssetType string
	Kinds     []string
}

// ResourceActivityProjectionRule declares which repository-backed product
// resources should project which AGS activity streams into MeshX activitystore.
type ResourceActivityProjectionRule struct {
	ResourceType string
	RepoRoles    []string
	AssetType    string
	EventKinds   []string
}

var defaultResourceActivityProjectionRules = []ResourceActivityProjectionRule{
	{
		ResourceType: reposstore.ResourceTypeMemory,
		RepoRoles:    []string{reposstore.RepoRolePrimary},
		AssetType:    activitystore.AssetTypeMemory,
		EventKinds:   []string{activitystore.ProjectorKindCommits, activitystore.ProjectorKindReleases},
	},
	{
		ResourceType: reposstore.ResourceTypeSkillRepository,
		RepoRoles:    []string{reposstore.RepoRolePrimary},
		AssetType:    activitystore.AssetTypeSkill,
		EventKinds:   []string{activitystore.ProjectorKindCommits, activitystore.ProjectorKindReleases},
	},
	{
		ResourceType: reposstore.ResourceTypeKBSource,
		RepoRoles:    []string{reposstore.RepoRolePrimary, reposstore.RepoRoleMirror},
		AssetType:    activitystore.AssetTypeKB,
		EventKinds:   []string{activitystore.ProjectorKindCommits, activitystore.ProjectorKindReleases},
	},
	{
		ResourceType: reposstore.ResourceTypeKBSource,
		RepoRoles:    []string{reposstore.RepoRoleInsights},
		AssetType:    activitystore.AssetTypeKB,
		EventKinds:   []string{activitystore.ProjectorKindWiki},
	},
}

// ResourceActivityDataPlaneRepository identifies one AGS repository input to
// the projector, not a resource repository store row or API repository DTO.
type ResourceActivityDataPlaneRepository struct {
	WorkspaceID  string
	ResourceType string
	ResourceID   string
	AssetType    string
	AssetID      string
	RepoRole     string
	RepoOwner    string
	RepoName     string
	RepoFullName string
	DefaultRef   string
}

// ResourceActivityDataPlaneListOptions describes one bounded upstream activity page read.
type ResourceActivityDataPlaneListOptions struct {
	Since         *time.Time
	BackfillSince time.Time
	Limit         int
	StateJSON     string
	Now           time.Time
}

// ResourceActivityDataPlaneBatch is one upstream projector page plus its continuation cursor.
type ResourceActivityDataPlaneBatch struct {
	Events    []ResourceActivityDataPlaneEvent
	StateJSON string
}

// ResourceActivityDataPlaneEvent is the normalized upstream event accepted by
// the projector, not a persistence model or public event DTO.
type ResourceActivityDataPlaneEvent struct {
	SourceEventID string
	EventType     string
	ActorType     string
	ActorID       string
	ActorName     string
	SummaryKey    string
	Details       any
	Display       ResourceActivityDisplay
	TargetKind    string
	TargetID      string
	TargetURL     string
	OccurredAt    time.Time
}

// ResourceActivityDataPlaneSource is the narrow AGS activity read port consumed by the projector.
type ResourceActivityDataPlaneSource interface {
	ListActivityEvents(ctx context.Context, repo ResourceActivityDataPlaneRepository, eventKind string, opts ResourceActivityDataPlaneListOptions) (ResourceActivityDataPlaneBatch, error)
}

// ResourceActivityProjectorRunOptions scopes one projector run to bounded repositories and resource types.
type ResourceActivityProjectorRunOptions struct {
	Limit int
}

// ResourceActivityProjectorRefreshOptions scopes an on-demand refresh to one repository and selected event kinds.
type ResourceActivityProjectorRefreshOptions struct {
	RefreshAfter time.Duration
	Force        bool
}

// ResourceActivityProjectorRunResult reports bounded projection work completed by one service run.
type ResourceActivityProjectorRunResult struct {
	Repositories int
	Events       int
	Errors       int
}

// ResourceActivityProjectorRefresher is the narrow on-demand projection port used by resource domains.
type ResourceActivityProjectorRefresher interface {
	RefreshRepositories(ctx context.Context, repos []reposstore.ResourceRepository, opts ResourceActivityProjectorRefreshOptions) (ResourceActivityProjectorRunResult, error)
}

// RunOnce discovers a bounded repository set and projects configured AGS event
// streams. Invalid optional wiring yields no work rather than a worker failure.
func (p *ResourceActivityProjector) RunOnce(ctx context.Context, opts ResourceActivityProjectorRunOptions) (ResourceActivityProjectorRunResult, error) {
	if p == nil || p.Repositories == nil || p.Activity == nil || p.Source == nil {
		return ResourceActivityProjectorRunResult{}, nil
	}
	cfg := p.normalizedConfig()
	limit := opts.Limit
	if limit <= 0 {
		limit = cfg.RepoLimit
	}
	repos, err := p.Repositories.ListActivityProjectorRepositories(ctx, reposstore.ListActivityProjectorRepositoriesOptions{
		Limit:     limit,
		Selectors: resourceActivityProjectorRepositorySelectors(cfg.ProjectionRules),
	})
	if err != nil {
		return ResourceActivityProjectorRunResult{}, err
	}
	return p.refreshRepositories(ctx, repos, cfg, 0)
}

// RefreshRepositories projects activity for an explicit repository set while
// honoring checkpoint freshness unless Force is set.
func (p *ResourceActivityProjector) RefreshRepositories(ctx context.Context, repos []reposstore.ResourceRepository, opts ResourceActivityProjectorRefreshOptions) (ResourceActivityProjectorRunResult, error) {
	if p == nil || p.Activity == nil || p.Source == nil {
		return ResourceActivityProjectorRunResult{}, nil
	}
	refreshAfter := opts.RefreshAfter
	if opts.Force {
		refreshAfter = 0
	} else if refreshAfter <= 0 {
		refreshAfter = DefaultResourceActivityRefreshAfter
	}
	return p.refreshRepositories(ctx, repos, p.normalizedConfig(), refreshAfter)
}

func (p *ResourceActivityProjector) refreshRepositories(ctx context.Context, repos []reposstore.ResourceRepository, cfg ResourceActivityProjectorConfig, refreshAfter time.Duration) (ResourceActivityProjectorRunResult, error) {
	result := ResourceActivityProjectorRunResult{Repositories: len(repos)}
	for _, rawRepo := range repos {
		repo, ok := resourceActivityProjectorRepository(rawRepo)
		if !ok {
			continue
		}
		target, ok := resourceActivityProjectorTargetForRepo(rawRepo, cfg.ProjectionRules)
		if !ok {
			continue
		}
		repo.AssetType = target.AssetType
		for _, kind := range target.Kinds {
			written, err := p.projectRepoKind(ctx, repo, kind, cfg, refreshAfter)
			result.Events += written
			if err != nil {
				result.Errors++
			}
		}
	}
	return result, nil
}

func (p *ResourceActivityProjector) projectRepoKind(ctx context.Context, repo ResourceActivityDataPlaneRepository, kind string, cfg ResourceActivityProjectorConfig, refreshAfter time.Duration) (int, error) {
	now := p.now()
	identity := activitystore.ProjectorCheckpointIdentity{
		WorkspaceID:  repo.WorkspaceID,
		ResourceType: repo.ResourceType,
		ResourceID:   repo.ResourceID,
		AssetType:    repo.AssetType,
		AssetID:      repo.AssetID,
		RepoRole:     repo.RepoRole,
		RepoFullName: repo.RepoFullName,
		EventKind:    resourceActivityProjectorCheckpointEventKind(repo, kind),
	}
	checkpoint, err := p.Activity.GetProjectorCheckpoint(ctx, identity)
	hasCheckpoint := true
	if errors.Is(err, gorm.ErrRecordNotFound) {
		hasCheckpoint = false
	} else if err != nil {
		return 0, err
	}
	if refreshAfter > 0 && hasFreshProjectorCheckpoint(checkpoint, hasCheckpoint, now, refreshAfter) {
		return 0, nil
	}
	project := func() (any, error) {
		return p.projectRepoKindWithCheckpoint(ctx, repo, kind, cfg, now, identity, checkpoint, hasCheckpoint)
	}
	if refreshAfter <= 0 {
		written, err := project()
		if err != nil {
			return 0, err
		}
		return written.(int), nil
	}
	written, err, _ := p.refreshFlights.Do(resourceActivityProjectorFlightKey(identity), project)
	if err != nil {
		return 0, err
	}
	count, ok := written.(int)
	if !ok {
		return 0, nil
	}
	return count, nil
}

func (p *ResourceActivityProjector) projectRepoKindWithCheckpoint(ctx context.Context, repo ResourceActivityDataPlaneRepository, kind string, cfg ResourceActivityProjectorConfig, now time.Time, identity activitystore.ProjectorCheckpointIdentity, checkpoint activitystore.ResourceActivityProjectorCheckpoint, hasCheckpoint bool) (int, error) {
	backfillSince := now.Add(-cfg.BackfillWindow)
	opts := ResourceActivityDataPlaneListOptions{
		BackfillSince: backfillSince,
		Limit:         cfg.PerKindLimit,
		Now:           now,
	}
	if hasCheckpoint {
		opts.Since = checkpoint.LastSeenAt
		opts.StateJSON = checkpoint.StateJSON
	}
	batch, err := p.Source.ListActivityEvents(ctx, repo, kind, opts)
	if err != nil {
		status := activitystore.ProjectorStatusError
		if errors.Is(err, ErrResourceActivityProjectorRateLimited) {
			status = activitystore.ProjectorStatusRateLimited
		}
		_, markErr := p.Activity.MarkProjectorCheckpointError(ctx, identity, activitystore.ProjectorCheckpointUpdate{
			Status:       status,
			LastError:    err.Error(),
			LastPolledAt: now,
		})
		if markErr != nil {
			return 0, markErr
		}
		return 0, err
	}
	candidates := selectProjectableDataPlaneEvents(batch.Events, checkpoint.LastSeenAt, checkpoint.LastSourceEventID, hasCheckpoint, backfillSince, cfg.PerKindLimit)
	lastSeen := checkpoint.LastSeenAt
	lastSourceEventID := checkpoint.LastSourceEventID
	written := 0
	for _, event := range candidates {
		if strings.TrimSpace(event.SourceEventID) == "" || strings.TrimSpace(event.EventType) == "" || event.OccurredAt.IsZero() {
			continue
		}
		event.Display = DataPlaneActivityDisplay(repo, event)
		if err := WriteResourceActivityEvent(ctx, p.Activity, ResourceActivityEventInput{
			WorkspaceID:   repo.WorkspaceID,
			AssetType:     repo.AssetType,
			AssetID:       repo.AssetID,
			RepoFullName:  repo.RepoFullName,
			EventType:     event.EventType,
			ActorType:     firstNonEmpty(event.ActorType, ActivityActorTypeUser),
			ActorID:       event.ActorID,
			ActorName:     event.ActorName,
			SummaryKey:    firstNonEmpty(event.SummaryKey, event.EventType),
			Details:       event.Details,
			Display:       event.Display,
			TargetKind:    event.TargetKind,
			TargetID:      event.TargetID,
			TargetURL:     event.TargetURL,
			Source:        activitystore.SourceGitea,
			SourceEventID: event.SourceEventID,
			OccurredAt:    event.OccurredAt,
		}); err != nil {
			_, markErr := p.Activity.MarkProjectorCheckpointError(ctx, identity, activitystore.ProjectorCheckpointUpdate{
				Status:       activitystore.ProjectorStatusError,
				LastError:    err.Error(),
				LastPolledAt: now,
			})
			if markErr != nil {
				return written, markErr
			}
			return written, err
		}
		written++
		if eventIsAfterCursor(event.OccurredAt, event.SourceEventID, lastSeen, lastSourceEventID) {
			occurred := event.OccurredAt.UTC()
			lastSeen = &occurred
			lastSourceEventID = event.SourceEventID
		}
	}
	stateJSON := strings.TrimSpace(batch.StateJSON)
	if stateJSON == "" {
		stateJSON = checkpoint.StateJSON
	}
	_, err = p.Activity.MarkProjectorCheckpointSuccess(ctx, identity, activitystore.ProjectorCheckpointUpdate{
		LastSeenAt:        lastSeen,
		LastSourceEventID: lastSourceEventID,
		StateJSON:         stateJSON,
		LastPolledAt:      now,
	})
	if err != nil {
		return written, err
	}
	return written, nil
}

func hasFreshProjectorCheckpoint(checkpoint activitystore.ResourceActivityProjectorCheckpoint, hasCheckpoint bool, now time.Time, refreshAfter time.Duration) bool {
	if !hasCheckpoint || checkpoint.LastPolledAt == nil || checkpoint.LastPolledAt.IsZero() {
		return false
	}
	lastPolledAt := checkpoint.LastPolledAt.UTC()
	return now.UTC().Sub(lastPolledAt) < refreshAfter
}

func selectProjectableDataPlaneEvents(events []ResourceActivityDataPlaneEvent, lastSeen *time.Time, lastSourceEventID string, hasCheckpoint bool, backfillSince time.Time, limit int) []ResourceActivityDataPlaneEvent {
	if limit <= 0 {
		limit = 500
	}
	candidates := make([]ResourceActivityDataPlaneEvent, 0, len(events))
	for _, event := range events {
		event.SourceEventID = strings.TrimSpace(event.SourceEventID)
		event.EventType = strings.TrimSpace(event.EventType)
		if event.SourceEventID == "" || event.EventType == "" || event.OccurredAt.IsZero() {
			continue
		}
		event.OccurredAt = event.OccurredAt.UTC()
		if (!hasCheckpoint || lastSeen == nil || lastSeen.IsZero()) && event.OccurredAt.Before(backfillSince) {
			continue
		}
		if hasCheckpoint && !eventIsAfterCursor(event.OccurredAt, event.SourceEventID, lastSeen, lastSourceEventID) {
			continue
		}
		candidates = append(candidates, event)
	}
	sort.SliceStable(candidates, func(i, j int) bool {
		if !candidates[i].OccurredAt.Equal(candidates[j].OccurredAt) {
			return candidates[i].OccurredAt.After(candidates[j].OccurredAt)
		}
		return candidates[i].SourceEventID > candidates[j].SourceEventID
	})
	if len(candidates) > limit {
		candidates = candidates[:limit]
	}
	sort.SliceStable(candidates, func(i, j int) bool {
		if !candidates[i].OccurredAt.Equal(candidates[j].OccurredAt) {
			return candidates[i].OccurredAt.Before(candidates[j].OccurredAt)
		}
		return candidates[i].SourceEventID < candidates[j].SourceEventID
	})
	return candidates
}

func eventIsAfterCursor(occurredAt time.Time, sourceEventID string, lastSeen *time.Time, lastSourceEventID string) bool {
	if lastSeen == nil || lastSeen.IsZero() {
		return true
	}
	occurredAt = occurredAt.UTC()
	last := lastSeen.UTC()
	if occurredAt.After(last) {
		return true
	}
	if occurredAt.Equal(last) && strings.TrimSpace(sourceEventID) > strings.TrimSpace(lastSourceEventID) {
		return true
	}
	return false
}

func (p *ResourceActivityProjector) normalizedConfig() ResourceActivityProjectorConfig {
	cfg := p.Config
	if cfg.RepoLimit <= 0 {
		cfg.RepoLimit = 25
	}
	if cfg.PerKindLimit <= 0 {
		cfg.PerKindLimit = 500
	}
	if cfg.BackfillWindow <= 0 {
		cfg.BackfillWindow = 90 * 24 * time.Hour
	}
	cfg.ProjectionRules = normalizeResourceActivityProjectionRules(cfg.ProjectionRules)
	if len(cfg.ProjectionRules) == 0 {
		cfg.ProjectionRules = DefaultResourceActivityProjectionRules()
	}
	return cfg
}

func (p *ResourceActivityProjector) now() time.Time {
	if p != nil && p.Now != nil {
		return p.Now().UTC()
	}
	return time.Now().UTC()
}

func resourceActivityProjectorRepository(repo reposstore.ResourceRepository) (ResourceActivityDataPlaneRepository, bool) {
	owner := strings.TrimSpace(repo.RepoOwner)
	name := strings.TrimSpace(repo.RepoName)
	fullName := strings.TrimSpace(repo.RepoFullName)
	if fullName == "" && owner != "" && name != "" {
		fullName = owner + "/" + name
	}
	if owner == "" || name == "" || fullName == "" {
		return ResourceActivityDataPlaneRepository{}, false
	}
	return ResourceActivityDataPlaneRepository{
		WorkspaceID:  strings.TrimSpace(repo.WorkspaceID),
		ResourceType: strings.TrimSpace(repo.ResourceType),
		ResourceID:   strings.TrimSpace(repo.ResourceID),
		AssetID:      strings.TrimSpace(repo.ResourceID),
		RepoRole:     strings.TrimSpace(repo.RepoRole),
		RepoOwner:    owner,
		RepoName:     name,
		RepoFullName: fullName,
		DefaultRef:   strings.TrimSpace(repo.DefaultRef),
	}, true
}

// ResourceActivityDataPlaneRepositoryForResourceRepository converts a valid
// repository binding into the projector's service input shape.
func ResourceActivityDataPlaneRepositoryForResourceRepository(repo reposstore.ResourceRepository) (ResourceActivityDataPlaneRepository, bool) {
	return resourceActivityProjectorRepository(repo)
}

// ResourceActivityAssetTypeForResourceType maps supported resource repository
// types to activity-store asset categories.
func ResourceActivityAssetTypeForResourceType(resourceType string) (string, bool) {
	switch strings.TrimSpace(resourceType) {
	case reposstore.ResourceTypeMemory:
		return activitystore.AssetTypeMemory, true
	case reposstore.ResourceTypeSkillRepository:
		return activitystore.AssetTypeSkill, true
	case reposstore.ResourceTypeKBSource:
		return activitystore.AssetTypeKB, true
	default:
		return "", false
	}
}

// DefaultResourceActivityProjectionRules returns a copy of the service-owned
// default projection policy so callers cannot mutate shared state.
func DefaultResourceActivityProjectionRules() []ResourceActivityProjectionRule {
	return cloneResourceActivityProjectionRules(defaultResourceActivityProjectionRules)
}

func resourceActivityProjectorRepositorySelectors(rules []ResourceActivityProjectionRule) []reposstore.ActivityProjectorRepositorySelector {
	rules = normalizeResourceActivityProjectionRules(rules)
	rolesByType := map[string]map[string]struct{}{}
	roleOrderByType := map[string][]string{}
	typeOrder := make([]string, 0, len(rules))
	for _, rule := range rules {
		if _, ok := rolesByType[rule.ResourceType]; !ok {
			rolesByType[rule.ResourceType] = map[string]struct{}{}
			roleOrderByType[rule.ResourceType] = []string{}
			typeOrder = append(typeOrder, rule.ResourceType)
		}
		for _, role := range rule.RepoRoles {
			if _, ok := rolesByType[rule.ResourceType][role]; ok {
				continue
			}
			rolesByType[rule.ResourceType][role] = struct{}{}
			roleOrderByType[rule.ResourceType] = append(roleOrderByType[rule.ResourceType], role)
		}
	}
	selectors := make([]reposstore.ActivityProjectorRepositorySelector, 0, len(typeOrder))
	for _, resourceType := range typeOrder {
		selectors = append(selectors, reposstore.ActivityProjectorRepositorySelector{
			ResourceType: resourceType,
			RepoRoles:    append([]string(nil), roleOrderByType[resourceType]...),
		})
	}
	return selectors
}

// ProjectableResourceActivityRepositories filters repository bindings down to
// the default activity-projected surfaces.
func ProjectableResourceActivityRepositories(repos []reposstore.ResourceRepository) []reposstore.ResourceRepository {
	return ResourceActivityRepositoriesForRules(DefaultResourceActivityProjectionRules(), repos)
}

// ResourceActivityRepositoriesForRules filters repository bindings according to
// the supplied activity projection rules.
func ResourceActivityRepositoriesForRules(rules []ResourceActivityProjectionRule, repos []reposstore.ResourceRepository) []reposstore.ResourceRepository {
	rules = normalizeResourceActivityProjectionRules(rules)
	if len(rules) == 0 {
		return nil
	}
	allowed := map[string]map[string]struct{}{}
	for _, rule := range rules {
		roles := allowed[rule.ResourceType]
		if roles == nil {
			roles = map[string]struct{}{}
			allowed[rule.ResourceType] = roles
		}
		for _, role := range rule.RepoRoles {
			roles[role] = struct{}{}
		}
	}
	out := make([]reposstore.ResourceRepository, 0, len(repos))
	for _, repo := range repos {
		roles := allowed[strings.TrimSpace(repo.ResourceType)]
		if len(roles) == 0 {
			continue
		}
		if _, ok := roles[strings.TrimSpace(repo.RepoRole)]; !ok {
			continue
		}
		if strings.TrimSpace(repo.RepoFullName) == "" {
			continue
		}
		out = append(out, repo)
	}
	return out
}

func resourceActivityProjectorTargetForRepo(repo reposstore.ResourceRepository, rules []ResourceActivityProjectionRule) (resourceActivityProjectorTargetSpec, bool) {
	resourceType := strings.TrimSpace(repo.ResourceType)
	repoRole := strings.TrimSpace(repo.RepoRole)
	if resourceType == "" || repoRole == "" {
		return resourceActivityProjectorTargetSpec{}, false
	}
	assetType := ""
	kinds := make([]string, 0)
	seenKinds := map[string]struct{}{}
	for _, rule := range normalizeResourceActivityProjectionRules(rules) {
		if rule.ResourceType != resourceType || !resourceActivityRuleHasRepoRole(rule, repoRole) {
			continue
		}
		if assetType == "" {
			assetType = rule.AssetType
		}
		if rule.AssetType != assetType {
			return resourceActivityProjectorTargetSpec{}, false
		}
		for _, kind := range rule.EventKinds {
			if _, ok := seenKinds[kind]; ok {
				continue
			}
			kinds = append(kinds, kind)
			seenKinds[kind] = struct{}{}
		}
	}
	if assetType == "" || len(kinds) == 0 {
		return resourceActivityProjectorTargetSpec{}, false
	}
	return resourceActivityProjectorTargetSpec{AssetType: assetType, Kinds: kinds}, true
}

func normalizeResourceActivityProjectionRules(rules []ResourceActivityProjectionRule) []ResourceActivityProjectionRule {
	out := make([]ResourceActivityProjectionRule, 0, len(rules))
	for _, rule := range rules {
		resourceType := strings.TrimSpace(rule.ResourceType)
		assetType := strings.TrimSpace(rule.AssetType)
		roles := uniqueTrimmedStrings(rule.RepoRoles)
		kinds := uniqueTrimmedStrings(rule.EventKinds)
		if resourceType == "" || assetType == "" || len(roles) == 0 || len(kinds) == 0 {
			continue
		}
		out = append(out, ResourceActivityProjectionRule{
			ResourceType: resourceType,
			RepoRoles:    roles,
			AssetType:    assetType,
			EventKinds:   kinds,
		})
	}
	return out
}

func cloneResourceActivityProjectionRules(rules []ResourceActivityProjectionRule) []ResourceActivityProjectionRule {
	out := make([]ResourceActivityProjectionRule, 0, len(rules))
	for _, rule := range rules {
		out = append(out, ResourceActivityProjectionRule{
			ResourceType: rule.ResourceType,
			RepoRoles:    append([]string(nil), rule.RepoRoles...),
			AssetType:    rule.AssetType,
			EventKinds:   append([]string(nil), rule.EventKinds...),
		})
	}
	return out
}

func resourceActivityRuleHasRepoRole(rule ResourceActivityProjectionRule, repoRole string) bool {
	for _, role := range rule.RepoRoles {
		if strings.TrimSpace(role) == repoRole {
			return true
		}
	}
	return false
}

func uniqueTrimmedStrings(values []string) []string {
	out := make([]string, 0, len(values))
	seen := map[string]struct{}{}
	for _, value := range values {
		value = strings.TrimSpace(value)
		if value == "" {
			continue
		}
		if _, ok := seen[value]; ok {
			continue
		}
		out = append(out, value)
		seen[value] = struct{}{}
	}
	return out
}

func resourceActivityProjectorFlightKey(identity activitystore.ProjectorCheckpointIdentity) string {
	return strings.Join([]string{
		strings.TrimSpace(identity.WorkspaceID),
		strings.TrimSpace(identity.ResourceType),
		strings.TrimSpace(identity.ResourceID),
		strings.TrimSpace(identity.RepoRole),
		strings.TrimSpace(identity.EventKind),
	}, "\x00")
}

func resourceActivityProjectorCheckpointEventKind(repo ResourceActivityDataPlaneRepository, kind string) string {
	kind = strings.TrimSpace(kind)
	if kind != activitystore.ProjectorKindCommits {
		return kind
	}
	ref := strings.TrimSpace(repo.DefaultRef)
	if ref == "" {
		return kind
	}
	normalizedRef := strings.TrimSpace(strings.TrimPrefix(ref, "refs/heads/"))
	if normalizedRef == "" {
		normalizedRef = ref
	}
	value := kind + ":" + normalizedRef
	if len(value) <= 64 {
		return value
	}
	sum := sha256.Sum256([]byte(ref))
	return kind + ":ref:" + hex.EncodeToString(sum[:])[:16]
}

func sourceEventID(repo ResourceActivityDataPlaneRepository, parts ...string) string {
	return ResourceActivitySourceEventID(repo, parts...)
}