package provider

import (
	"bytes"
	"context"
	"embed"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"slices"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"

	providerevents "github.com/codex-k8s/kodex/libs/go/platformevents/provider"
	postgreslib "github.com/codex-k8s/kodex/libs/go/postgres"
	"github.com/codex-k8s/kodex/services/internal/provider-hub/internal/domain/errs"
	providerrepo "github.com/codex-k8s/kodex/services/internal/provider-hub/internal/domain/repository/provider"
	"github.com/codex-k8s/kodex/services/internal/provider-hub/internal/domain/types/entity"
	"github.com/codex-k8s/kodex/services/internal/provider-hub/internal/domain/types/enum"
	"github.com/codex-k8s/kodex/services/internal/provider-hub/internal/domain/types/query"
)

// SQLFiles contains named SQL queries for the provider-hub PostgreSQL repository.
//
//go:embed sql/*.sql
var SQLFiles embed.FS

var _ providerrepo.Repository = (*Repository)(nil)

type database interface {
	execer
	queryer
	BeginTx(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error)
}

type execer interface {
	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
}

type queryer interface {
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
}

// Repository stores provider-hub state in PostgreSQL.
type Repository struct {
	db database
}

// NewRepository creates a PostgreSQL repository.
func NewRepository(pool *pgxpool.Pool) *Repository {
	if pool == nil {
		panic("provider-hub postgres pool is required")
	}
	return &Repository{db: pool}
}

// Ping verifies connectivity to provider-hub storage.
func (r *Repository) Ping(ctx context.Context) error {
	pool, ok := r.db.(*pgxpool.Pool)
	if !ok {
		return nil
	}
	if err := pool.Ping(ctx); err != nil {
		return fmt.Errorf("ping provider-hub postgres: %w", err)
	}
	return nil
}

// StoreWebhookEvent stores a webhook inbox record, projections, normalized provider events and outbox events atomically.
func (r *Repository) StoreWebhookEvent(ctx context.Context, webhook entity.WebhookEvent, projectionUpdate providerrepo.ProjectionUpdate, providerEvents []entity.ProviderEvent, outboxEvents []entity.OutboxEvent) (entity.WebhookEvent, []entity.ProviderEvent, error) {
	var stored entity.WebhookEvent
	var storedProviderEvents []entity.ProviderEvent
	err := postgreslib.WithTx(ctx, r.db, func(tx pgx.Tx) error {
		var insertErr error
		stored, insertErr = queryOne(ctx, tx, operationStoreWebhookEvent, queryWebhookEventInsert, webhookEventArgs(webhook), scanWebhookEvent)
		if errors.Is(insertErr, errs.ErrNotFound) {
			replayed, replayErr := queryOne(ctx, tx, operationStoreWebhookEvent, queryWebhookEventGetByDelivery, webhookEventIdentityArgs(webhook), scanWebhookEvent)
			if replayErr != nil {
				return replayErr
			}
			if !sameWebhookEvent(webhook, replayed) {
				return errs.ErrConflict
			}
			stored = replayed
			events, _, eventErr := queryPage(ctx, tx, operationListProviderEvents, queryProviderEventList, providerEventFilterArgs(query.ProviderEventFilter{
				SourceWebhookEventID: &stored.ID,
			}), scanProviderEvent)
			storedProviderEvents = events
			return eventErr
		}
		if insertErr != nil {
			return insertErr
		}
		projectionResult, err := applyProjectionUpdate(ctx, tx, operationStoreWebhookEvent, projectionUpdate)
		if err != nil {
			return err
		}
		filteredProviderEvents := filterProviderEvents(providerEvents, projectionUpdate, projectionResult)
		insertedProviderEvents, err := insertProviderEvents(ctx, tx, operationStoreWebhookEvent, filteredProviderEvents)
		if err != nil {
			return err
		}
		filteredOutboxEvents := filterOutboxEvents(outboxEvents, filteredProviderEvents, projectionResult)
		if err := insertOutboxEvents(ctx, tx, operationStoreWebhookEvent, filteredOutboxEvents); err != nil {
			return err
		}
		storedProviderEvents = insertedProviderEvents
		return nil
	})
	if err != nil {
		return entity.WebhookEvent{}, nil, wrapError(operationStoreWebhookEvent, err)
	}
	return stored, storedProviderEvents, nil
}

// ProcessWebhookEvent updates processing state and stores projection changes atomically.
func (r *Repository) ProcessWebhookEvent(ctx context.Context, webhook entity.WebhookEvent, projectionUpdate providerrepo.ProjectionUpdate, providerEvents []entity.ProviderEvent, outboxEvents []entity.OutboxEvent) (entity.WebhookEvent, error) {
	var stored entity.WebhookEvent
	err := postgreslib.WithTx(ctx, r.db, func(tx pgx.Tx) error {
		var updateErr error
		stored, updateErr = queryOne(ctx, tx, operationProcessWebhookEvent, queryWebhookEventUpdateProcessing, webhookEventProcessingArgs(webhook), scanWebhookEvent)
		if updateErr != nil {
			return updateErr
		}
		projectionResult, err := applyProjectionUpdate(ctx, tx, operationProcessWebhookEvent, projectionUpdate)
		if err != nil {
			return err
		}
		filteredProviderEvents := filterProviderEvents(providerEvents, projectionUpdate, projectionResult)
		if _, err := insertProviderEvents(ctx, tx, operationProcessWebhookEvent, filteredProviderEvents); err != nil {
			return err
		}
		filteredOutboxEvents := filterOutboxEvents(outboxEvents, filteredProviderEvents, projectionResult)
		return insertOutboxEvents(ctx, tx, operationProcessWebhookEvent, filteredOutboxEvents)
	})
	if err != nil {
		return entity.WebhookEvent{}, wrapError(operationProcessWebhookEvent, err)
	}
	return stored, nil
}

// GetWorkItemProjection returns one Issue or PR/MR projection.
func (r *Repository) GetWorkItemProjection(ctx context.Context, lookup query.ProviderTargetLookup) (entity.ProviderWorkItemProjection, error) {
	return queryOne(ctx, r.db, operationGetWorkItemProjection, queryWorkItemProjectionGet, workItemProjectionLookupArgs(lookup), scanWorkItemProjection)
}

// GetCommentProjectionByProviderID returns one normalized comment by provider id within one work item projection.
func (r *Repository) GetCommentProjectionByProviderID(ctx context.Context, workItemProjectionID uuid.UUID, providerCommentID string) (entity.ProviderCommentProjection, error) {
	args := commentProjectionLookupArgs(workItemProjectionID, providerCommentID)
	return queryOne(ctx, r.db, operationGetCommentProjectionByProviderID, queryCommentProjectionGetByProviderID, args, scanCommentProjection)
}

// ListWorkItemProjections returns Issue and PR/MR projections.
func (r *Repository) ListWorkItemProjections(ctx context.Context, filter query.WorkItemProjectionFilter) ([]entity.ProviderWorkItemProjection, query.PageResult, error) {
	return queryPage(ctx, r.db, operationListWorkItemProjections, queryWorkItemProjectionList, workItemProjectionFilterArgs(filter), scanWorkItemProjection)
}

// ListComments returns comment projections for one work item.
func (r *Repository) ListComments(ctx context.Context, filter query.CommentProjectionFilter) ([]entity.ProviderCommentProjection, query.PageResult, error) {
	return queryPage(ctx, r.db, operationListComments, queryCommentProjectionList, commentProjectionFilterArgs(filter), scanCommentProjection)
}

// ListRelationships returns normalized relationships.
func (r *Repository) ListRelationships(ctx context.Context, filter query.RelationshipFilter) ([]entity.ProviderRelationship, query.PageResult, error) {
	return queryPage(ctx, r.db, operationListRelationships, queryRelationshipList, relationshipFilterArgs(filter), scanRelationship)
}

// GetRelationshipByIdentity returns one relationship by its natural identity.
func (r *Repository) GetRelationshipByIdentity(ctx context.Context, lookup query.RelationshipLookup) (entity.ProviderRelationship, error) {
	return queryOne(ctx, r.db, operationGetRelationshipByIdentity, queryRelationshipGetByIdentity, relationshipLookupArgs(lookup), scanRelationship)
}

// RegisterProviderArtifactSignal records signal identity and enqueues cursors atomically.
func (r *Repository) RegisterProviderArtifactSignal(ctx context.Context, signal entity.ProviderArtifactSignal, request entity.ReconciliationRequest, cursors []entity.SyncCursor) ([]entity.SyncCursor, error) {
	var stored []entity.SyncCursor
	err := postgreslib.WithTx(ctx, r.db, func(tx pgx.Tx) error {
		if _, err := storeProviderArtifactSignal(ctx, tx, operationRegisterProviderArtifactSignal, signal); err != nil {
			return err
		}
		enqueued, err := enqueueSyncCursors(ctx, tx, operationRegisterProviderArtifactSignal, request, cursors)
		if err != nil {
			return err
		}
		stored = enqueued
		return nil
	})
	if err != nil {
		return nil, wrapError(operationRegisterProviderArtifactSignal, err)
	}
	return stored, nil
}

// EnqueueSyncCursors records one idempotent enqueue request and creates all requested cursors atomically.
func (r *Repository) EnqueueSyncCursors(ctx context.Context, request entity.ReconciliationRequest, cursors []entity.SyncCursor) ([]entity.SyncCursor, error) {
	var stored []entity.SyncCursor
	err := postgreslib.WithTx(ctx, r.db, func(tx pgx.Tx) error {
		var err error
		stored, err = enqueueSyncCursors(ctx, tx, operationEnqueueSyncCursors, request, cursors)
		return err
	})
	if err != nil {
		return nil, wrapError(operationEnqueueSyncCursors, err)
	}
	return stored, nil
}

// GetSyncCursor returns one reconciliation cursor by id.
func (r *Repository) GetSyncCursor(ctx context.Context, id uuid.UUID) (entity.SyncCursor, error) {
	return queryOne(ctx, r.db, operationGetSyncCursor, querySyncCursorGet, pgx.NamedArgs{"id": id}, scanSyncCursor)
}

// ListSyncCursors returns reconciliation cursors by supported filters.
func (r *Repository) ListSyncCursors(ctx context.Context, filter query.SyncCursorFilter) ([]entity.SyncCursor, query.PageResult, error) {
	return queryPage(ctx, r.db, operationListSyncCursors, querySyncCursorList, syncCursorFilterArgs(filter), scanSyncCursor)
}

// ClaimSyncCursor leases one due reconciliation cursor for a worker.
func (r *Repository) ClaimSyncCursor(ctx context.Context, claim providerrepo.SyncCursorClaim) (entity.SyncCursor, error) {
	return queryOne(ctx, r.db, operationClaimSyncCursor, querySyncCursorClaim, syncCursorClaimArgs(claim), scanSyncCursor)
}

// ApplyReconciliationBatch stores provider snapshots and final cursor state atomically.
func (r *Repository) ApplyReconciliationBatch(ctx context.Context, completion providerrepo.ReconciliationBatchCompletion) (entity.SyncCursor, []entity.ProviderEvent, error) {
	var cursor entity.SyncCursor
	var storedProviderEvents []entity.ProviderEvent
	err := postgreslib.WithTx(ctx, r.db, func(tx pgx.Tx) error {
		projectionResult, err := applyProjectionUpdate(ctx, tx, operationApplyReconciliationBatch, completion.ProjectionUpdate)
		if err != nil {
			return err
		}
		filteredProviderEvents := filterProviderEvents(completion.ProviderEvents, completion.ProjectionUpdate, projectionResult)
		storedEvents, err := insertProviderEvents(ctx, tx, operationApplyReconciliationBatch, filteredProviderEvents)
		if err != nil {
			return err
		}
		filteredOutboxEvents := filterOutboxEvents(completion.OutboxEvents, filteredProviderEvents, projectionResult)
		if err := insertOutboxEvents(ctx, tx, operationApplyReconciliationBatch, filteredOutboxEvents); err != nil {
			return err
		}
		for _, snapshot := range completion.LimitSnapshots {
			if _, err := queryOne(ctx, tx, operationRecordLimitSnapshot, queryLimitSnapshotUpsert, limitSnapshotArgs(snapshot), scanLimitSnapshot); err != nil {
				return err
			}
		}
		if completion.RuntimeState != nil {
			if _, err := queryOne(ctx, tx, operationUpsertAccountRuntimeState, queryAccountRuntimeStateUpsertFromSnapshot, accountRuntimeStateArgs(*completion.RuntimeState), scanAccountRuntimeState); err != nil {
				return err
			}
		}
		updatedCursor, err := queryOne(ctx, tx, operationApplyReconciliationBatch, querySyncCursorComplete, syncCursorCompletionArgs(completion), scanSyncCursor)
		if err != nil {
			return err
		}
		cursor = updatedCursor
		storedProviderEvents = storedEvents
		return nil
	})
	if err != nil {
		return entity.SyncCursor{}, nil, wrapError(operationApplyReconciliationBatch, err)
	}
	return cursor, storedProviderEvents, nil
}

// GetWebhookEvent returns a stored webhook inbox record by id.
func (r *Repository) GetWebhookEvent(ctx context.Context, id uuid.UUID) (entity.WebhookEvent, error) {
	return queryOne(ctx, r.db, operationGetWebhookEvent, queryWebhookEventGet, pgx.NamedArgs{"id": id}, scanWebhookEvent)
}

// ListWebhookEvents returns stored webhook inbox records.
func (r *Repository) ListWebhookEvents(ctx context.Context, filter query.WebhookEventFilter) ([]entity.WebhookEvent, query.PageResult, error) {
	return queryPage(ctx, r.db, operationListWebhookEvents, queryWebhookEventList, webhookEventFilterArgs(filter), scanWebhookEvent)
}

// CleanupExpiredWebhookPayloads removes full provider payload from expired retryable inbox rows.
func (r *Repository) CleanupExpiredWebhookPayloads(ctx context.Context, now time.Time, limit int32) ([]entity.WebhookEvent, error) {
	webhooks, err := postgreslib.QueryRows(ctx, r.db, queryWebhookEventCleanupExpiredPayloads, pgx.NamedArgs{
		"now":   now,
		"limit": limit,
	}, scanWebhookEvent)
	if err != nil {
		return nil, wrapError(operationCleanupExpiredWebhookPayloads, err)
	}
	return webhooks, nil
}

// ListProviderEvents returns normalized provider events.
func (r *Repository) ListProviderEvents(ctx context.Context, filter query.ProviderEventFilter) ([]entity.ProviderEvent, query.PageResult, error) {
	return queryPage(ctx, r.db, operationListProviderEvents, queryProviderEventList, providerEventFilterArgs(filter), scanProviderEvent)
}

// UpsertAccountRuntimeState creates or updates provider-side account state.
func (r *Repository) UpsertAccountRuntimeState(ctx context.Context, state entity.ProviderAccountRuntimeState) (entity.ProviderAccountRuntimeState, error) {
	return queryOne(ctx, r.db, operationUpsertAccountRuntimeState, queryAccountRuntimeStateUpsert, accountRuntimeStateArgs(state), scanAccountRuntimeState)
}

// GetAccountRuntimeState returns provider-side account state.
func (r *Repository) GetAccountRuntimeState(ctx context.Context, lookup query.AccountRuntimeStateLookup) (entity.ProviderAccountRuntimeState, error) {
	return queryOne(ctx, r.db, operationGetAccountRuntimeState, queryAccountRuntimeStateGet, accountRuntimeStateLookupArgs(lookup), scanAccountRuntimeState)
}

// ListAccountRuntimeStates returns provider-side account states.
func (r *Repository) ListAccountRuntimeStates(ctx context.Context, filter query.AccountRuntimeStateFilter) ([]entity.ProviderAccountRuntimeState, query.PageResult, error) {
	return queryPage(ctx, r.db, operationListAccountRuntimeStates, queryAccountRuntimeStateList, accountRuntimeStateFilterArgs(filter), scanAccountRuntimeState)
}

// RecordLimitSnapshot stores a provider limit snapshot and updates account runtime state atomically.
func (r *Repository) RecordLimitSnapshot(ctx context.Context, snapshot entity.ProviderLimitSnapshot, state entity.ProviderAccountRuntimeState) (entity.ProviderLimitSnapshot, error) {
	var stored entity.ProviderLimitSnapshot
	err := postgreslib.WithTx(ctx, r.db, func(tx pgx.Tx) error {
		var recordErr error
		stored, recordErr = queryOne(ctx, tx, operationRecordLimitSnapshot, queryLimitSnapshotUpsert, limitSnapshotArgs(snapshot), scanLimitSnapshot)
		if errors.Is(recordErr, errs.ErrNotFound) {
			stored, recordErr = queryOne(ctx, tx, operationRecordLimitSnapshot, queryLimitSnapshotGetReplay, limitSnapshotArgs(snapshot), scanLimitSnapshot)
			if errors.Is(recordErr, errs.ErrNotFound) {
				return errs.ErrConflict
			}
			return recordErr
		}
		if recordErr != nil {
			return recordErr
		}
		_, err := queryOne(ctx, tx, operationUpsertAccountRuntimeState, queryAccountRuntimeStateUpsertFromSnapshot, accountRuntimeStateArgs(state), scanAccountRuntimeState)
		return err
	})
	if err != nil {
		return entity.ProviderLimitSnapshot{}, wrapError(operationRecordLimitSnapshot, err)
	}
	return stored, nil
}

// ListLimitSnapshots returns provider limit snapshots.
func (r *Repository) ListLimitSnapshots(ctx context.Context, filter query.LimitSnapshotFilter) ([]entity.ProviderLimitSnapshot, query.PageResult, error) {
	return queryPage(ctx, r.db, operationListLimitSnapshots, queryLimitSnapshotList, limitSnapshotFilterArgs(filter), scanLimitSnapshot)
}

// RecordProviderOperation stores a provider operation audit record and reports whether this call acquired a new row.
func (r *Repository) RecordProviderOperation(ctx context.Context, operation entity.ProviderOperation) (entity.ProviderOperation, bool, error) {
	stored, replay, err := recordProviderOperation(ctx, r.db, operationRecordProviderOperation, operation)
	return stored, !replay && err == nil, err
}

// ApplyProviderOperation stores one finalized provider operation and its outbox side effects atomically.
func (r *Repository) ApplyProviderOperation(ctx context.Context, completion providerrepo.ProviderOperationCompletion) (entity.ProviderOperation, error) {
	var stored entity.ProviderOperation
	err := postgreslib.WithTx(ctx, r.db, func(tx pgx.Tx) error {
		var err error
		stored, err = queryOne(ctx, tx, operationApplyProviderOperation, queryProviderOperationComplete, providerOperationArgs(completion.Operation), scanProviderOperation)
		if errors.Is(err, errs.ErrNotFound) {
			var replay bool
			stored, replay, err = recordProviderOperation(ctx, tx, operationApplyProviderOperation, completion.Operation)
			if err != nil {
				return err
			}
			if replay {
				return nil
			}
		} else if err != nil {
			return err
		}
		projectionResult, err := applyProjectionUpdate(ctx, tx, operationApplyProviderOperation, completion.ProjectionUpdate)
		if err != nil {
			return err
		}
		filteredProviderEvents := filterProviderEvents(completion.ProviderEvents, completion.ProjectionUpdate, projectionResult)
		if _, err := insertProviderEvents(ctx, tx, operationApplyProviderOperation, filteredProviderEvents); err != nil {
			return err
		}
		filteredOutboxEvents := filterOutboxEvents(completion.OutboxEvents, filteredProviderEvents, projectionResult)
		return insertOutboxEvents(ctx, tx, operationApplyProviderOperation, filteredOutboxEvents)
	})
	if err != nil {
		return entity.ProviderOperation{}, wrapError(operationApplyProviderOperation, err)
	}
	return stored, nil
}

// ListProviderOperations returns provider operation audit records.
func (r *Repository) ListProviderOperations(ctx context.Context, filter query.ProviderOperationFilter) ([]entity.ProviderOperation, query.PageResult, error) {
	return queryPage(ctx, r.db, operationListProviderOperations, queryProviderOperationList, providerOperationFilterArgs(filter), scanProviderOperation)
}

// GetProviderOperationByCommand returns an already recorded command result.
func (r *Repository) GetProviderOperationByCommand(ctx context.Context, operationType enum.ProviderOperationType, commandID string) (entity.ProviderOperation, error) {
	return queryOne(ctx, r.db, operationGetProviderOperationByCommand, queryProviderOperationGetByCommand, pgx.NamedArgs{
		"operation_type": string(operationType),
		"command_id":     commandID,
	}, scanProviderOperation)
}

// GetRepositoryMergeSignal returns one safe provider-owned merge signal.
func (r *Repository) GetRepositoryMergeSignal(ctx context.Context, lookup query.RepositoryMergeSignalLookup) (entity.RepositoryMergeSignal, error) {
	return queryOne(ctx, r.db, operationGetRepositoryMergeSignal, queryRepositoryMergeSignalGet, repositoryMergeSignalLookupArgs(lookup), scanRepositoryMergeSignal)
}

// ListRepositoryMergeSignals returns safe provider-owned merge signals by supported filters.
func (r *Repository) ListRepositoryMergeSignals(ctx context.Context, filter query.RepositoryMergeSignalFilter) ([]entity.RepositoryMergeSignal, query.PageResult, error) {
	return queryPage(ctx, r.db, operationListRepositoryMergeSignals, queryRepositoryMergeSignalList, repositoryMergeSignalFilterArgs(filter), scanRepositoryMergeSignal)
}

// GetRepositoryChangeSignal returns one safe provider-owned repository change signal.
func (r *Repository) GetRepositoryChangeSignal(ctx context.Context, lookup query.RepositoryChangeSignalLookup) (entity.RepositoryChangeSignal, error) {
	return queryOne(ctx, r.db, operationGetRepositoryChangeSignal, queryRepositoryChangeSignalGet, repositoryChangeSignalLookupArgs(lookup), scanRepositoryChangeSignal)
}

// ListRepositoryChangeSignals returns safe provider-owned repository change signals by supported filters.
func (r *Repository) ListRepositoryChangeSignals(ctx context.Context, filter query.RepositoryChangeSignalFilter) ([]entity.RepositoryChangeSignal, query.PageResult, error) {
	return queryPage(ctx, r.db, operationListRepositoryChangeSignals, queryRepositoryChangeSignalList, repositoryChangeSignalFilterArgs(filter), scanRepositoryChangeSignal)
}

// GetRepositoryAdoptionScan returns one safe provider-owned adoption scan snapshot.
func (r *Repository) GetRepositoryAdoptionScan(ctx context.Context, lookup query.RepositoryAdoptionScanLookup) (entity.RepositoryAdoptionScanSnapshot, error) {
	return queryOne(ctx, r.db, operationGetRepositoryAdoptionScan, queryRepositoryAdoptionScanGet, repositoryAdoptionScanLookupArgs(lookup), scanRepositoryAdoptionScan)
}

// ListRepositoryAdoptionScans returns safe provider-owned adoption scan snapshots by supported filters.
func (r *Repository) ListRepositoryAdoptionScans(ctx context.Context, filter query.RepositoryAdoptionScanFilter) ([]entity.RepositoryAdoptionScanSnapshot, query.PageResult, error) {
	return queryPage(ctx, r.db, operationListRepositoryAdoptionScans, queryRepositoryAdoptionScanList, repositoryAdoptionScanFilterArgs(filter), scanRepositoryAdoptionScan)
}

// GetRepositoryAdoptionScanByOperation returns the safe adoption scan snapshot produced by one operation.
func (r *Repository) GetRepositoryAdoptionScanByOperation(ctx context.Context, providerOperationID uuid.UUID) (entity.RepositoryAdoptionScanSnapshot, error) {
	return queryOne(ctx, r.db, operationGetRepositoryAdoptionScanByOperation, queryRepositoryAdoptionScanGetByOperation, repositoryAdoptionScanOperationArgs(providerOperationID), scanRepositoryAdoptionScan)
}

// ClaimOutboxEvents leases unpublished outbox events for delivery.
func (r *Repository) ClaimOutboxEvents(ctx context.Context, limit int, now time.Time, lockedUntil time.Time) ([]entity.OutboxEvent, error) {
	events, ok, err := postgreslib.ClaimOutboxRows(ctx, r.db, queryOutboxEventClaim, limit, now, lockedUntil, scanOutboxEvent)
	if !ok {
		return nil, wrapError(operationClaimOutboxEvents, errs.ErrInvalidArgument)
	}
	return events, wrapError(operationClaimOutboxEvents, err)
}

// MarkOutboxEventPublished marks a leased outbox event as published.
func (r *Repository) MarkOutboxEventPublished(ctx context.Context, id uuid.UUID, attemptCount int, publishedAt time.Time) error {
	err := postgreslib.ApplyOutboxPublished(ctx, r.db, queryOutboxEventMarkPublished, errs.ErrInvalidArgument, id, attemptCount, publishedAt)
	return wrapError(operationMarkOutboxEventPublished, err)
}

// MarkOutboxEventFailed schedules a leased outbox event for retry.
func (r *Repository) MarkOutboxEventFailed(ctx context.Context, id uuid.UUID, attemptCount int, nextAttemptAt time.Time, lastError string) error {
	return r.markOutboxEventDeliveryFailure(ctx, operationMarkOutboxEventFailed, queryOutboxEventMarkFailed, id, attemptCount, "next_attempt_at", nextAttemptAt, lastError)
}

// MarkOutboxEventPermanentlyFailed moves a leased outbox event to terminal failure.
func (r *Repository) MarkOutboxEventPermanentlyFailed(ctx context.Context, id uuid.UUID, attemptCount int, failedAt time.Time, lastError string) error {
	return r.markOutboxEventDeliveryFailure(ctx, operationMarkOutboxEventPermanentlyFailed, queryOutboxEventMarkPermanentlyFailed, id, attemptCount, "failed_permanently_at", failedAt, lastError)
}

func (r *Repository) markOutboxEventDeliveryFailure(ctx context.Context, operation string, queryText string, id uuid.UUID, attemptCount int, timestampName string, timestampValue time.Time, lastError string) error {
	err := postgreslib.ApplyOutboxDeliveryFailure(ctx, r.db, queryText, errs.ErrInvalidArgument, id, attemptCount, timestampName, timestampValue, lastError)
	return wrapError(operation, err)
}

const (
	operationStoreWebhookEvent                    = "domain.Repository.StoreWebhookEvent"
	operationProcessWebhookEvent                  = "domain.Repository.ProcessWebhookEvent"
	operationGetWebhookEvent                      = "domain.Repository.GetWebhookEvent"
	operationListWebhookEvents                    = "domain.Repository.ListWebhookEvents"
	operationCleanupExpiredWebhookPayloads        = "domain.Repository.CleanupExpiredWebhookPayloads"
	operationListProviderEvents                   = "domain.Repository.ListProviderEvents"
	operationGetWorkItemProjection                = "domain.Repository.GetWorkItemProjection"
	operationGetCommentProjectionByProviderID     = "domain.Repository.GetCommentProjectionByProviderID"
	operationListWorkItemProjections              = "domain.Repository.ListWorkItemProjections"
	operationListComments                         = "domain.Repository.ListComments"
	operationGetRelationshipByIdentity            = "domain.Repository.GetRelationshipByIdentity"
	operationListRelationships                    = "domain.Repository.ListRelationships"
	operationRegisterProviderArtifactSignal       = "domain.Repository.RegisterProviderArtifactSignal"
	operationEnqueueSyncCursors                   = "domain.Repository.EnqueueSyncCursors"
	operationGetSyncCursor                        = "domain.Repository.GetSyncCursor"
	operationListSyncCursors                      = "domain.Repository.ListSyncCursors"
	operationClaimSyncCursor                      = "domain.Repository.ClaimSyncCursor"
	operationApplyReconciliationBatch             = "domain.Repository.ApplyReconciliationBatch"
	operationGetAccountRuntimeState               = "domain.Repository.GetAccountRuntimeState"
	operationListAccountRuntimeStates             = "domain.Repository.ListAccountRuntimeStates"
	operationUpsertAccountRuntimeState            = "domain.Repository.UpsertAccountRuntimeState"
	operationRecordLimitSnapshot                  = "domain.Repository.RecordLimitSnapshot"
	operationListLimitSnapshots                   = "domain.Repository.ListLimitSnapshots"
	operationRecordProviderOperation              = "domain.Repository.RecordProviderOperation"
	operationApplyProviderOperation               = "domain.Repository.ApplyProviderOperation"
	operationListProviderOperations               = "domain.Repository.ListProviderOperations"
	operationGetProviderOperationByCommand        = "domain.Repository.GetProviderOperationByCommand"
	operationGetRepositoryMergeSignal             = "domain.Repository.GetRepositoryMergeSignal"
	operationListRepositoryMergeSignals           = "domain.Repository.ListRepositoryMergeSignals"
	operationGetRepositoryChangeSignal            = "domain.Repository.GetRepositoryChangeSignal"
	operationListRepositoryChangeSignals          = "domain.Repository.ListRepositoryChangeSignals"
	operationGetRepositoryAdoptionScan            = "domain.Repository.GetRepositoryAdoptionScan"
	operationListRepositoryAdoptionScans          = "domain.Repository.ListRepositoryAdoptionScans"
	operationGetRepositoryAdoptionScanByOperation = "domain.Repository.GetRepositoryAdoptionScanByOperation"
	operationClaimOutboxEvents                    = "domain.Repository.ClaimOutboxEvents"
	operationMarkOutboxEventPublished             = "domain.Repository.MarkOutboxEventPublished"
	operationMarkOutboxEventFailed                = "domain.Repository.MarkOutboxEventFailed"
	operationMarkOutboxEventPermanentlyFailed     = "domain.Repository.MarkOutboxEventPermanentlyFailed"
)

func recordProviderOperation(ctx context.Context, db queryer, operation string, providerOperation entity.ProviderOperation) (entity.ProviderOperation, bool, error) {
	args := providerOperationArgs(providerOperation)
	stored, err := queryOne(ctx, db, operation, queryProviderOperationInsert, args, scanProviderOperation)
	if errors.Is(err, errs.ErrNotFound) {
		stored, err = queryOne(ctx, db, operation, queryProviderOperationGetReplay, args, scanProviderOperation)
		if errors.Is(err, errs.ErrNotFound) {
			return entity.ProviderOperation{}, false, errs.ErrConflict
		}
		return stored, true, err
	}
	return stored, false, err
}

func queryOne[T any](ctx context.Context, db queryer, operation string, sql string, args pgx.NamedArgs, scan func(postgreslib.RowScanner) (T, error)) (T, error) {
	result, err := scan(db.QueryRow(ctx, sql, args))
	if err != nil {
		return result, wrapError(operation, err)
	}
	return result, nil
}

func queryPage[T any](ctx context.Context, db queryer, operation string, sql string, paging pageQueryArgs, scan func(postgreslib.RowScanner) (T, error)) ([]T, query.PageResult, error) {
	rows, err := db.Query(ctx, sql, paging.args)
	if err != nil {
		return nil, query.PageResult{}, wrapError(operation, err)
	}
	items, err := postgreslib.ScanRows(rows, scan)
	if err != nil {
		return nil, query.PageResult{}, wrapError(operation, err)
	}
	pageItems, page := pageResult(items, paging.limit, paging.nextOffset)
	return pageItems, page, nil
}

func insertProviderEvents(ctx context.Context, db queryer, operation string, events []entity.ProviderEvent) ([]entity.ProviderEvent, error) {
	if len(events) == 0 {
		return nil, nil
	}
	result := make([]entity.ProviderEvent, 0, len(events))
	for _, event := range events {
		inserted, err := queryOne(ctx, db, operation, queryProviderEventInsert, providerEventArgs(event), scanProviderEvent)
		if err != nil {
			return nil, err
		}
		result = append(result, inserted)
	}
	return result, nil
}

func insertOutboxEvents(ctx context.Context, db execer, operation string, events []entity.OutboxEvent) error {
	for _, event := range events {
		if _, err := db.Exec(ctx, queryOutboxEventCreate, outboxEventArgs(event)); err != nil {
			return wrapError(operation, err)
		}
	}
	return nil
}

func storeProviderArtifactSignal(ctx context.Context, db queryer, operation string, signal entity.ProviderArtifactSignal) (entity.ProviderArtifactSignal, error) {
	stored, err := queryOne(ctx, db, operation, queryProviderArtifactSignalInsert, providerArtifactSignalArgs(signal), scanProviderArtifactSignal)
	if errors.Is(err, errs.ErrNotFound) {
		stored, err = queryOne(ctx, db, operation, queryProviderArtifactSignalGetByIdentity, providerArtifactSignalIdentityArgs(signal), scanProviderArtifactSignal)
		if err == nil && !sameProviderArtifactSignal(signal, stored) {
			err = errs.ErrConflict
		}
	}
	if err != nil {
		return entity.ProviderArtifactSignal{}, err
	}
	if !sameProviderArtifactSignal(signal, stored) {
		return entity.ProviderArtifactSignal{}, errs.ErrConflict
	}
	return stored, nil
}

func storeRepositoryMergeSignal(ctx context.Context, db queryer, operation string, signal entity.RepositoryMergeSignal) (entity.RepositoryMergeSignal, bool, error) {
	return storeIdentifiedRecord(
		ctx,
		db,
		operation,
		queryRepositoryMergeSignalInsert,
		queryRepositoryMergeSignalGetByKey,
		repositoryMergeSignalArgs(signal),
		repositoryMergeSignalIdentityArgs(signal),
		scanRepositoryMergeSignal,
		sameRepositoryMergeSignal,
		signal,
	)
}

func storeRepositoryChangeSignal(ctx context.Context, db queryer, operation string, signal entity.RepositoryChangeSignal) (entity.RepositoryChangeSignal, bool, error) {
	return storeIdentifiedRecord(
		ctx,
		db,
		operation,
		queryRepositoryChangeSignalInsert,
		queryRepositoryChangeSignalGetByKey,
		repositoryChangeSignalArgs(signal),
		repositoryChangeSignalIdentityArgs(signal),
		scanRepositoryChangeSignal,
		sameRepositoryChangeSignal,
		signal,
	)
}

func storeRepositoryAdoptionScan(ctx context.Context, db queryer, operation string, snapshot entity.RepositoryAdoptionScanSnapshot) (entity.RepositoryAdoptionScanSnapshot, bool, error) {
	stored, err := queryOne(ctx, db, operation, queryRepositoryAdoptionScanInsert, repositoryAdoptionScanArgs(snapshot), scanRepositoryAdoptionScan)
	if errors.Is(err, errs.ErrNotFound) {
		stored, err = queryOne(ctx, db, operation, queryRepositoryAdoptionScanGetByOperation, repositoryAdoptionScanOperationArgs(snapshot.ProviderOperationID), scanRepositoryAdoptionScan)
		if err == nil && !sameRepositoryAdoptionScan(snapshot, stored) {
			err = errs.ErrConflict
		}
		return stored, false, err
	}
	if err != nil {
		return entity.RepositoryAdoptionScanSnapshot{}, false, err
	}
	if !sameRepositoryAdoptionScan(snapshot, stored) {
		return entity.RepositoryAdoptionScanSnapshot{}, false, errs.ErrConflict
	}
	return stored, true, nil
}

func storeIdentifiedRecord[T any](
	ctx context.Context,
	db queryer,
	operation string,
	insertQuery string,
	lookupQuery string,
	insertArgs pgx.NamedArgs,
	lookupArgs pgx.NamedArgs,
	scan func(postgreslib.RowScanner) (T, error),
	same func(T, T) bool,
	expected T,
) (T, bool, error) {
	var zero T
	stored, err := queryOne(ctx, db, operation, insertQuery, insertArgs, scan)
	if errors.Is(err, errs.ErrNotFound) {
		stored, err = queryOne(ctx, db, operation, lookupQuery, lookupArgs, scan)
		if err == nil && !same(expected, stored) {
			err = errs.ErrConflict
		}
		return stored, false, err
	}
	if err != nil {
		return zero, false, err
	}
	if !same(expected, stored) {
		return zero, false, errs.ErrConflict
	}
	return stored, true, nil
}

func enqueueSyncCursors(ctx context.Context, db queryer, operation string, request entity.ReconciliationRequest, cursors []entity.SyncCursor) ([]entity.SyncCursor, error) {
	inserted, err := queryOne(ctx, db, operation, queryReconciliationRequestInsert, reconciliationRequestArgs(request), scanReconciliationRequest)
	if errors.Is(err, errs.ErrNotFound) {
		replayed, replayErr := queryOne(ctx, db, operation, queryReconciliationRequestGetByIdentity, syncCursorRequestLookupArgs(request), scanReconciliationRequest)
		if replayErr != nil {
			return nil, replayErr
		}
		if !sameReconciliationRequest(request, replayed) {
			return nil, errs.ErrConflict
		}
		existing, _, listErr := queryPage(ctx, db, operation, querySyncCursorListForRequest, allRowsArgs(syncCursorRequestLookupArgs(request), len(request.ArtifactKinds)), scanSyncCursor)
		if listErr == nil && len(existing) != len(request.ArtifactKinds) {
			return nil, errs.ErrConflict
		}
		return existing, listErr
	}
	if err != nil {
		return nil, err
	}
	if !sameReconciliationRequest(request, inserted) {
		return nil, errs.ErrConflict
	}
	upserted, _, upsertErr := queryPage(ctx, db, operation, querySyncCursorUpsertMany, allRowsArgs(syncCursorsBatchArgs(cursors), len(cursors)), scanSyncCursor)
	if upsertErr == nil && len(upserted) != len(cursors) {
		return nil, errs.ErrConflict
	}
	return upserted, upsertErr
}

func sameReconciliationRequest(expected entity.ReconciliationRequest, actual entity.ReconciliationRequest) bool {
	return expected.ProviderSlug == actual.ProviderSlug &&
		expected.ExternalAccountID == actual.ExternalAccountID &&
		expected.ScopeType == actual.ScopeType &&
		expected.ScopeRef == actual.ScopeRef &&
		expected.IdempotencyKey == actual.IdempotencyKey &&
		expected.Priority == actual.Priority &&
		slices.Equal(expected.ArtifactKinds, actual.ArtifactKinds)
}

func sameProviderArtifactSignal(expected entity.ProviderArtifactSignal, actual entity.ProviderArtifactSignal) bool {
	return expected.IdentityKey == actual.IdentityKey &&
		expected.ProviderSlug == actual.ProviderSlug &&
		expected.ExternalAccountID == actual.ExternalAccountID &&
		expected.Source == actual.Source &&
		expected.ScopeType == actual.ScopeType &&
		expected.ScopeRef == actual.ScopeRef &&
		expected.ObservedAt.Equal(actual.ObservedAt) &&
		slices.Equal(expected.ArtifactKinds, actual.ArtifactKinds) &&
		sameJSON(expected.TargetJSON, actual.TargetJSON) &&
		sameJSON(expected.PayloadJSON, actual.PayloadJSON)
}

type repositorySignalScope struct {
	signalKey             string
	kind                  string
	providerSlug          string
	projectID             *uuid.UUID
	repositoryID          *uuid.UUID
	repositoryFullName    string
	providerRepositoryID  string
	pullRequestNumber     int64
	pullRequestProviderID string
	pullRequestURL        string
	status                string
}

func sameRepositorySignalScope(expected repositorySignalScope, actual repositorySignalScope) bool {
	return expected.signalKey == actual.signalKey &&
		expected.kind == actual.kind &&
		expected.providerSlug == actual.providerSlug &&
		sameOptionalUUID(expected.projectID, actual.projectID) &&
		sameOptionalUUID(expected.repositoryID, actual.repositoryID) &&
		expected.repositoryFullName == actual.repositoryFullName &&
		expected.providerRepositoryID == actual.providerRepositoryID &&
		expected.pullRequestNumber == actual.pullRequestNumber &&
		expected.pullRequestProviderID == actual.pullRequestProviderID &&
		expected.pullRequestURL == actual.pullRequestURL &&
		expected.status == actual.status
}

func repositoryChangeSignalScope(signal entity.RepositoryChangeSignal) repositorySignalScope {
	return repositorySignalScopeFromFields(signal.SignalKey, string(signal.Kind), string(signal.ProviderSlug), signal.ProjectID, signal.RepositoryID, signal.RepositoryFullName, signal.ProviderRepositoryID, signal.PullRequestNumber, signal.PullRequestProviderID, signal.PullRequestURL, string(signal.Status))
}

func repositorySignalScopeFromFields(
	signalKey string,
	kind string,
	providerSlug string,
	projectID *uuid.UUID,
	repositoryID *uuid.UUID,
	repositoryFullName string,
	providerRepositoryID string,
	pullRequestNumber int64,
	pullRequestProviderID string,
	pullRequestURL string,
	status string,
) repositorySignalScope {
	return repositorySignalScope{
		signalKey:             signalKey,
		kind:                  kind,
		providerSlug:          providerSlug,
		projectID:             projectID,
		repositoryID:          repositoryID,
		repositoryFullName:    repositoryFullName,
		providerRepositoryID:  providerRepositoryID,
		pullRequestNumber:     pullRequestNumber,
		pullRequestProviderID: pullRequestProviderID,
		pullRequestURL:        pullRequestURL,
		status:                status,
	}
}

func sameRepositoryMergeSignal(expected entity.RepositoryMergeSignal, actual entity.RepositoryMergeSignal) bool {
	expectedScope := repositorySignalScopeFromFields(expected.SignalKey, string(expected.Kind), string(expected.ProviderSlug), expected.ProjectID, expected.RepositoryID, expected.RepositoryFullName, expected.ProviderRepositoryID, expected.PullRequestNumber, expected.PullRequestProviderID, expected.PullRequestURL, string(expected.Status))
	actualScope := repositorySignalScopeFromFields(actual.SignalKey, string(actual.Kind), string(actual.ProviderSlug), actual.ProjectID, actual.RepositoryID, actual.RepositoryFullName, actual.ProviderRepositoryID, actual.PullRequestNumber, actual.PullRequestProviderID, actual.PullRequestURL, string(actual.Status))
	if !sameRepositorySignalScope(expectedScope, actualScope) {
		return false
	}
	if expected.WorkItemProjectionID != actual.WorkItemProjectionID ||
		expected.ProviderWorkItemID != actual.ProviderWorkItemID ||
		expected.BaseBranch != actual.BaseBranch ||
		expected.HeadBranch != actual.HeadBranch {
		return false
	}
	return expected.MergeCommitSHA == actual.MergeCommitSHA &&
		expected.SourceRef == actual.SourceRef &&
		expected.RelatedProviderOperationRef == actual.RelatedProviderOperationRef &&
		expected.WatermarkDigest == actual.WatermarkDigest
}

func sameRepositoryChangeSignal(expected entity.RepositoryChangeSignal, actual entity.RepositoryChangeSignal) bool {
	if !sameRepositorySignalScope(repositoryChangeSignalScope(expected), repositoryChangeSignalScope(actual)) {
		return false
	}
	if expected.Ref != actual.Ref ||
		expected.CommitSHA != actual.CommitSHA ||
		expected.BeforeSHA != actual.BeforeSHA ||
		expected.PathSummaryStatus != actual.PathSummaryStatus ||
		expected.ChangedPathCount != actual.ChangedPathCount ||
		expected.PathDigest != actual.PathDigest {
		return false
	}
	return expected.BaseBranch == actual.BaseBranch &&
		expected.SourceRef == actual.SourceRef &&
		slices.Equal(expected.PathCategories, actual.PathCategories) &&
		expected.ServicesPolicyChanged == actual.ServicesPolicyChanged &&
		expected.DeployRelevantChanged == actual.DeployRelevantChanged &&
		expected.ChangeFingerprint == actual.ChangeFingerprint
}

func sameRepositoryAdoptionScan(expected entity.RepositoryAdoptionScanSnapshot, actual entity.RepositoryAdoptionScanSnapshot) bool {
	if expected.SnapshotKey != actual.SnapshotKey ||
		expected.ProviderOperationID != actual.ProviderOperationID ||
		expected.ExternalAccountID != actual.ExternalAccountID ||
		expected.ProviderSlug != actual.ProviderSlug {
		return false
	}
	if expected.RepositoryFullName != actual.RepositoryFullName ||
		expected.ProviderRepositoryID != actual.ProviderRepositoryID ||
		expected.RepositoryURL != actual.RepositoryURL ||
		expected.DefaultBranch != actual.DefaultBranch {
		return false
	}
	return expected.RequestedRef == actual.RequestedRef &&
		expected.ScannedRef == actual.ScannedRef &&
		expected.HeadSHA == actual.HeadSHA &&
		expected.Status == actual.Status &&
		expected.FileCount == actual.FileCount &&
		expected.VisibleFileCount == actual.VisibleFileCount &&
		expected.TreeTruncated == actual.TreeTruncated &&
		expected.SnapshotDigest == actual.SnapshotDigest &&
		slices.Equal(expected.Warnings, actual.Warnings) &&
		slices.Equal(expected.Markers, actual.Markers)
}

func sameOptionalUUID(expected *uuid.UUID, actual *uuid.UUID) bool {
	if expected == nil || actual == nil {
		return expected == nil && actual == nil
	}
	return *expected == *actual
}

type projectionUpdater interface {
	queryer
	execer
}

type projectionApplyResult struct {
	hasProjection               bool
	workItemApplied             bool
	workItemProjectionID        uuid.UUID
	mergeSignalApplied          bool
	mergeSignalID               uuid.UUID
	changeSignalApplied         bool
	changeSignalID              uuid.UUID
	adoptionScanApplied         bool
	adoptionScanID              uuid.UUID
	appliedCommentProjectionIDs map[uuid.UUID]struct{}
	appliedCommentProviderIDs   map[string]struct{}
	appliedRelationshipIDs      map[uuid.UUID]struct{}
}

func applyProjectionUpdate(ctx context.Context, db projectionUpdater, operation string, update providerrepo.ProjectionUpdate) (projectionApplyResult, error) {
	result := projectionApplyResult{
		appliedCommentProjectionIDs: make(map[uuid.UUID]struct{}),
		appliedCommentProviderIDs:   make(map[string]struct{}),
		appliedRelationshipIDs:      make(map[uuid.UUID]struct{}),
	}
	result.hasProjection = update.WorkItem != nil ||
		len(update.Comments) > 0 ||
		len(update.Relationships) > 0 ||
		update.MergeSignal != nil ||
		update.ChangeSignal != nil ||
		update.AdoptionScan != nil
	if update.WorkItem != nil {
		storedWorkItem, workItemApplied, err := upsertFreshWorkItemProjection(ctx, db, operation, *update.WorkItem)
		if err != nil {
			return result, err
		}
		result.workItemApplied = workItemApplied
		result.workItemProjectionID = storedWorkItem.ID
		for _, comment := range update.Comments {
			comment.WorkItemProjectionID = storedWorkItem.ID
			storedComment, commentApplied, err := upsertFreshCommentProjection(ctx, db, operation, comment)
			if err != nil {
				return result, err
			}
			if commentApplied {
				result.appliedCommentProjectionIDs[storedComment.ID] = struct{}{}
				result.appliedCommentProviderIDs[storedComment.ProviderCommentID] = struct{}{}
			}
		}
		if workItemApplied {
			watermarkRelationships, nonWatermarkRelationships := splitWatermarkRelationships(update.Relationships)
			if err := rebuildWatermarkRelationships(ctx, db, operation, storedWorkItem.ID, watermarkRelationships, result.appliedRelationshipIDs); err != nil {
				return result, err
			}
			if err := upsertRelationships(ctx, db, operation, nonWatermarkRelationships, result.appliedRelationshipIDs); err != nil {
				return result, err
			}
		}
		if update.MergeSignal != nil {
			signal := *update.MergeSignal
			signal.WorkItemProjectionID = storedWorkItem.ID
			if signal.ProjectID == nil {
				signal.ProjectID = storedWorkItem.ProjectID
			}
			if signal.RepositoryID == nil {
				signal.RepositoryID = storedWorkItem.RepositoryID
			}
			if signal.RepositoryFullName == "" {
				signal.RepositoryFullName = storedWorkItem.RepositoryFullName
			}
			storedSignal, applied, err := storeRepositoryMergeSignal(ctx, db, operation, signal)
			if err != nil {
				return result, err
			}
			result.mergeSignalApplied = applied
			result.mergeSignalID = storedSignal.ID
		}
		if update.ChangeSignal != nil {
			signal := *update.ChangeSignal
			if signal.ProjectID == nil {
				signal.ProjectID = storedWorkItem.ProjectID
			}
			if signal.RepositoryID == nil {
				signal.RepositoryID = storedWorkItem.RepositoryID
			}
			if signal.RepositoryFullName == "" {
				signal.RepositoryFullName = storedWorkItem.RepositoryFullName
			}
			storedSignal, applied, err := storeRepositoryChangeSignal(ctx, db, operation, signal)
			if err != nil {
				return result, err
			}
			result.changeSignalApplied = applied
			result.changeSignalID = storedSignal.ID
		}
		return result, nil
	}
	if err := upsertRelationships(ctx, db, operation, update.Relationships, result.appliedRelationshipIDs); err != nil {
		return result, err
	}
	if update.ChangeSignal != nil {
		storedSignal, applied, err := storeRepositoryChangeSignal(ctx, db, operation, *update.ChangeSignal)
		if err != nil {
			return result, err
		}
		result.changeSignalApplied = applied
		result.changeSignalID = storedSignal.ID
	}
	if update.AdoptionScan != nil {
		storedSnapshot, applied, err := storeRepositoryAdoptionScan(ctx, db, operation, *update.AdoptionScan)
		if err != nil {
			return result, err
		}
		result.adoptionScanApplied = applied
		result.adoptionScanID = storedSnapshot.ID
	}
	return result, nil
}

func upsertFreshWorkItemProjection(ctx context.Context, db queryer, operation string, incoming entity.ProviderWorkItemProjection) (entity.ProviderWorkItemProjection, bool, error) {
	existing, err := queryOne(ctx, db, operation, queryWorkItemProjectionGet, workItemProjectionLookupArgs(query.ProviderTargetLookup{
		ProviderSlug:     incoming.ProviderSlug,
		ProviderObjectID: incoming.ProviderWorkItemID,
	}), scanWorkItemProjection)
	if err != nil && !errors.Is(err, errs.ErrNotFound) {
		return entity.ProviderWorkItemProjection{}, false, err
	}
	if err == nil && !isProviderUpdateFresh(incoming.ProviderUpdatedAt, existing.ProviderUpdatedAt) {
		return existing, false, nil
	}
	stored, err := queryOne(ctx, db, operation, queryWorkItemProjectionUpsert, workItemProjectionArgs(incoming), scanWorkItemProjection)
	if err != nil {
		return entity.ProviderWorkItemProjection{}, false, err
	}
	return stored, providerTimestampMatches(incoming.ProviderUpdatedAt, stored.ProviderUpdatedAt), nil
}

func upsertFreshCommentProjection(ctx context.Context, db queryer, operation string, incoming entity.ProviderCommentProjection) (entity.ProviderCommentProjection, bool, error) {
	existing, err := queryOne(ctx, db, operation, queryCommentProjectionGetByProviderID, commentProjectionLookupArgs(incoming.WorkItemProjectionID, incoming.ProviderCommentID), scanCommentProjection)
	if err != nil && !errors.Is(err, errs.ErrNotFound) {
		return entity.ProviderCommentProjection{}, false, err
	}
	if err == nil && !isProviderUpdateFresh(incoming.ProviderUpdatedAt, existing.ProviderUpdatedAt) {
		return existing, false, nil
	}
	stored, err := queryOne(ctx, db, operation, queryCommentProjectionUpsert, commentProjectionArgs(incoming), scanCommentProjection)
	if err != nil {
		return entity.ProviderCommentProjection{}, false, err
	}
	return stored, providerTimestampMatches(incoming.ProviderUpdatedAt, stored.ProviderUpdatedAt), nil
}

func rebuildWatermarkRelationships(ctx context.Context, db projectionUpdater, operation string, sourceWorkItemID uuid.UUID, relationships []entity.ProviderRelationship, applied map[uuid.UUID]struct{}) error {
	currentRelationshipIDs := make([]uuid.UUID, 0, len(relationships))
	for _, relationship := range relationships {
		relationship.SourceWorkItemID = sourceWorkItemID
		stored, err := queryOne(ctx, db, operation, queryRelationshipUpsert, relationshipArgs(relationship), scanRelationship)
		if err != nil {
			return err
		}
		currentRelationshipIDs = append(currentRelationshipIDs, stored.ID)
		applied[stored.ID] = struct{}{}
	}
	_, err := db.Exec(ctx, queryRelationshipDeleteMissingWatermark, watermarkRelationshipCleanupArgs(sourceWorkItemID, currentRelationshipIDs))
	if err != nil {
		return wrapError(operation, err)
	}
	return nil
}

func upsertRelationships(ctx context.Context, db projectionUpdater, operation string, relationships []entity.ProviderRelationship, applied map[uuid.UUID]struct{}) error {
	for _, relationship := range relationships {
		stored, err := queryOne(ctx, db, operation, queryRelationshipUpsert, relationshipArgs(relationship), scanRelationship)
		if err != nil {
			return err
		}
		applied[stored.ID] = struct{}{}
	}
	return nil
}

func splitWatermarkRelationships(relationships []entity.ProviderRelationship) ([]entity.ProviderRelationship, []entity.ProviderRelationship) {
	watermark := make([]entity.ProviderRelationship, 0, len(relationships))
	other := make([]entity.ProviderRelationship, 0, len(relationships))
	for _, relationship := range relationships {
		if relationship.Source == enum.RelationshipSourceWatermark {
			watermark = append(watermark, relationship)
			continue
		}
		other = append(other, relationship)
	}
	return watermark, other
}

func isProviderUpdateFresh(incoming *time.Time, current *time.Time) bool {
	if current == nil {
		return true
	}
	if incoming == nil {
		return false
	}
	return !incoming.Before(*current)
}

func providerTimestampMatches(incoming *time.Time, stored *time.Time) bool {
	if incoming == nil {
		return stored == nil
	}
	if stored == nil {
		return false
	}
	return incoming.Equal(*stored)
}

func filterProviderEvents(events []entity.ProviderEvent, update providerrepo.ProjectionUpdate, result projectionApplyResult) []entity.ProviderEvent {
	if !result.hasProjection {
		return events
	}
	filtered := make([]entity.ProviderEvent, 0, len(events))
	for _, event := range events {
		switch event.AggregateType {
		case providerevents.AggregateWorkItem:
			if update.WorkItem != nil && result.workItemApplied && event.AggregateID == update.WorkItem.ProviderWorkItemID {
				filtered = append(filtered, event)
			}
		case providerevents.AggregateComment:
			if _, ok := result.appliedCommentProviderIDs[event.AggregateID]; ok {
				filtered = append(filtered, event)
			}
		case providerevents.AggregateRepositoryChangeSignal:
			if result.changeSignalApplied && update.ChangeSignal != nil && event.AggregateID == update.ChangeSignal.SignalKey {
				filtered = append(filtered, event)
			}
		default:
			filtered = append(filtered, event)
		}
	}
	return filtered
}

func filterOutboxEvents(events []entity.OutboxEvent, providerEvents []entity.ProviderEvent, result projectionApplyResult) []entity.OutboxEvent {
	if !result.hasProjection {
		return events
	}
	providerEventIDs := make(map[uuid.UUID]struct{}, len(providerEvents))
	for _, event := range providerEvents {
		providerEventIDs[event.ID] = struct{}{}
	}
	filtered := make([]entity.OutboxEvent, 0, len(events))
	for _, event := range events {
		switch event.EventType {
		case providerevents.EventWebhookNormalized:
			if _, ok := providerEventIDs[event.AggregateID]; ok {
				filtered = append(filtered, event)
			}
		case providerevents.EventWorkItemSynced:
			if result.workItemApplied && event.AggregateID == result.workItemProjectionID {
				filtered = append(filtered, event)
			}
		case providerevents.EventCommentSynced:
			if _, ok := result.appliedCommentProjectionIDs[event.AggregateID]; ok {
				filtered = append(filtered, event)
			}
		case providerevents.EventRelationshipSynced:
			if _, ok := result.appliedRelationshipIDs[event.AggregateID]; ok {
				filtered = append(filtered, event)
			}
		case providerevents.EventRepositoryBootstrapMerged, providerevents.EventRepositoryAdoptionMerged:
			if result.mergeSignalApplied && event.AggregateID == result.mergeSignalID {
				filtered = append(filtered, event)
			}
		case providerevents.EventRepositoryChanged:
			if result.changeSignalApplied && event.AggregateID == result.changeSignalID {
				filtered = append(filtered, event)
			}
		case providerevents.EventRepositoryAdoptionScanCompleted:
			if result.adoptionScanApplied && event.AggregateID == result.adoptionScanID {
				filtered = append(filtered, event)
			}
		default:
			filtered = append(filtered, event)
		}
	}
	return filtered
}

func sameWebhookEvent(left entity.WebhookEvent, right entity.WebhookEvent) bool {
	samePayload := bytes.Equal(compactJSON(left.PayloadJSON), compactJSON(right.PayloadJSON))
	if left.PayloadDigest != "" && right.PayloadDigest != "" && left.PayloadDigest == right.PayloadDigest {
		samePayload = true
	}
	if migratedWebhookEnvelopeReplaysByIdentity(right) {
		samePayload = true
	}
	return left.ProviderSlug == right.ProviderSlug &&
		left.DeliveryID == right.DeliveryID &&
		left.EventName == right.EventName &&
		left.RepositoryProviderID == right.RepositoryProviderID &&
		samePayload
}

func migratedWebhookEnvelopeReplaysByIdentity(event entity.WebhookEvent) bool {
	switch event.ProcessingStatus {
	case enum.WebhookProcessingStatusProcessed, enum.WebhookProcessingStatusIgnored,
		enum.WebhookProcessingStatusPending, enum.WebhookProcessingStatusFailed:
	default:
		return false
	}
	var envelope struct {
		PayloadCleanupReason string `json:"payload_cleanup_reason"`
		PayloadDigestSource  string `json:"payload_digest_source"`
		PayloadStorage       string `json:"payload_storage"`
	}
	if err := json.Unmarshal(event.PayloadJSON, &envelope); err != nil {
		return false
	}
	if envelope.PayloadDigestSource != "postgres_jsonb_text" {
		return false
	}
	switch event.ProcessingStatus {
	case enum.WebhookProcessingStatusProcessed, enum.WebhookProcessingStatusIgnored:
		return envelope.PayloadStorage == "redacted_after_terminal_processing"
	case enum.WebhookProcessingStatusPending, enum.WebhookProcessingStatusFailed:
		return envelope.PayloadStorage == "safe_envelope_only" &&
			envelope.PayloadCleanupReason == "raw_payload_removed"
	default:
		return false
	}
}

func sameJSON(left []byte, right []byte) bool {
	var leftValue any
	var rightValue any
	if json.Unmarshal(left, &leftValue) != nil || json.Unmarshal(right, &rightValue) != nil {
		return bytes.Equal(compactJSON(left), compactJSON(right))
	}
	return reflect.DeepEqual(leftValue, rightValue)
}

func compactJSON(raw []byte) []byte {
	var compacted bytes.Buffer
	if err := json.Compact(&compacted, raw); err != nil {
		return bytes.TrimSpace(raw)
	}
	return compacted.Bytes()
}
