package server

import (
	"context"
	"encoding/json"
	"errors"
	"log/slog"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/observability/logging"
	"github.com/tidbcloud/AgentTeam/internal/service/resource/activity"

	agsserver "github.com/pingcap/agent-git-service/server"

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

var errResourceActivityBridgeNotStarted = errors.New("resource activity repository event bridge is not started")

const defaultAGSResourceActivityEventQueueSize = 256

type agsResourceActivityEventBridge struct {
	repositories     *reposstore.Store
	activity         *activitystore.Store
	contentAuthority kbSourceEventAuthorityReader
	queue            chan agsserver.RepositoryEvent
	done             chan struct{}
	startOnce        sync.Once
	stopOnce         sync.Once
	mu               sync.RWMutex
	started          bool
	closed           bool
	processCtx       context.Context
	cancel           context.CancelFunc
	processEvent     func(context.Context, agsserver.RepositoryEvent) error

	receivedTotal    atomic.Uint64
	droppedTotal     atomic.Uint64
	writeFailedTotal atomic.Uint64
	unsupportedTotal atomic.Uint64
}

type kbSourceEventAuthorityReader interface {
	ListAuthoritiesForBindings(context.Context, string, string, []string) (map[string]content.Authority, error)
}

type kbSourceEventBinding struct {
	workspaceID string
	sourceID    string
}

func newAGSResourceActivityEventBridge(repositories *reposstore.Store, activityStore *activitystore.Store, queueSize int) *agsResourceActivityEventBridge {
	if queueSize <= 0 {
		queueSize = defaultAGSResourceActivityEventQueueSize
	}
	return &agsResourceActivityEventBridge{
		repositories: repositories,
		activity:     activityStore,
		queue:        make(chan agsserver.RepositoryEvent, queueSize),
		done:         make(chan struct{}),
	}
}

func (b *agsResourceActivityEventBridge) Start() {
	if b == nil {
		return
	}
	b.startOnce.Do(func() {
		b.mu.Lock()
		defer b.mu.Unlock()
		if b.closed {
			return
		}
		b.started = true
		b.processCtx, b.cancel = context.WithCancel(context.Background())
		logging.Go(slog.Default(), b.run)
	})
}

func (b *agsResourceActivityEventBridge) HandleRepositoryEvent(ctx context.Context, event agsserver.RepositoryEvent) error {
	if b == nil {
		return nil
	}
	b.mu.RLock()
	if b.closed || !b.started {
		b.mu.RUnlock()
		return errResourceActivityBridgeNotStarted
	}
	b.receivedTotal.Add(1)
	safe := agsserver.RepositoryEvent{
		RepositoryID: event.RepositoryID,
		Event:        strings.TrimSpace(event.Event),
		Action:       strings.TrimSpace(event.Action),
		PayloadJSON:  append([]byte(nil), event.PayloadJSON...),
	}
	select {
	case b.queue <- safe:
		b.mu.RUnlock()
		return nil
	default:
		b.mu.RUnlock()
		b.droppedTotal.Add(1)
		slog.WarnContext(ctx, "resource activity AGS event bridge queue full; dropping event", "repo_id", event.RepositoryID, "event", event.Event, "action", event.Action)
		return nil
	}
}

func (b *agsResourceActivityEventBridge) Close() {
	if b == nil {
		return
	}
	_ = b.Quiesce(context.Background())
}

func (b *agsResourceActivityEventBridge) Quiesce(ctx context.Context) error {
	if b == nil {
		return nil
	}
	b.stopOnce.Do(func() {
		b.mu.Lock()
		b.closed = true
		if b.started {
			close(b.queue)
		} else {
			close(b.done)
		}
		b.mu.Unlock()
	})
	select {
	case <-b.done:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (b *agsResourceActivityEventBridge) run() {
	defer close(b.done)
	for event := range b.queue {
		func() {
			defer func() {
				if recovered := recover(); recovered != nil {
					b.writeFailedTotal.Add(1)
					slog.Warn("resource activity AGS event bridge panicked", "repo_id", event.RepositoryID, "event", event.Event, "action", event.Action, "panic", recovered)
				}
			}()
			process := b.process
			if b.processEvent != nil {
				process = b.processEvent
			}
			if err := process(b.processCtx, event); err != nil {
				b.writeFailedTotal.Add(1)
				slog.Warn("resource activity AGS event bridge failed", "repo_id", event.RepositoryID, "event", event.Event, "action", event.Action, "error", err)
			}
		}()
	}
}

func (b *agsResourceActivityEventBridge) process(ctx context.Context, event agsserver.RepositoryEvent) error {
	if b == nil || b.repositories == nil || b.activity == nil {
		return nil
	}
	var payload agsWebhookPayload
	if len(event.PayloadJSON) > 0 {
		if err := json.Unmarshal(event.PayloadJSON, &payload); err != nil {
			return err
		}
	}
	repos, err := b.repositories.ListByAGSRepoID(ctx, strconv.FormatUint(uint64(event.RepositoryID), 10))
	if err != nil {
		return err
	}
	if len(repos) == 0 && strings.TrimSpace(payload.Repository.FullName) != "" {
		repos, err = b.repositories.ListByRepoFullName(ctx, payload.Repository.FullName)
		if err != nil {
			return err
		}
	}
	if len(repos) == 0 {
		return nil
	}
	nativeKBSources, err := nativeKBSourceEventBindings(ctx, b.contentAuthority, repos)
	if err != nil {
		return err
	}
	now := time.Now().UTC()
	for _, repo := range repos {
		if strings.TrimSpace(repo.ResourceType) == reposstore.ResourceTypeAutomationRepository {
			continue
		}
		if _, native := nativeKBSources[kbSourceEventBinding{workspaceID: strings.TrimSpace(repo.WorkspaceID), sourceID: strings.TrimSpace(repo.ResourceID)}]; native {
			continue
		}
		dpRepo, ok := activity.ResourceActivityDataPlaneRepositoryForResourceRepository(repo)
		if !ok {
			continue
		}
		assetType, ok := activity.ResourceActivityAssetTypeForResourceType(repo.ResourceType)
		if !ok {
			continue
		}
		dpRepo.AssetType = assetType
		events := mapAGSEventToResourceActivity(dpRepo, event, payload, now)
		if len(events) == 0 {
			b.unsupportedTotal.Add(1)
			slog.WarnContext(ctx, "resource activity AGS event bridge ignored unsupported event", "repo_id", event.RepositoryID, "event", event.Event, "action", event.Action, "repository", repo.RepoFullName)
			continue
		}
		for _, mapped := range events {
			if err := activity.WriteResourceActivityEvent(ctx, b.activity, activity.ResourceActivityEventInput{
				WorkspaceID:   dpRepo.WorkspaceID,
				AssetType:     dpRepo.AssetType,
				AssetID:       dpRepo.AssetID,
				RepoFullName:  dpRepo.RepoFullName,
				EventType:     mapped.EventType,
				ActorType:     mapped.ActorType,
				ActorID:       mapped.ActorID,
				ActorName:     mapped.ActorName,
				SummaryKey:    mapped.SummaryKey,
				Details:       mapped.Details,
				Display:       mapped.Display,
				TargetKind:    mapped.TargetKind,
				TargetID:      mapped.TargetID,
				TargetURL:     mapped.TargetURL,
				Source:        activitystore.SourceGitea,
				SourceEventID: mapped.SourceEventID,
				OccurredAt:    mapped.OccurredAt,
			}); err != nil {
				return err
			}
		}
	}
	return nil
}

// nativeKBSourceEventBindings resolves only the exact KB source mappings
// selected by one AGS repository event. Native bindings must not consume the
// retired AGS event stream after their one-way authority cutover; missing rows
// remain legacy and unknown persisted states fail closed.
func nativeKBSourceEventBindings(ctx context.Context, authorityReader kbSourceEventAuthorityReader, repos []reposstore.ResourceRepository) (map[kbSourceEventBinding]struct{}, error) {
	native := make(map[kbSourceEventBinding]struct{})
	if authorityReader == nil {
		return native, nil
	}
	byWorkspace := make(map[string][]string)
	for _, repo := range repos {
		if strings.TrimSpace(repo.ResourceType) != reposstore.ResourceTypeKBSource {
			continue
		}
		workspaceID := strings.TrimSpace(repo.WorkspaceID)
		sourceID := strings.TrimSpace(repo.ResourceID)
		if workspaceID == "" || sourceID == "" {
			return nil, errors.New("KB source activity repository binding is incomplete")
		}
		byWorkspace[workspaceID] = append(byWorkspace[workspaceID], sourceID)
	}
	for workspaceID, sourceIDs := range byWorkspace {
		for start := 0; start < len(sourceIDs); start += content.MaxAuthorityBindingBatch {
			end := min(start+content.MaxAuthorityBindingBatch, len(sourceIDs))
			authorities, err := authorityReader.ListAuthoritiesForBindings(ctx, workspaceID, "kb_source", sourceIDs[start:end])
			if err != nil {
				return nil, err
			}
			for _, sourceID := range sourceIDs[start:end] {
				authority, found := authorities[sourceID]
				if !found || strings.TrimSpace(authority.State) == content.AuthorityStateLegacyActive {
					continue
				}
				if strings.TrimSpace(authority.State) != content.AuthorityStateNativeActive {
					return nil, errors.New("KB source activity authority is invalid")
				}
				native[kbSourceEventBinding{workspaceID: workspaceID, sourceID: sourceID}] = struct{}{}
			}
		}
	}
	return native, nil
}

type agsWebhookPayload struct {
	Action      string             `json:"action"`
	Ref         string             `json:"ref"`
	Before      string             `json:"before"`
	After       string             `json:"after"`
	Created     bool               `json:"created"`
	Deleted     bool               `json:"deleted"`
	Compare     string             `json:"compare"`
	Commits     []agsWebhookCommit `json:"commits"`
	Repository  agsWebhookRepo     `json:"repository"`
	Pusher      agsWebhookActor    `json:"pusher"`
	Sender      agsWebhookUser     `json:"sender"`
	Release     agsWebhookRelease  `json:"release"`
	Issue       agsWebhookIssue    `json:"issue"`
	Comment     agsWebhookComment  `json:"comment"`
	PullRequest agsWebhookPull     `json:"pull_request"`
	Review      agsWebhookReview   `json:"review"`
	Requested   agsWebhookUser     `json:"requested_reviewer"`
	Team        agsWebhookTeam     `json:"requested_team"`
	Number      int                `json:"number"`
}

type agsWebhookRepo struct {
	FullName string `json:"full_name"`
	HTMLURL  string `json:"html_url"`
}

type agsWebhookActor struct {
	Name  string `json:"name"`
	Email string `json:"email"`
}

type agsWebhookUser struct {
	ID       int64  `json:"id"`
	Login    string `json:"login"`
	UserName string `json:"username"`
	Name     string `json:"name"`
	FullName string `json:"full_name"`
	Email    string `json:"email"`
}

type agsWebhookTeam struct {
	ID   int64  `json:"id"`
	Name string `json:"name"`
	Slug string `json:"slug"`
}

type agsWebhookCommit struct {
	ID        string          `json:"id"`
	SHA       string          `json:"sha"`
	Message   string          `json:"message"`
	Timestamp *time.Time      `json:"timestamp"`
	URL       string          `json:"url"`
	HTMLURL   string          `json:"html_url"`
	Author    agsWebhookActor `json:"author"`
	Committer agsWebhookActor `json:"committer"`
}

type agsWebhookRelease struct {
	ID          int64          `json:"id"`
	TagName     string         `json:"tag_name"`
	Name        string         `json:"name"`
	HTMLURL     string         `json:"html_url"`
	Author      agsWebhookUser `json:"author"`
	CreatedAt   *time.Time     `json:"created_at"`
	PublishedAt *time.Time     `json:"published_at"`
	UpdatedAt   *time.Time     `json:"updated_at"`
}

type agsWebhookIssue struct {
	ID        int64          `json:"id"`
	Number    int            `json:"number"`
	Title     string         `json:"title"`
	State     string         `json:"state"`
	HTMLURL   string         `json:"html_url"`
	User      agsWebhookUser `json:"user"`
	CreatedAt *time.Time     `json:"created_at"`
	UpdatedAt *time.Time     `json:"updated_at"`
	ClosedAt  *time.Time     `json:"closed_at"`
}

type agsWebhookPull struct {
	ID             int64          `json:"id"`
	Number         int            `json:"number"`
	Title          string         `json:"title"`
	State          string         `json:"state"`
	HTMLURL        string         `json:"html_url"`
	User           agsWebhookUser `json:"user"`
	Merged         bool           `json:"merged"`
	MergeCommitSHA string         `json:"merge_commit_sha"`
	CreatedAt      *time.Time     `json:"created_at"`
	UpdatedAt      *time.Time     `json:"updated_at"`
	ClosedAt       *time.Time     `json:"closed_at"`
	MergedAt       *time.Time     `json:"merged_at"`
	Head           struct {
		SHA string `json:"sha"`
		Ref string `json:"ref"`
	} `json:"head"`
	Base struct {
		SHA string `json:"sha"`
		Ref string `json:"ref"`
	} `json:"base"`
}

type agsWebhookReview struct {
	ID          int64          `json:"id"`
	State       string         `json:"state"`
	HTMLURL     string         `json:"html_url"`
	User        agsWebhookUser `json:"user"`
	SubmittedAt *time.Time     `json:"submitted_at"`
	UpdatedAt   *time.Time     `json:"updated_at"`
}

type agsWebhookComment struct {
	ID        int64          `json:"id"`
	HTMLURL   string         `json:"html_url"`
	User      agsWebhookUser `json:"user"`
	Body      string         `json:"body"`
	CreatedAt *time.Time     `json:"created_at"`
	UpdatedAt *time.Time     `json:"updated_at"`
}

func mapAGSEventToResourceActivity(repo activity.ResourceActivityDataPlaneRepository, event agsserver.RepositoryEvent, payload agsWebhookPayload, now time.Time) []activity.ResourceActivityDataPlaneEvent {
	if now.IsZero() {
		now = time.Now().UTC()
	}
	switch strings.TrimSpace(event.Event) {
	case "push":
		return mapAGSPushEvent(repo, payload, now)
	case "release":
		return mapAGSReleaseEvent(repo, event.Action, payload, now)
	case "issues":
		return mapAGSIssueEvent(repo, event.Action, payload, now)
	case "issue_comment":
		return mapAGSIssueCommentEvent(repo, event.Action, payload, now)
	case "pull_request":
		return mapAGSPullRequestEvent(repo, event.Action, payload, now)
	case "pull_request_review":
		return mapAGSPullRequestReviewEvent(repo, event.Action, payload, now)
	case "pull_request_review_comment":
		return mapAGSPullRequestReviewCommentEvent(repo, event.Action, payload, now)
	default:
		return nil
	}
}

func mapAGSPushEvent(repo activity.ResourceActivityDataPlaneRepository, payload agsWebhookPayload, now time.Time) []activity.ResourceActivityDataPlaneEvent {
	out := make([]activity.ResourceActivityDataPlaneEvent, 0, len(payload.Commits)+1)
	refKind, refName := pushRefKindName(payload.Ref)
	for _, commit := range payload.Commits {
		sha := firstNonEmptyString(commit.ID, commit.SHA)
		if strings.TrimSpace(sha) == "" {
			continue
		}
		occurred := optionalTime(commit.Timestamp)
		if occurred.IsZero() {
			occurred = now
		}
		actorID, actorName := commitActor(commit, payload)
		gitDisplay := activity.GitActorActivityDisplay(actorID, actorName, firstNonEmptyString(commit.Author.Email, commit.Committer.Email), actorIDFromWebhookUser(payload.Sender, ""))
		ref := strings.TrimSpace(payload.Ref)
		out = append(out, activity.WithDataPlaneActivityDisplay(repo, activity.ResourceActivityDataPlaneEvent{
			SourceEventID: activity.ResourceActivitySourceEventID(repo, "commit", ref, sha),
			EventType:     "commit_pushed",
			ActorType:     activity.ActivityActorTypeUser,
			ActorID:       actorID,
			ActorName:     actorName,
			SummaryKey:    "commit_pushed",
			TargetKind:    "commit",
			TargetID:      strings.TrimSpace(sha),
			TargetURL:     firstNonEmptyString(commit.HTMLURL, commit.URL),
			OccurredAt:    occurred,
			Details: map[string]any{
				"sha":     shortSHA(sha),
				"message": firstLine(commit.Message),
				"ref":     ref,
			},
		}, gitDisplay))
	}
	if refName == "" || (!payload.Created && !payload.Deleted) {
		return out
	}
	if refKind == "tag" {
		if payload.Created {
			out = append(out, activity.ResourceActivityDataPlaneEvent{
				SourceEventID: activity.ResourceActivitySourceEventID(repo, "tag", refName, "created", firstNonEmptyString(payload.After, "unknown")),
				EventType:     "tag_created",
				ActorType:     activity.ActivityActorTypeUser,
				ActorID:       actorIDFromWebhookUser(payload.Sender, payload.Pusher.Name),
				ActorName:     actorNameFromWebhookUser(payload.Sender, payload.Pusher.Name),
				SummaryKey:    "tag_created",
				Details:       map[string]any{"tagName": refName, "sha": shortSHA(payload.After)},
				TargetKind:    "tag",
				TargetID:      refName,
				OccurredAt:    now,
			})
		}
		if payload.Deleted {
			out = append(out, activity.ResourceActivityDataPlaneEvent{
				SourceEventID: activity.ResourceActivitySourceEventID(repo, "tag", refName, "deleted"),
				EventType:     "tag_deleted",
				ActorType:     activity.ActivityActorTypeUser,
				ActorID:       actorIDFromWebhookUser(payload.Sender, payload.Pusher.Name),
				ActorName:     actorNameFromWebhookUser(payload.Sender, payload.Pusher.Name),
				SummaryKey:    "tag_deleted",
				Details:       map[string]any{"tagName": refName},
				TargetKind:    "tag",
				TargetID:      refName,
				OccurredAt:    now,
			})
		}
		return out
	}
	if refKind == "branch" {
		action := "created"
		eventType := "branch_created"
		sourceSHA := firstNonEmptyString(payload.After, "unknown")
		if payload.Deleted {
			action = "deleted"
			eventType = "branch_deleted"
			sourceSHA = firstNonEmptyString(payload.Before, "unknown")
		}
		out = append(out, activity.ResourceActivityDataPlaneEvent{
			SourceEventID: activity.ResourceActivitySourceEventID(repo, "branch", refName, action, sourceSHA),
			EventType:     eventType,
			ActorType:     activity.ActivityActorTypeUser,
			ActorID:       actorIDFromWebhookUser(payload.Sender, payload.Pusher.Name),
			ActorName:     actorNameFromWebhookUser(payload.Sender, payload.Pusher.Name),
			SummaryKey:    eventType,
			Details:       map[string]any{"branchName": refName, "sha": shortSHA(sourceSHA)},
			TargetKind:    "branch",
			TargetID:      refName,
			OccurredAt:    now,
		})
	}
	return out
}

func mapAGSReleaseEvent(repo activity.ResourceActivityDataPlaneRepository, action string, payload agsWebhookPayload, now time.Time) []activity.ResourceActivityDataPlaneEvent {
	release := payload.Release
	if release.ID == 0 {
		return nil
	}
	action = normalizeAction(action, payload.Action)
	eventTypeByAction := map[string]string{"published": "release_published", "edited": "release_edited", "deleted": "release_deleted"}
	eventType := eventTypeByAction[action]
	if eventType == "" {
		return nil
	}
	occurred := firstNonZeroTime(optionalTime(release.PublishedAt), optionalTime(release.UpdatedAt), optionalTime(release.CreatedAt), now)
	id := strconv.FormatInt(release.ID, 10)
	return []activity.ResourceActivityDataPlaneEvent{{
		SourceEventID: activity.ResourceActivitySourceEventID(repo, "release", id, action),
		EventType:     eventType,
		ActorType:     activity.ActivityActorTypeUser,
		ActorID:       actorIDFromWebhookUser(payload.Sender, actorIDFromWebhookUser(release.Author, "")),
		ActorName:     actorNameFromWebhookUser(payload.Sender, actorNameFromWebhookUser(release.Author, "")),
		SummaryKey:    eventType,
		Details:       map[string]any{"releaseName": firstNonEmptyString(release.Name, release.TagName, id), "tagName": strings.TrimSpace(release.TagName)},
		TargetKind:    "release",
		TargetID:      id,
		TargetURL:     strings.TrimSpace(release.HTMLURL),
		OccurredAt:    occurred,
	}}
}

func mapAGSIssueEvent(repo activity.ResourceActivityDataPlaneRepository, action string, payload agsWebhookPayload, now time.Time) []activity.ResourceActivityDataPlaneEvent {
	issue := payload.Issue
	if issue.Number <= 0 {
		return nil
	}
	action = normalizeAction(action, payload.Action)
	eventTypeByAction := map[string]string{"opened": "issue_opened", "closed": "issue_closed", "reopened": "issue_reopened", "edited": "issue_edited", "deleted": "issue_deleted"}
	eventType := eventTypeByAction[action]
	if eventType == "" {
		return nil
	}
	occurred := issueOccurredAt(issue.CreatedAt, issue.UpdatedAt, issue.ClosedAt, action, now)
	number := strconv.Itoa(issue.Number)
	parts := []string{"issue", number, action}
	if action == "closed" || action == "reopened" {
		parts = append(parts, unixNanoString(occurred))
	}
	return []activity.ResourceActivityDataPlaneEvent{{
		SourceEventID: activity.ResourceActivitySourceEventID(repo, parts...),
		EventType:     eventType,
		ActorType:     activity.ActivityActorTypeUser,
		ActorID:       actorIDFromWebhookUser(payload.Sender, actorIDFromWebhookUser(issue.User, "")),
		ActorName:     actorNameFromWebhookUser(payload.Sender, actorNameFromWebhookUser(issue.User, "")),
		SummaryKey:    eventType,
		Details:       map[string]any{"number": issue.Number, "title": strings.TrimSpace(issue.Title), "state": strings.TrimSpace(issue.State)},
		TargetKind:    "issue",
		TargetID:      number,
		TargetURL:     strings.TrimSpace(issue.HTMLURL),
		OccurredAt:    occurred,
	}}
}

func mapAGSIssueCommentEvent(repo activity.ResourceActivityDataPlaneRepository, action string, payload agsWebhookPayload, now time.Time) []activity.ResourceActivityDataPlaneEvent {
	comment := payload.Comment
	if comment.ID == 0 {
		return nil
	}
	action = normalizeAction(action, payload.Action)
	eventTypeByAction := map[string]string{"created": "issue_comment_created", "edited": "issue_comment_edited", "deleted": "issue_comment_deleted"}
	eventType := eventTypeByAction[action]
	if eventType == "" {
		return nil
	}
	occurred := firstNonZeroTime(optionalTime(comment.UpdatedAt), optionalTime(comment.CreatedAt), now)
	id := strconv.FormatInt(comment.ID, 10)
	return []activity.ResourceActivityDataPlaneEvent{{
		SourceEventID: activity.ResourceActivitySourceEventID(repo, "issue_comment", id, action),
		EventType:     eventType,
		ActorType:     activity.ActivityActorTypeUser,
		ActorID:       actorIDFromWebhookUser(payload.Sender, actorIDFromWebhookUser(comment.User, "")),
		ActorName:     actorNameFromWebhookUser(payload.Sender, actorNameFromWebhookUser(comment.User, "")),
		SummaryKey:    eventType,
		Details:       map[string]any{"commentID": id, "number": payload.Issue.Number},
		TargetKind:    "issue_comment",
		TargetID:      id,
		TargetURL:     strings.TrimSpace(comment.HTMLURL),
		OccurredAt:    occurred,
	}}
}

func mapAGSPullRequestEvent(repo activity.ResourceActivityDataPlaneRepository, action string, payload agsWebhookPayload, now time.Time) []activity.ResourceActivityDataPlaneEvent {
	pull := payload.PullRequest
	number := pull.Number
	if number <= 0 {
		number = payload.Number
	}
	if number <= 0 {
		return nil
	}
	action = normalizeAction(action, payload.Action)
	eventTypeByAction := map[string]string{
		"opened":                 "pr_opened",
		"closed":                 "pr_closed",
		"reopened":               "pr_reopened",
		"synchronize":            "pr_synchronized",
		"edited":                 "pr_edited",
		"ready_for_review":       "pr_ready_for_review",
		"review_requested":       "pr_review_requested",
		"review_request_removed": "pr_review_request_removed",
	}
	if action == "closed" && pull.Merged {
		eventTypeByAction["closed"] = "pr_merged"
	}
	eventType := eventTypeByAction[action]
	if eventType == "" {
		return nil
	}
	occurred := pullOccurredAt(pull, action, now)
	numberText := strconv.Itoa(number)
	sourceSuffix := action
	if eventType == "pr_merged" {
		sourceSuffix = "merged"
	}
	parts := []string{"pull", numberText, sourceSuffix}
	if action == "opened" || action == "closed" || action == "reopened" || eventType == "pr_merged" {
		parts = append(parts, unixNanoString(occurred))
	} else if action == "synchronize" {
		parts = append(parts, firstNonEmptyString(pull.Head.SHA, unixNanoString(occurred)))
	} else {
		parts = append(parts, unixNanoString(occurred))
	}
	details := map[string]any{"number": number, "title": strings.TrimSpace(pull.Title), "state": strings.TrimSpace(pull.State)}
	if payload.Requested.Login != "" || payload.Requested.ID != 0 {
		details["requestedReviewer"] = actorNameFromWebhookUser(payload.Requested, actorIDFromWebhookUser(payload.Requested, ""))
	}
	if payload.Team.Name != "" || payload.Team.Slug != "" {
		details["requestedTeam"] = firstNonEmptyString(payload.Team.Name, payload.Team.Slug)
	}
	return []activity.ResourceActivityDataPlaneEvent{{
		SourceEventID: activity.ResourceActivitySourceEventID(repo, parts...),
		EventType:     eventType,
		ActorType:     activity.ActivityActorTypeUser,
		ActorID:       actorIDFromWebhookUser(payload.Sender, actorIDFromWebhookUser(pull.User, "")),
		ActorName:     actorNameFromWebhookUser(payload.Sender, actorNameFromWebhookUser(pull.User, "")),
		SummaryKey:    eventType,
		Details:       details,
		TargetKind:    "pull",
		TargetID:      numberText,
		TargetURL:     strings.TrimSpace(pull.HTMLURL),
		OccurredAt:    occurred,
	}}
}

func mapAGSPullRequestReviewEvent(repo activity.ResourceActivityDataPlaneRepository, action string, payload agsWebhookPayload, now time.Time) []activity.ResourceActivityDataPlaneEvent {
	review := payload.Review
	if review.ID == 0 {
		return nil
	}
	number := payload.PullRequest.Number
	if number <= 0 {
		number = payload.Number
	}
	if number <= 0 {
		return nil
	}
	action = normalizeAction(action, payload.Action)
	eventTypeByAction := map[string]string{"created": "pr_review_created", "submitted": "pr_reviewed", "edited": "pr_review_edited", "dismissed": "pr_review_dismissed", "deleted": "pr_review_deleted"}
	eventType := eventTypeByAction[action]
	if eventType == "" {
		return nil
	}
	occurred := firstNonZeroTime(optionalTime(review.SubmittedAt), optionalTime(review.UpdatedAt), now)
	numberText := strconv.Itoa(number)
	reviewID := strconv.FormatInt(review.ID, 10)
	parts := []string{"pull", numberText, "review", reviewID}
	if action != "submitted" {
		parts = append(parts, action)
	}
	return []activity.ResourceActivityDataPlaneEvent{{
		SourceEventID: activity.ResourceActivitySourceEventID(repo, parts...),
		EventType:     eventType,
		ActorType:     activity.ActivityActorTypeUser,
		ActorID:       actorIDFromWebhookUser(payload.Sender, actorIDFromWebhookUser(review.User, "")),
		ActorName:     actorNameFromWebhookUser(payload.Sender, actorNameFromWebhookUser(review.User, "")),
		SummaryKey:    eventType,
		Details:       map[string]any{"number": number, "title": strings.TrimSpace(payload.PullRequest.Title), "state": strings.TrimSpace(review.State), "reviewID": reviewID},
		TargetKind:    "pull",
		TargetID:      numberText,
		TargetURL:     firstNonEmptyString(review.HTMLURL, payload.PullRequest.HTMLURL),
		OccurredAt:    occurred,
	}}
}

func mapAGSPullRequestReviewCommentEvent(repo activity.ResourceActivityDataPlaneRepository, action string, payload agsWebhookPayload, now time.Time) []activity.ResourceActivityDataPlaneEvent {
	comment := payload.Comment
	if comment.ID == 0 {
		return nil
	}
	number := payload.PullRequest.Number
	if number <= 0 {
		number = payload.Number
	}
	if number <= 0 {
		return nil
	}
	action = normalizeAction(action, payload.Action)
	eventTypeByAction := map[string]string{
		"created":    "pr_review_comment_created",
		"edited":     "pr_review_comment_edited",
		"deleted":    "pr_review_comment_deleted",
		"resolved":   "pr_review_comment_resolved",
		"unresolved": "pr_review_comment_unresolved",
	}
	eventType := eventTypeByAction[action]
	if eventType == "" {
		return nil
	}
	occurred := firstNonZeroTime(optionalTime(comment.UpdatedAt), optionalTime(comment.CreatedAt), now)
	id := strconv.FormatInt(comment.ID, 10)
	return []activity.ResourceActivityDataPlaneEvent{{
		SourceEventID: activity.ResourceActivitySourceEventID(repo, "pull", strconv.Itoa(number), "review_comment", id, action),
		EventType:     eventType,
		ActorType:     activity.ActivityActorTypeUser,
		ActorID:       actorIDFromWebhookUser(payload.Sender, actorIDFromWebhookUser(comment.User, "")),
		ActorName:     actorNameFromWebhookUser(payload.Sender, actorNameFromWebhookUser(comment.User, "")),
		SummaryKey:    eventType,
		Details:       map[string]any{"number": number, "commentID": id},
		TargetKind:    "pull_review_comment",
		TargetID:      id,
		TargetURL:     strings.TrimSpace(comment.HTMLURL),
		OccurredAt:    occurred,
	}}
}

func pushRefKindName(ref string) (string, string) {
	ref = strings.TrimSpace(ref)
	switch {
	case strings.HasPrefix(ref, "refs/heads/"):
		return "branch", strings.TrimPrefix(ref, "refs/heads/")
	case strings.HasPrefix(ref, "refs/tags/"):
		return "tag", strings.TrimPrefix(ref, "refs/tags/")
	default:
		return "", ""
	}
}

func commitActor(commit agsWebhookCommit, payload agsWebhookPayload) (string, string) {
	id := firstNonEmptyString(commit.Author.Email, commit.Committer.Email, actorIDFromWebhookUser(payload.Sender, payload.Pusher.Name))
	name := firstNonEmptyString(commit.Author.Name, commit.Committer.Name, actorNameFromWebhookUser(payload.Sender, payload.Pusher.Name))
	return id, name
}

func normalizeAction(values ...string) string {
	for _, value := range values {
		if trimmed := strings.TrimSpace(value); trimmed != "" {
			return trimmed
		}
	}
	return ""
}

func actorIDFromWebhookUser(user agsWebhookUser, fallback string) string {
	id := ""
	if user.ID > 0 {
		id = strconv.FormatInt(user.ID, 10)
	}
	return firstNonEmptyString(user.Login, user.UserName, id, fallback)
}

func actorNameFromWebhookUser(user agsWebhookUser, fallback string) string {
	return firstNonEmptyString(user.FullName, user.Name, user.Login, user.UserName, fallback)
}

func optionalTime(value *time.Time) time.Time {
	if value == nil || value.IsZero() {
		return time.Time{}
	}
	return value.UTC()
}

func firstNonZeroTime(values ...time.Time) time.Time {
	for _, value := range values {
		if !value.IsZero() {
			return value.UTC()
		}
	}
	return time.Now().UTC()
}

func issueOccurredAt(createdAt, updatedAt, closedAt *time.Time, action string, fallback time.Time) time.Time {
	switch action {
	case "opened":
		return firstNonZeroTime(optionalTime(createdAt), optionalTime(updatedAt), fallback)
	case "closed":
		return firstNonZeroTime(optionalTime(closedAt), optionalTime(updatedAt), fallback)
	default:
		return firstNonZeroTime(optionalTime(updatedAt), optionalTime(createdAt), fallback)
	}
}

func pullOccurredAt(pull agsWebhookPull, action string, fallback time.Time) time.Time {
	if action == "opened" {
		return firstNonZeroTime(optionalTime(pull.CreatedAt), optionalTime(pull.UpdatedAt), fallback)
	}
	if action == "closed" && pull.Merged {
		return firstNonZeroTime(optionalTime(pull.MergedAt), optionalTime(pull.ClosedAt), optionalTime(pull.UpdatedAt), fallback)
	}
	if action == "closed" {
		return firstNonZeroTime(optionalTime(pull.ClosedAt), optionalTime(pull.UpdatedAt), fallback)
	}
	return firstNonZeroTime(optionalTime(pull.UpdatedAt), optionalTime(pull.CreatedAt), fallback)
}

func shortSHA(value string) string {
	value = strings.TrimSpace(value)
	if len(value) <= 12 {
		return value
	}
	return value[:12]
}

func firstLine(value string) string {
	value = strings.TrimSpace(value)
	if idx := strings.IndexByte(value, '\n'); idx >= 0 {
		return strings.TrimSpace(value[:idx])
	}
	return value
}

func unixNanoString(value time.Time) string {
	if value.IsZero() {
		return "0"
	}
	return strconv.FormatInt(value.UTC().UnixNano(), 10)
}

func firstNonEmptyString(values ...string) string {
	for _, value := range values {
		if trimmed := strings.TrimSpace(value); trimmed != "" {
			return trimmed
		}
	}
	return ""
}