package app

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/core/identifiers"
	"github.com/tidbcloud/AgentTeam/internal/core/runeffect"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/app/webhooks"
	"github.com/tidbcloud/AgentTeam/internal/store/coordinator"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/serviceactor"
	"github.com/tidbcloud/AgentTeam/internal/store/storeerr"
)

const serviceEventDeliveryWebhookReclaimAfter = 5 * time.Minute

type serviceEventDeliveryOccurrence struct {
	WorkspaceID      string
	SourceType       string
	SourceSubjectKey string
	EventID          string
	EventAction      string
}

type serviceEventDeliveryDispatcher struct {
	store         *Store
	app           webhooks.AppDefinition
	event         serviceEventProviderWebhookEvent
	payload       []byte
	now           time.Time
	deliveryLogID string
	receiptID     string

	resultReady bool
	result      AppWebhookResult
	delivered   []serviceEventDeliveryCandidate
	deliveries  []message.MessageDelivery
}

func newServiceEventDeliveryDispatcher(store *Store, app webhooks.AppDefinition, event serviceEventProviderWebhookEvent, payload []byte) (*serviceEventDeliveryDispatcher, error) {
	deliveryLogID, err := identifiers.NewEntityID("appwd")
	if err != nil {
		return nil, fmt.Errorf("generate service event webhook delivery id: %w", err)
	}
	return &serviceEventDeliveryDispatcher{
		store:         store,
		app:           app,
		event:         event,
		payload:       payload,
		now:           time.Now().UTC(),
		deliveryLogID: deliveryLogID,
	}, nil
}

func (dispatcher *serviceEventDeliveryDispatcher) DispatchEvent(ctx context.Context) (AppWebhookResult, error) {
	if dispatcher == nil || dispatcher.store == nil {
		return AppWebhookResult{}, badRequest("service event delivery dispatcher is not configured")
	}
	occurrence := dispatcher.event.serviceEventOccurrence()
	if occurrence.SourceType == "" || !serviceEventDeliverySupportsPullRequestSourceType(occurrence.SourceType) {
		return AppWebhookResult{}, badRequest("service event source type is not supported")
	}
	candidates, err := dispatcher.matchAndReserveDeliveries(ctx, occurrence)
	if err != nil {
		return AppWebhookResult{}, err
	}
	if dispatcher.resultReady {
		return dispatcher.result, nil
	}
	groups, err := dispatcher.groupDeliveryCandidates(candidates)
	if err != nil {
		return AppWebhookResult{}, err
	}
	for _, group := range groups {
		if err := dispatcher.deliverCandidateGroup(ctx, occurrence, group); err != nil {
			for _, candidate := range group.Candidates {
				if failErr := dispatcher.failCandidateDelivery(ctx, candidate, "delivery_failed", err.Error()); failErr != nil {
					return AppWebhookResult{}, failErr
				}
			}
		}
	}
	if err := dispatcher.finalizeWebhookReceipt(ctx); err != nil {
		return AppWebhookResult{}, err
	}
	dispatcher.result = AppWebhookResult{
		MatchedSubscriptionCount: len(dispatcher.delivered),
		WorkspaceID:              dispatcher.event.Installation.WorkspaceID,
		Deliveries:               dispatcher.deliveries,
	}
	dispatcher.resultReady = true
	return dispatcher.result, nil
}

func serviceEventDeliverySupportsPullRequestSourceType(sourceType string) bool {
	for _, supported := range pullRequestSubscriptionSourceTypes() {
		if strings.TrimSpace(supported) == strings.TrimSpace(sourceType) {
			return true
		}
	}
	return false
}

func (dispatcher *serviceEventDeliveryDispatcher) matchAndReserveDeliveries(ctx context.Context, occurrence serviceEventDeliveryOccurrence) ([]serviceEventDeliveryCandidate, error) {
	matches := make([]serviceEventDeliveryCandidate, 0)
	if err := coordinator.New(dispatcher.store.db).Run(ctx, "service event delivery match", func(tx coredb.Tx) error {
		receipt, inserted, err := dispatcher.reserveWebhookReceipt(ctx, tx)
		if err != nil {
			return err
		}
		dispatcher.receiptID = receipt.ID
		if !inserted {
			reserved, duplicate, err := dispatcher.reserveExistingWebhookReceipt(ctx, tx, receipt)
			if err != nil {
				return err
			}
			if duplicate {
				dispatcher.result = AppWebhookResult{Duplicate: true, WorkspaceID: dispatcher.event.Installation.WorkspaceID}
				dispatcher.resultReady = true
				return nil
			}
			if !reserved {
				return conflict("service event webhook delivery is already processing; retry later")
			}
		}

		candidates, fanoutExceeded, err := dispatcher.loadCandidatesForOccurrence(ctx, tx, occurrence)
		if err != nil {
			return err
		}
		if fanoutExceeded {
			msg := fmt.Sprintf("service event webhook matched more than %d active subscriptions", serviceEventDeliveryCandidateMax)
			if err := dispatcher.completeWebhookReceipt(ctx, tx, nil, "too_many_matching_subscriptions", msg); err != nil {
				return err
			}
			dispatcher.result = AppWebhookResult{WorkspaceID: dispatcher.event.Installation.WorkspaceID}
			dispatcher.resultReady = true
			return nil
		}
		if len(candidates) == 0 {
			if err := dispatcher.completeWebhookReceipt(ctx, tx, nil, "no_matching_subscriptions", ""); err != nil {
				return err
			}
			dispatcher.result = AppWebhookResult{WorkspaceID: dispatcher.event.Installation.WorkspaceID}
			dispatcher.resultReady = true
			return nil
		}
		if err := updateServiceEventDeliveryBranchMetadataForCandidates(ctx, tx, candidates, dispatcher.event.Event, dispatcher.now); err != nil {
			return err
		}

		filtered := serviceEventDeliveryCandidatesMatchingEvent(candidates, dispatcher.event.Event)
		if len(filtered) == 0 {
			if err := dispatcher.completeWebhookReceipt(ctx, tx, nil, "filtered_by_subscription", "provider event filters did not match"); err != nil {
				return err
			}
			dispatcher.result = AppWebhookResult{WorkspaceID: dispatcher.event.Installation.WorkspaceID}
			dispatcher.resultReady = true
			return nil
		}
		filtered, err = dispatcher.filterCandidatesByCurrentPolicy(ctx, tx, filtered)
		if err != nil {
			return err
		}
		if len(filtered) == 0 {
			if err := dispatcher.completeWebhookReceipt(ctx, tx, nil, "filtered_by_policy", "service event delivery policy no longer allows these subscriptions"); err != nil {
				return err
			}
			dispatcher.result = AppWebhookResult{WorkspaceID: dispatcher.event.Installation.WorkspaceID}
			dispatcher.resultReady = true
			return nil
		}

		if err := recordPullRequestSubscriptionRateLimitEvent(ctx, tx, pullRequestSubscriptionRateLimitEvent{
			WorkspaceID:       dispatcher.event.Installation.WorkspaceID,
			AppInstallationID: dispatcher.event.Installation.InstallationID,
			ActorType:         "app",
			ActorID:           dispatcher.event.Installation.InstallationID,
			Action:            pullRequestSubscriptionRateLimitActionWebhookDelivery,
			ResourceKey:       occurrence.SourceSubjectKey,
			CorrelationID:     occurrence.EventID,
			CreatedAt:         dispatcher.now,
		}); err != nil {
			if errors.Is(err, storeerr.ErrConflict) {
				for _, candidate := range filtered {
					if failErr := dispatcher.reserveAndFailServiceEventDeliveryInTx(ctx, tx, candidate, "provider_event_rate_limited", "provider event webhook delivery rate limited"); failErr != nil {
						return failErr
					}
				}
				if logErr := dispatcher.completeWebhookReceipt(ctx, tx, filtered, "rate_limited", "service event webhook delivery rate limited"); logErr != nil {
					return logErr
				}
				dispatcher.result = AppWebhookResult{WorkspaceID: dispatcher.event.Installation.WorkspaceID}
				dispatcher.resultReady = true
				return nil
			}
			return err
		}

		for _, candidate := range filtered {
			delivery, duplicate, err := dispatcher.reserveServiceEventDelivery(ctx, tx, candidate)
			if err != nil {
				return err
			}
			if duplicate {
				dispatcher.delivered = append(dispatcher.delivered, candidate)
				continue
			}
			candidate.ServiceEventDeliveryID = delivery.ID
			matches = append(matches, candidate)
		}
		return nil
	}); err != nil {
		return nil, err
	}
	return matches, nil
}

func (dispatcher *serviceEventDeliveryDispatcher) reserveWebhookReceipt(ctx context.Context, tx coredb.Tx) (serviceactor.WebhookReceipt, bool, error) {
	txStore := dispatcher.store.serviceActorStore().InTx(tx)
	return txStore.ReserveWebhookReceipt(ctx, dispatcher.webhookReceiptReserve())
}

func (dispatcher *serviceEventDeliveryDispatcher) reserveExistingWebhookReceipt(ctx context.Context, tx coredb.Tx, receipt serviceactor.WebhookReceipt) (bool, bool, error) {
	if strings.TrimSpace(receipt.Status) != serviceactor.WebhookReceiptStatusReceived || receipt.ProcessedAt != nil {
		return false, true, nil
	}
	if receipt.UpdatedAt.After(dispatcher.now.Add(-serviceEventDeliveryWebhookReclaimAfter)) {
		return false, false, nil
	}
	if _, err := dispatcher.store.serviceActorStore().InTx(tx).ResetWebhookReceiptForProcessing(ctx, dispatcher.webhookReceiptReserve()); err != nil {
		return false, false, err
	}
	return true, false, nil
}

func (dispatcher *serviceEventDeliveryDispatcher) completeWebhookReceipt(ctx context.Context, tx coredb.Tx, candidates []serviceEventDeliveryCandidate, outcome string, detail string) error {
	return dispatcher.completeWebhookReceiptRecord(ctx, tx, len(candidates), outcome, detail)
}

func (dispatcher *serviceEventDeliveryDispatcher) webhookReceiptReserve() serviceactor.WebhookReceiptReserve {
	return serviceactor.WebhookReceiptReserve{
		ID:                       dispatcher.receiptID,
		WorkspaceID:              dispatcher.event.Installation.WorkspaceID,
		ServiceDefinitionID:      dispatcher.app.AppID,
		ServiceInstallationID:    dispatcher.event.Installation.InstallationID,
		ServiceConnectionID:      dispatcher.event.ServiceConnectionID,
		ServiceConnectionRef:     dispatcher.event.Installation.ExternalSubjectRef,
		ServiceWebhookEndpointID: dispatcher.event.ServiceWebhookEndpointID,
		ServiceKind:              serviceactor.KindApp,
		EndpointKind:             serviceactor.EndpointKindProviderWebhook,
		DeliveryKey:              dispatcher.event.DeliveryKey,
		EventType:                dispatcher.event.EventType,
		EventAction:              providerEventActionSnapshot(dispatcher.event.Event),
		EventID:                  dispatcher.event.Event.EventID,
		EventKey:                 buildServiceEventSourceOccurrenceRef(dispatcher.event),
		ExternalResourceType:     PullRequestSubscriptionResourcePullRequest,
		ExternalResourceKey:      dispatcher.event.Event.ResourceIdentifier,
		ExternalResourceRef:      dispatcher.event.Event.ResourceIdentifier,
		ExternalResourceURL:      dispatcher.event.Event.ResourceURL,
		PayloadExcerpt:           payloadExcerptForLog(dispatcher.payload),
		Payload:                  dispatcher.payload,
		NormalizedEvent:          dispatcher.normalizedWebhookEventPayload(),
		Now:                      dispatcher.now,
	}
}

func (dispatcher *serviceEventDeliveryDispatcher) loadCandidatesForOccurrence(ctx context.Context, tx coredb.Tx, occurrence serviceEventDeliveryOccurrence) ([]serviceEventDeliveryCandidate, bool, error) {
	if providerEventIsBaseBranchPush(dispatcher.event.Event) {
		return loadServiceEventDeliveryCandidatesForBaseBranchPush(
			ctx,
			tx,
			occurrence.WorkspaceID,
			dispatcher.event.Installation.InstallationID,
			dispatcher.event.Event.Provider,
			dispatcher.event.Installation.ExternalSubjectRef,
			dispatcher.event.Event.Extra["repository"],
			dispatcher.event.Event.Extra["branch"],
			dispatcher.event.Event.Extra["default_branch"],
		)
	}
	if providerEventIsCommitStatus(dispatcher.event.Event) {
		return loadServiceEventDeliveryCandidatesForCommitStatus(
			ctx,
			tx,
			occurrence.WorkspaceID,
			dispatcher.event.Installation.InstallationID,
			dispatcher.event.Event.Provider,
			dispatcher.event.Installation.ExternalSubjectRef,
			dispatcher.event.Event.Extra["repository"],
			dispatcher.event.Event.Extra["sha"],
		)
	}
	return loadServiceEventDeliveryCandidates(ctx, tx, occurrence.WorkspaceID, dispatcher.event.Installation.InstallationID, dispatcher.event.Event.Provider, occurrence.SourceSubjectKey)
}

func (dispatcher *serviceEventDeliveryDispatcher) webhookEventForCandidate(candidate serviceEventDeliveryCandidate) serviceEventProviderWebhookEvent {
	event := dispatcher.event
	if !providerEventUsesCandidatePullRequest(dispatcher.event.Event) {
		return event
	}
	candidateEvent := event.Event
	candidateEvent.ResourceIdentifier = strings.TrimSpace(candidate.Subscription.ResourceKey)
	if strings.TrimSpace(candidateEvent.ResourceURL) == "" || providerEventIsBaseBranchPush(candidateEvent) {
		candidateEvent.ResourceURL = strings.TrimSpace(candidate.Subscription.ResourceRef)
	}
	if candidateEvent.Extra == nil {
		candidateEvent.Extra = map[string]string{}
	} else {
		cloned := make(map[string]string, len(candidateEvent.Extra)+2)
		for key, value := range candidateEvent.Extra {
			cloned[key] = value
		}
		candidateEvent.Extra = cloned
	}
	if baseBranch := serviceEventDeliveryBranchMetadata(candidate.Subscription.Metadata, "base_branch", "target_branch"); baseBranch != "" {
		candidateEvent.Extra["base_branch"] = baseBranch
	}
	event.Event = candidateEvent
	return event
}

func (dispatcher *serviceEventDeliveryDispatcher) filterCandidatesByCurrentPolicy(ctx context.Context, tx coredb.Tx, candidates []serviceEventDeliveryCandidate) ([]serviceEventDeliveryCandidate, error) {
	filtered := make([]serviceEventDeliveryCandidate, 0, len(candidates))
	for _, candidate := range candidates {
		if strings.TrimSpace(candidate.Subscription.CreatedByActorType) == "human" {
			// Human-created subscriptions current-read the installation ACL:
			// a revoked grant must fence active watches immediately.
			err := requireHumanInstallationUse(ctx, tx, candidate.Subscription.WorkspaceID, candidate.Subscription.CreatedByMemberID, candidate.Subscription.AppInstallationID)
			if err == nil {
				filtered = append(filtered, candidate)
				continue
			}
			if failureErr := dispatcher.reserveAndFailServiceEventDeliveryInTx(ctx, tx, candidate, "app_installation_not_authorized", err.Error()); failureErr != nil {
				return nil, failureErr
			}
			if failureErr := markServiceEventSubscriptionDeliveryFailure(ctx, tx, candidate.Subscription.ID, "app_installation_not_authorized", err.Error(), dispatcher.now); failureErr != nil {
				return nil, failureErr
			}
			continue
		}
		if strings.TrimSpace(candidate.Subscription.CreatedByActorType) != "agent" {
			filtered = append(filtered, candidate)
			continue
		}
		_, err := checkAgentAppAuthorization(ctx, tx, candidate.Subscription.WorkspaceID, candidate.Subscription.CreatedByActorID, candidate.Subscription.AppInstallationID, []string{ServiceEventSubscriptionScopeWrite})
		if err == nil {
			filtered = append(filtered, candidate)
			continue
		}
		if failureErr := dispatcher.reserveAndFailServiceEventDeliveryInTx(ctx, tx, candidate, "app_not_authorized_for_agent", err.Error()); failureErr != nil {
			return nil, failureErr
		}
		if failureErr := markServiceEventSubscriptionDeliveryFailure(ctx, tx, candidate.Subscription.ID, "app_not_authorized_for_agent", err.Error(), dispatcher.now); failureErr != nil {
			return nil, failureErr
		}
	}
	return filtered, nil
}

type serviceEventDeliveryCandidateGroup struct {
	Key                     serviceEventDeliveryCandidateGroupKey
	Candidates              []serviceEventDeliveryCandidate
	Rendered                string
	NotifyPrincipalMemberID string
}

type serviceEventDeliveryCandidateGroupKey struct {
	TargetKind              string
	DeliveryScopeID         string
	Rendered                string
	NotifyPrincipalMemberID string
}

func (dispatcher *serviceEventDeliveryDispatcher) groupDeliveryCandidates(candidates []serviceEventDeliveryCandidate) ([]serviceEventDeliveryCandidateGroup, error) {
	byKey := make(map[serviceEventDeliveryCandidateGroupKey]int, len(candidates))
	groups := make([]serviceEventDeliveryCandidateGroup, 0, len(candidates))
	for _, candidate := range candidates {
		key, rendered, notifyPrincipalMemberID, err := dispatcher.deliveryCandidateGroupKey(candidate)
		if err != nil {
			return nil, err
		}
		if index, ok := byKey[key]; ok {
			groups[index].Candidates = append(groups[index].Candidates, candidate)
			continue
		}
		byKey[key] = len(groups)
		groups = append(groups, serviceEventDeliveryCandidateGroup{
			Key:                     key,
			Candidates:              []serviceEventDeliveryCandidate{candidate},
			Rendered:                rendered,
			NotifyPrincipalMemberID: notifyPrincipalMemberID,
		})
	}
	sort.Slice(groups, func(i, j int) bool {
		left := groups[i].Key
		right := groups[j].Key
		if left.DeliveryScopeID != right.DeliveryScopeID {
			return left.DeliveryScopeID < right.DeliveryScopeID
		}
		if left.TargetKind != right.TargetKind {
			return left.TargetKind < right.TargetKind
		}
		if left.Rendered != right.Rendered {
			return left.Rendered < right.Rendered
		}
		return left.NotifyPrincipalMemberID < right.NotifyPrincipalMemberID
	})
	for index := range groups {
		sort.Slice(groups[index].Candidates, func(i, j int) bool {
			return strings.TrimSpace(groups[index].Candidates[i].Subscription.ID) < strings.TrimSpace(groups[index].Candidates[j].Subscription.ID)
		})
	}
	return groups, nil
}

func (dispatcher *serviceEventDeliveryDispatcher) deliveryCandidateGroupKey(candidate serviceEventDeliveryCandidate) (serviceEventDeliveryCandidateGroupKey, string, string, error) {
	rendered := formatServiceEventDeliveryMessage(dispatcher.webhookEventForCandidate(candidate).Event)
	notifyPrincipalMemberID := ""
	if strings.TrimSpace(candidate.Subscription.DeliveryPolicy) == pullRequestSubscriptionDeliveryPolicy &&
		candidate.Subscription.NotifyPrincipal {
		principal, err := serviceEventDeliveryPrincipalForCandidate(candidate)
		if err != nil {
			return serviceEventDeliveryCandidateGroupKey{}, "", "", err
		}
		notifyPrincipalMemberID = strings.TrimSpace(principal.PrincipalMemberID)
	}
	key := serviceEventDeliveryCandidateGroupKey{
		TargetKind:              strings.TrimSpace(candidate.Binding.TargetKind),
		DeliveryScopeID:         strings.TrimSpace(candidate.Binding.DeliveryScopeID),
		Rendered:                rendered,
		NotifyPrincipalMemberID: notifyPrincipalMemberID,
	}
	return key, rendered, notifyPrincipalMemberID, nil
}

func (dispatcher *serviceEventDeliveryDispatcher) deliverCandidateGroup(ctx context.Context, occurrence serviceEventDeliveryOccurrence, group serviceEventDeliveryCandidateGroup) error {
	if len(group.Candidates) == 0 {
		return nil
	}
	firstCandidate := group.Candidates[0]
	var messageDelivery message.MessageDelivery
	if err := coordinator.New(dispatcher.store.db).Run(ctx, "service event delivery delivery", func(tx coredb.Tx) error {
		var effect runeffect.Effect
		if strings.TrimSpace(firstCandidate.Binding.TargetKind) == "task" {
			appMessage, taskDelivery, err := appendServiceEventTaskDiscussionTx(
				ctx,
				dispatcher.store,
				tx,
				dispatcher.event.Installation.WorkspaceID,
				firstCandidate.Binding,
				group.Rendered,
				serviceEventTaskDiscussionAppendOptions{
					AuthorMemberID:          firstCandidate.Binding.ServiceVisibleMemberID,
					AuthorTargetID:          strings.TrimSpace(dispatcher.event.Installation.Handle),
					AuthorDisplayName:       strings.TrimSpace(dispatcher.event.Installation.AppName),
					AuthorServiceSlug:       strings.TrimSpace(dispatcher.event.Installation.AppSlug),
					NotifyPrincipalMemberID: group.NotifyPrincipalMemberID,
					Source:                  serviceEventDeliveryContributionSource(firstCandidate, occurrence),
				},
				dispatcher.now,
			)
			if err != nil {
				return err
			}
			taskDelivery.AppMessage = &appMessage
			messageDelivery = taskDelivery
			effect = runeffect.MessageEffect(taskDelivery.MessageID)
		} else {
			target, err := loadServiceEventDeliveryTarget(ctx, tx, firstCandidate.Binding)
			if err != nil {
				return err
			}
			options := appendMessageOptions{
				Source:                  serviceEventDeliveryContributionSource(firstCandidate, occurrence),
				NotifyPrincipalMemberID: group.NotifyPrincipalMemberID,
			}
			delivery, _, err := dispatcher.store.appendAppServiceMessageWithOptions(
				ctx, tx, dispatcher.event.Installation.RuntimeAppInstallation,
				target, group.Rendered, nil, false, options,
				appServiceMessageProviderEvent,
				nil,
			)
			if err != nil {
				return err
			}
			messageDelivery = delivery
			effect = runeffect.MessageEffect(delivery.MessageID)
		}
		return dispatcher.completeServiceEventGroupInTx(ctx, tx, group.Candidates, effect)
	}); err != nil {
		return err
	}
	if strings.TrimSpace(messageDelivery.MessageID) != "" {
		dispatcher.deliveries = append(dispatcher.deliveries, messageDelivery)
	}
	dispatcher.delivered = append(dispatcher.delivered, group.Candidates...)
	return nil
}

type serviceEventDeliveryPrincipal struct {
	PrincipalMemberID string
}

func serviceEventDeliveryPrincipalForCandidate(candidate serviceEventDeliveryCandidate) (serviceEventDeliveryPrincipal, error) {
	appMemberID := strings.TrimSpace(candidate.Binding.ServiceVisibleMemberID)
	installationID := strings.TrimSpace(candidate.Binding.ServiceInstallationID)
	if appMemberID == "" || installationID == "" {
		return serviceEventDeliveryPrincipal{}, badRequest("service event app principal is invalid")
	}
	principalMemberID := strings.TrimSpace(candidate.Subscription.CreatedByMemberID)
	if principalMemberID == "" {
		return serviceEventDeliveryPrincipal{}, badRequest("service event human principal is invalid")
	}
	return serviceEventDeliveryPrincipal{PrincipalMemberID: principalMemberID}, nil
}

func serviceEventDeliveryContributionSource(candidate serviceEventDeliveryCandidate, occurrence serviceEventDeliveryOccurrence) message.ContributionSource {
	deliveryID := strings.TrimSpace(candidate.ServiceEventDeliveryID)
	if deliveryID == "" {
		return message.ContributionSource{}
	}
	return message.ContributionSourceForServiceEventDelivery(message.ServiceEventDeliveryContributionSourceParams{
		DeliveryID:  deliveryID,
		Label:       pullRequestSubscriptionName(candidate.Subscription.Provider, candidate.Subscription.ResourceKey),
		EventID:     occurrence.EventID,
		EventType:   occurrence.SourceType,
		EventAction: occurrence.EventAction,
	})
}

func (dispatcher *serviceEventDeliveryDispatcher) completeServiceEventGroupInTx(ctx context.Context, tx coredb.Tx, candidates []serviceEventDeliveryCandidate, effect runeffect.Effect) error {
	subscriptionIDs := make([]string, 0, len(candidates))
	for _, candidate := range candidates {
		if err := dispatcher.completeServiceEventDeliveryInTx(ctx, tx, candidate, effect); err != nil {
			return err
		}
		subscriptionIDs = append(subscriptionIDs, candidate.Subscription.ID)
	}
	if err := markServiceEventSubscriptionDelivered(ctx, tx, subscriptionIDs, dispatcher.now); err != nil {
		return err
	}
	if !serviceEventProviderEventCompletesSubscription(dispatcher.event.Event) {
		return nil
	}
	for _, candidate := range candidates {
		if err := completeServiceEventSubscriptionAfterTerminalEvent(ctx, tx, candidate.Subscription.WorkspaceID, candidate.Subscription.ID, dispatcher.now); err != nil {
			return err
		}
	}
	return nil
}

func (dispatcher *serviceEventDeliveryDispatcher) failCandidateDelivery(ctx context.Context, candidate serviceEventDeliveryCandidate, code string, detail string) error {
	return coordinator.New(dispatcher.store.db).Run(ctx, "service event delivery delivery fail", func(tx coredb.Tx) error {
		if err := dispatcher.failServiceEventDeliveryInTx(ctx, tx, candidate, code, detail); err != nil {
			return err
		}
		return markServiceEventSubscriptionDeliveryFailure(ctx, tx, candidate.Subscription.ID, code, detail, dispatcher.now)
	})
}

func (dispatcher *serviceEventDeliveryDispatcher) finalizeWebhookReceipt(ctx context.Context) error {
	outcome := "processed"
	errorMessage := ""
	if len(dispatcher.delivered) == 0 {
		outcome = "delivery_failed"
		errorMessage = "no active subscription could be delivered"
	}
	return coordinator.New(dispatcher.store.db).Run(ctx, "service event webhook receipt finalize", func(tx coredb.Tx) error {
		return dispatcher.completeWebhookReceipt(ctx, tx, dispatcher.delivered, outcome, errorMessage)
	})
}

func (dispatcher *serviceEventDeliveryDispatcher) completeWebhookReceiptRecord(ctx context.Context, tx coredb.Tx, matchedSubscriptionCount int, outcome string, detail string) error {
	return dispatcher.store.serviceActorStore().InTx(tx).CompleteWebhookReceipt(ctx, serviceactor.WebhookReceiptOutcome{
		ID:                       dispatcher.receiptID,
		MatchedSubscriptionCount: matchedSubscriptionCount,
		Status:                   serviceEventWebhookReceiptStatus(outcome),
		ErrorCode:                serviceEventWebhookReceiptErrorCode(outcome),
		ErrorDetail:              detail,
		NormalizedEvent:          dispatcher.normalizedWebhookEventPayload(),
		Now:                      dispatcher.now,
	})
}

func (dispatcher *serviceEventDeliveryDispatcher) normalizedWebhookEventPayload() []byte {
	payload, err := serviceEventDeliveryPayload(dispatcher.event)
	if err != nil {
		return nil
	}
	return payload
}

func serviceEventWebhookReceiptStatus(outcome string) string {
	trimmed := strings.TrimSpace(outcome)
	if trimmed == "" || trimmed == "processed" {
		return serviceactor.WebhookReceiptStatusProcessed
	}
	return trimmed
}

func serviceEventWebhookReceiptErrorCode(outcome string) string {
	trimmed := strings.TrimSpace(outcome)
	if trimmed == "" || trimmed == "processed" {
		return ""
	}
	return trimmed
}

func (event serviceEventProviderWebhookEvent) serviceEventOccurrence() serviceEventDeliveryOccurrence {
	return serviceEventDeliveryOccurrence{
		WorkspaceID:      strings.TrimSpace(event.Installation.WorkspaceID),
		SourceType:       pullRequestSubscriptionSourceType(event.Event.Provider, PullRequestSubscriptionResourcePullRequest),
		SourceSubjectKey: strings.TrimSpace(event.Event.ResourceIdentifier),
		EventID:          strings.TrimSpace(event.Event.EventID),
		EventAction:      providerEventActionSnapshot(event.Event),
	}
}