package serviceactor

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/core/secrets"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
)

const (
	deliveryDispatchDefaultLimit        = 10
	deliveryDispatchMaxLimit            = 100
	deliveryDispatchDefaultLease        = 2 * time.Minute
	deliveryDispatchMaxLease            = 15 * time.Minute
	deliveryDispatchDefaultRetryBackoff = 30 * time.Second
	deliveryDispatchMaxRetryBackoff     = 15 * time.Minute
)

// DeliveryEnvelope is the runtime-facing work item produced by the dispatcher.
// It couples the durable delivery row with the leased attempt and, for external
// transports only, the plaintext delivery token returned exactly once.
type DeliveryEnvelope struct {
	EnvelopeID    string
	Delivery      EventDelivery
	Attempt       DeliveryAttempt
	DeliveryToken string
	EventType     string
	EventAction   string
	EventPayload  []byte
	CreatedAt     time.Time
}

// DeliveryDispatchClaim selects ready deliveries and leases attempts for a
// callback sender, runner stream session, or AgentTeam-owned internal handler.
type DeliveryDispatchClaim struct {
	WorkspaceID                string
	ServiceInstallationID      string
	SubscriptionKind           string
	SubscriptionKinds          []string
	DeliveryPolicy             string
	ServiceConnectionRef       string
	TransportKind              string
	ServiceRuntimeCredentialID string
	ServiceRuntimeSessionID    string
	LeaseOwner                 string
	LeaseDuration              time.Duration
	Limit                      int
	Now                        time.Time
	CallbackURLSnapshot        string
}

// DeliveryAttemptLeaseExpire marks expired attempt leases and releases their
// parent deliveries for another dispatcher claim.
type DeliveryAttemptLeaseExpire struct {
	WorkspaceID           string
	ServiceInstallationID string
	SubscriptionKind      string
	SubscriptionKinds     []string
	DeliveryPolicy        string
	ServiceConnectionRef  string
	Now                   time.Time
	Limit                 int
}

// DeliveryAttemptRetrySchedule marks one transport attempt failed and schedules
// the parent delivery for a later retry.
type DeliveryAttemptRetrySchedule struct {
	WorkspaceID    string
	AttemptID      string
	ErrorCode      string
	ErrorDetail    string
	BaseBackoff    time.Duration
	MaxBackoff     time.Duration
	Now            time.Time
	ResponseStatus *int
	ResponseBody   string
}

// DeliveryAttemptRetryResult is the parent delivery and failed attempt after a
// retry schedule decision.
type DeliveryAttemptRetryResult struct {
	Delivery    EventDelivery
	Attempt     DeliveryAttempt
	NextRetryAt time.Time
}

// ClaimDeliveryEnvelopes leases ready deliveries and returns runtime envelopes.
func (store *Store) ClaimDeliveryEnvelopes(ctx context.Context, params DeliveryDispatchClaim) ([]DeliveryEnvelope, error) {
	if store == nil || store.db == nil {
		return nil, fmt.Errorf("service actor store database is not configured")
	}
	var envelopes []DeliveryEnvelope
	if err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		var err error
		envelopes, err = claimDeliveryEnvelopes(ctx, tx, params)
		return err
	}); err != nil {
		return nil, err
	}
	return envelopes, nil
}

// ClaimDeliveryEnvelopes leases ready deliveries in the existing transaction.
func (store *TxStore) ClaimDeliveryEnvelopes(ctx context.Context, params DeliveryDispatchClaim) ([]DeliveryEnvelope, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return nil, err
	}
	return claimDeliveryEnvelopes(ctx, execer, params)
}

// ExpireDeliveryAttemptLeases releases expired transport leases.
func (store *Store) ExpireDeliveryAttemptLeases(ctx context.Context, params DeliveryAttemptLeaseExpire) (int64, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return 0, err
	}
	return expireDeliveryAttemptLeases(ctx, execer, params)
}

// ExpireDeliveryAttemptLeases releases expired transport leases in a transaction.
func (store *TxStore) ExpireDeliveryAttemptLeases(ctx context.Context, params DeliveryAttemptLeaseExpire) (int64, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return 0, err
	}
	return expireDeliveryAttemptLeases(ctx, execer, params)
}

// MarkDeliveryAttemptDelivered records a successful transport delivery for the
// current leased attempt only.
func (store *Store) MarkDeliveryAttemptDelivered(ctx context.Context, params DeliveryAttemptDeliveredMark) (DeliveryAttempt, error) {
	if store == nil || store.db == nil {
		return DeliveryAttempt{}, fmt.Errorf("service actor store database is not configured")
	}
	var attempt DeliveryAttempt
	if err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		var err error
		attempt, err = markDeliveryAttemptDelivered(ctx, tx, params)
		return err
	}); err != nil {
		return DeliveryAttempt{}, err
	}
	return attempt, nil
}

// MarkDeliveryAttemptDelivered records a successful transport delivery in a transaction.
func (store *TxStore) MarkDeliveryAttemptDelivered(ctx context.Context, params DeliveryAttemptDeliveredMark) (DeliveryAttempt, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return DeliveryAttempt{}, err
	}
	return markDeliveryAttemptDelivered(ctx, execer, params)
}

// ScheduleDeliveryAttemptRetry records a failed transport attempt and schedules
// its delivery for retry using exponential backoff.
func (store *Store) ScheduleDeliveryAttemptRetry(ctx context.Context, params DeliveryAttemptRetrySchedule) (DeliveryAttemptRetryResult, error) {
	if store == nil || store.db == nil {
		return DeliveryAttemptRetryResult{}, fmt.Errorf("service actor store database is not configured")
	}
	var result DeliveryAttemptRetryResult
	if err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		var err error
		result, err = scheduleDeliveryAttemptRetry(ctx, tx, params)
		return err
	}); err != nil {
		return DeliveryAttemptRetryResult{}, err
	}
	return result, nil
}

// ScheduleDeliveryAttemptRetry records a failed transport attempt in a transaction.
func (store *TxStore) ScheduleDeliveryAttemptRetry(ctx context.Context, params DeliveryAttemptRetrySchedule) (DeliveryAttemptRetryResult, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return DeliveryAttemptRetryResult{}, err
	}
	return scheduleDeliveryAttemptRetry(ctx, execer, params)
}

func claimDeliveryEnvelopes(ctx context.Context, execer coredb.QueryExecer, params DeliveryDispatchClaim) ([]DeliveryEnvelope, error) {
	workspaceID := trim(params.WorkspaceID)
	transportKind := trim(params.TransportKind)
	if workspaceID == "" || transportKind == "" {
		return nil, badRequest("delivery claim workspace and transport are required")
	}
	if !validRuntimeTransportKind(transportKind) {
		return nil, badRequest("delivery claim transport is invalid")
	}
	deliveryPolicy, connectionRef, err := normalizeDeliveryDispatchRoute(transportKind, params.DeliveryPolicy, params.ServiceConnectionRef)
	if err != nil {
		return nil, err
	}
	params.DeliveryPolicy = deliveryPolicy
	params.ServiceConnectionRef = connectionRef
	if transportKind == RuntimeTransportKindInternalHandler {
		kinds := deliverySubscriptionKindFilter(params.SubscriptionKind, params.SubscriptionKinds)
		if kinds != SubscriptionKindAppNotificationAction {
			return nil, badRequest("internal handler delivery claim only supports App notification actions")
		}
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	limit := normalizeDeliveryDispatchLimit(params.Limit)
	leaseDuration := normalizeDeliveryLeaseDuration(params.LeaseDuration)
	leaseExpiresAt := now.Add(leaseDuration)
	leaseOwner := deliveryLeaseOwner(params, transportKind)
	if leaseOwner == "" {
		return nil, badRequest("delivery claim lease owner is required")
	}
	// Installation-scoped cancellation and notification lifecycle mutations
	// take this same fence before snapshotting active attempts. Serializing a
	// claim here, before runtime credential/session rows are locked, prevents a
	// newly leased attempt from being inserted between a canceler's attempt
	// snapshot and delivery transition without reversing uninstall lock order.
	if err := lockDeliveryClaimInstallation(ctx, execer, workspaceID, params.ServiceInstallationID); err != nil {
		return nil, err
	}
	claimIdentity, err := validateDeliveryDispatchIdentity(ctx, execer, workspaceID, transportKind, &params, now)
	if err != nil {
		return nil, err
	}
	if _, err := expireDeliveryAttemptLeases(ctx, execer, DeliveryAttemptLeaseExpire{
		WorkspaceID:           workspaceID,
		ServiceInstallationID: params.ServiceInstallationID,
		SubscriptionKind:      params.SubscriptionKind,
		SubscriptionKinds:     params.SubscriptionKinds,
		DeliveryPolicy:        deliveryPolicy,
		ServiceConnectionRef:  connectionRef,
		Now:                   now,
		// Active-attempt cardinality is normally one per delivery. Keep a
		// bounded repair margin for legacy duplicate leases so candidate locking
		// never needs to reverse delivery -> attempt ordering.
		Limit: limit * 4,
	}); err != nil {
		return nil, err
	}
	deliveries, err := listReadyDeliveryClaimsForUpdate(ctx, execer, deliveryReadyClaimQuery{
		WorkspaceID:           workspaceID,
		ServiceInstallationID: params.ServiceInstallationID,
		SubscriptionKind:      params.SubscriptionKind,
		SubscriptionKinds:     params.SubscriptionKinds,
		DeliveryPolicy:        deliveryPolicy,
		ServiceConnectionRef:  connectionRef,
		Now:                   now,
		Limit:                 limit,
	})
	if err != nil {
		return nil, err
	}
	envelopes := make([]DeliveryEnvelope, 0, len(deliveries))
	for _, delivery := range deliveries {
		if err := ensureDeliveryHasNoActiveLeaseAttempts(ctx, execer, workspaceID, delivery.ID); err != nil {
			return nil, err
		}
		var token string
		var tokenHash string
		var tokenExpiresAt *time.Time
		if transportKind != RuntimeTransportKindInternalHandler {
			token, err = secrets.NewToken(32)
			if err != nil {
				return nil, fmt.Errorf("generate service event delivery token: %w", err)
			}
			tokenHash = secrets.HashToken(token)
			tokenExpiresAt = &leaseExpiresAt
		}
		attempt, err := createDeliveryAttempt(ctx, execer, DeliveryAttemptCreate{
			WorkspaceID:                workspaceID,
			ServiceEventDeliveryID:     delivery.ID,
			TransportKind:              transportKind,
			ServiceRuntimeCredentialID: params.ServiceRuntimeCredentialID,
			ServiceRuntimeSessionID:    params.ServiceRuntimeSessionID,
			LeaseOwner:                 leaseOwner,
			LeaseExpiresAt:             &leaseExpiresAt,
			DeliveryTokenHash:          tokenHash,
			DeliveryTokenExpiresAt:     tokenExpiresAt,
			CallbackURLSnapshot:        params.CallbackURLSnapshot,
			Now:                        now,
		})
		if err != nil {
			return nil, err
		}
		if err := markDeliveryAttemptLeased(ctx, execer, delivery.ID, attempt, leaseOwner, leaseExpiresAt, now); err != nil {
			return nil, err
		}
		leasedDelivery, err := loadEventDelivery(ctx, execer, delivery.ID)
		if err != nil {
			return nil, err
		}
		envelopes = append(envelopes, DeliveryEnvelope{
			EnvelopeID:    attempt.ID,
			Delivery:      leasedDelivery,
			Attempt:       attempt,
			DeliveryToken: token,
			EventType:     leasedDelivery.EventType,
			EventAction:   leasedDelivery.EventAction,
			EventPayload:  leasedDelivery.EventPayload,
			CreatedAt:     now,
		})
	}
	if len(envelopes) > 0 && trim(params.ServiceRuntimeSessionID) != "" {
		if err := markRuntimeSessionClaimed(ctx, execer, workspaceID, params.ServiceRuntimeSessionID, claimIdentity.ServiceRuntimeCredentialID, claimIdentity.ServiceInstallationID, now); err != nil {
			return nil, err
		}
	}
	return envelopes, nil
}

func lockDeliveryClaimInstallation(ctx context.Context, queryer coredb.QueryRower, workspaceID, installationID string) error {
	workspaceID = trim(workspaceID)
	installationID = trim(installationID)
	if workspaceID == "" || installationID == "" {
		return badRequest("delivery claim workspace and installation are required")
	}
	var lockedID string
	if err := queryer.QueryRow(ctx, `
SELECT id
FROM service_installations
WHERE workspace_id = $1
  AND id = $2
  AND service_kind = 'app'
  AND status = 'active'
FOR UPDATE
`, workspaceID, installationID).Scan(&lockedID); err != nil {
		if coredb.IsNotFound(err) {
			return notFound("service installation not found")
		}
		return fmt.Errorf("lock service installation for delivery claim: %w", err)
	}
	return nil
}

type deliveryDispatchIdentity struct {
	ServiceInstallationID      string
	ServiceMemberID            string
	ServiceRuntimeCredentialID string
	ServiceRuntimeSessionID    string
}

func validateDeliveryDispatchIdentity(ctx context.Context, execer coredb.QueryExecer, workspaceID string, transportKind string, params *DeliveryDispatchClaim, now time.Time) (deliveryDispatchIdentity, error) {
	identity := deliveryDispatchIdentity{
		ServiceInstallationID:      trim(params.ServiceInstallationID),
		ServiceRuntimeCredentialID: trim(params.ServiceRuntimeCredentialID),
		ServiceRuntimeSessionID:    trim(params.ServiceRuntimeSessionID),
	}
	if transportKind == RuntimeTransportKindInternalHandler {
		if identity.ServiceInstallationID == "" {
			return identity, badRequest("internal handler delivery claim installation is required")
		}
		if identity.ServiceRuntimeCredentialID != "" || identity.ServiceRuntimeSessionID != "" {
			return identity, badRequest("internal handler delivery claim cannot use runtime credentials or sessions")
		}
		if trim(params.CallbackURLSnapshot) != "" {
			return identity, badRequest("internal handler delivery claim cannot use a callback URL")
		}
		return identity, nil
	}
	if transportKind == RuntimeTransportKindRunnerStream && (identity.ServiceRuntimeCredentialID == "" || identity.ServiceRuntimeSessionID == "") {
		return identity, badRequest("runner stream delivery claim credential and session are required")
	}
	if identity.ServiceRuntimeCredentialID != "" {
		credential, err := loadRuntimeCredentialForUpdate(ctx, execer, workspaceID, identity.ServiceRuntimeCredentialID)
		if err != nil {
			return identity, err
		}
		if credential.Status != RuntimeCredentialStatusActive {
			return identity, badRequest("runtime credential must be active to claim deliveries")
		}
		if credential.ExpiresAt != nil && !credential.ExpiresAt.After(now) {
			return identity, badRequest("runtime credential is expired")
		}
		if credential.TransportKind != transportKind {
			return identity, badRequest("runtime credential transport does not match delivery claim")
		}
		if identity.ServiceInstallationID != "" && identity.ServiceInstallationID != credential.ServiceInstallationID {
			return identity, badRequest("runtime credential installation does not match delivery claim")
		}
		identity.ServiceInstallationID = credential.ServiceInstallationID
		identity.ServiceMemberID = credential.ServiceMemberID
		params.ServiceInstallationID = credential.ServiceInstallationID
	}
	if identity.ServiceRuntimeSessionID != "" {
		session, err := loadRuntimeSessionForUpdate(ctx, execer, workspaceID, identity.ServiceRuntimeSessionID)
		if err != nil {
			return identity, err
		}
		if session.Status != RuntimeSessionStatusOnline {
			return identity, badRequest("runtime session must be online to claim deliveries")
		}
		if identity.ServiceInstallationID != "" && identity.ServiceInstallationID != session.ServiceInstallationID {
			return identity, badRequest("runtime session installation does not match delivery claim")
		}
		if identity.ServiceMemberID != "" && identity.ServiceMemberID != session.ServiceMemberID {
			return identity, badRequest("runtime session member does not match delivery claim")
		}
		if identity.ServiceRuntimeCredentialID != "" && identity.ServiceRuntimeCredentialID != session.ServiceRuntimeCredentialID {
			return identity, badRequest("runtime session credential does not match delivery claim")
		}
		identity.ServiceInstallationID = session.ServiceInstallationID
		identity.ServiceMemberID = session.ServiceMemberID
		identity.ServiceRuntimeCredentialID = session.ServiceRuntimeCredentialID
		params.ServiceInstallationID = session.ServiceInstallationID
		params.ServiceRuntimeCredentialID = session.ServiceRuntimeCredentialID
	}
	return identity, nil
}

type deliveryReadyClaimQuery struct {
	WorkspaceID           string
	ServiceInstallationID string
	SubscriptionKind      string
	SubscriptionKinds     []string
	DeliveryPolicy        string
	ServiceConnectionRef  string
	Now                   time.Time
	Limit                 int
}

type readyDeliveryClaimCandidate struct {
	ID          string
	CreatedAt   time.Time
	NextRetryAt *time.Time
}

func listReadyDeliveryClaimsForUpdate(ctx context.Context, queryer coredb.Queryer, params deliveryReadyClaimQuery) ([]EventDelivery, error) {
	workspaceID := trim(params.WorkspaceID)
	if workspaceID == "" {
		return nil, badRequest("ready delivery claim workspace is required")
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	limit := normalizeDeliveryDispatchLimit(params.Limit)
	noRetryCandidates, err := listReadyDeliveryClaimBranchCandidates(ctx, queryer, readyDeliveryClaimNoRetrySQL(), params, now, limit)
	if err != nil {
		return nil, err
	}
	dueRetryCandidates, err := listReadyDeliveryClaimBranchCandidates(ctx, queryer, readyDeliveryClaimDueRetrySQL(), params, now, limit)
	if err != nil {
		return nil, err
	}
	candidates := mergeReadyDeliveryClaimCandidates(noRetryCandidates, dueRetryCandidates, limit)
	return lockReadyDeliveryClaimsForUpdate(ctx, queryer, params, now, candidates)
}

func readyDeliveryClaimNoRetrySQL() string {
	return readyDeliveryClaimCandidateSQL("delivery.next_retry_at IS NULL", "delivery.created_at, delivery.id")
}

func readyDeliveryClaimDueRetrySQL() string {
	return readyDeliveryClaimCandidateSQL(`delivery.next_retry_at IS NOT NULL
  AND delivery.next_retry_at <= $6`, "delivery.next_retry_at, delivery.id")
}

func readyDeliveryClaimCandidateSQL(readyPredicate string, orderBy string) string {
	return `
SELECT delivery.id, delivery.created_at, delivery.next_retry_at
FROM service_event_deliveries delivery
` + readyDeliveryClaimFilterSQL(readyPredicate) + `
ORDER BY ` + orderBy + `
LIMIT $7
`
}

func readyDeliveryClaimLockSQL(readyPredicate string, idPlaceholders string) string {
	return eventDeliverySelectFromSQL("service_event_deliveries delivery", "delivery") + `
` + readyDeliveryClaimFilterSQL(readyPredicate) + `
  AND delivery.id IN (` + idPlaceholders + `)
ORDER BY delivery.id
FOR UPDATE
`
}

func readyDeliveryClaimFilterSQL(readyPredicate string) string {
	return `WHERE delivery.workspace_id = $1
  AND ($2 = '' OR delivery.service_installation_id = $2)
	AND ($3 = '' OR FIND_IN_SET(delivery.subscription_kind, $3) > 0)
  AND ($4 = '' OR delivery.delivery_policy = $4)
  AND ($5 = '' OR delivery.service_connection_ref = $5)
  AND delivery.status IN ('running', 'pending', 'retry_waiting')
  AND ` + readyPredicate + `
  AND EXISTS (
      SELECT 1
      FROM service_installations installation
      WHERE installation.id = delivery.service_installation_id
		AND installation.workspace_id = delivery.workspace_id
		AND installation.service_kind = delivery.service_kind
        AND installation.status = 'active'
  )
  AND EXISTS (
      SELECT 1
      FROM service_event_subscriptions subscription
      WHERE subscription.id = delivery.service_event_subscription_id
		AND subscription.workspace_id = delivery.workspace_id
        AND subscription.status = 'active'
        AND subscription.removed_at IS NULL
  )
  AND EXISTS (
      SELECT 1
      FROM service_target_bindings binding
      WHERE binding.id = delivery.service_target_binding_id
		AND binding.workspace_id = delivery.workspace_id
        AND binding.status = 'active'
        AND binding.removed_at IS NULL
  )
	AND (
		delivery.service_kind <> 'app'
		OR delivery.subscription_kind NOT IN ('app_mention', 'app_notification_action')
		OR EXISTS (
			SELECT 1
			FROM service_installations app_authority
			WHERE app_authority.id = delivery.service_installation_id
			  AND app_authority.workspace_id = delivery.workspace_id
			  AND app_authority.service_kind = delivery.service_kind
			  AND app_authority.status = 'active'
			  AND COALESCE(JSON_TYPE(JSON_EXTRACT(
			      app_authority.metadata,
			      '$.manifest.capabilities.runtime'
			  )), '') = 'OBJECT'
			  AND (
				(
					delivery.subscription_kind = 'app_mention'
					AND JSON_CONTAINS(
						COALESCE(JSON_EXTRACT(
							app_authority.metadata,
							'$.manifest.capabilities.runtime.events'
						), JSON_ARRAY()),
						JSON_QUOTE('agentteam.message.mention')
					) = 1
					AND EXISTS (
						SELECT 1
						FROM service_grants mention_grant
						WHERE mention_grant.service_installation_id = delivery.service_installation_id
						  AND mention_grant.scope = 'app.mentions.read'
						  AND mention_grant.revoked_at IS NULL
						  AND (mention_grant.expires_at IS NULL OR mention_grant.expires_at > $6)
					)
					AND EXISTS (
						SELECT 1
						FROM service_grants message_grant
						WHERE message_grant.service_installation_id = delivery.service_installation_id
						  AND message_grant.scope = 'messages.write'
						  AND message_grant.revoked_at IS NULL
						  AND (message_grant.expires_at IS NULL OR message_grant.expires_at > $6)
					)
				)
				OR (
					delivery.subscription_kind = 'app_notification_action'
					AND JSON_CONTAINS(
						COALESCE(JSON_EXTRACT(
							app_authority.metadata,
							'$.manifest.capabilities.runtime.events'
						), JSON_ARRAY()),
						JSON_QUOTE('agentteam.app.notification.action')
					) = 1
					AND EXISTS (
						SELECT 1
						FROM service_grants action_grant
						WHERE action_grant.service_installation_id = delivery.service_installation_id
						  AND action_grant.scope = 'app.notification_actions.read'
						  AND action_grant.revoked_at IS NULL
						  AND (action_grant.expires_at IS NULL OR action_grant.expires_at > $6)
					)
				)
			  )
		)
	)
	AND (
		$4 <> 'app_internal_handler'
		OR (
			delivery.service_kind = 'app'
			AND delivery.subscription_kind = 'app_notification_action'
			AND EXISTS (
				SELECT 1
				FROM service_installations app_installation
				WHERE app_installation.id = delivery.service_installation_id
				  AND app_installation.workspace_id = delivery.workspace_id
				  AND app_installation.service_kind = delivery.service_kind
				  AND app_installation.status = 'active'
				  AND COALESCE(JSON_TYPE(JSON_EXTRACT(
				      app_installation.metadata,
				      '$.manifest.capabilities.runtime'
				  )), '') = 'OBJECT'
				  AND JSON_CONTAINS(
				      COALESCE(JSON_EXTRACT(
				          app_installation.metadata,
				          '$.manifest.capabilities.runtime.events'
				      ), JSON_ARRAY()),
				      JSON_QUOTE('agentteam.app.notification.action')
				  ) = 1
			)
			AND EXISTS (
				SELECT 1
				FROM service_grants action_grant
				WHERE action_grant.service_installation_id = delivery.service_installation_id
				  AND action_grant.scope = 'app.notification_actions.read'
				  AND action_grant.revoked_at IS NULL
				  AND (action_grant.expires_at IS NULL OR action_grant.expires_at > $6)
			)
		)
	)
  AND NOT EXISTS (
      SELECT 1
      FROM service_event_delivery_attempts attempt
      WHERE attempt.service_event_delivery_id = delivery.id
        AND attempt.status IN ('leased', 'delivered')
  )
`
}

func listReadyDeliveryClaimBranchCandidates(ctx context.Context, queryer coredb.Queryer, query string, params deliveryReadyClaimQuery, now time.Time, limit int) ([]readyDeliveryClaimCandidate, error) {
	rows, err := queryer.Query(ctx, query, trim(params.WorkspaceID), trim(params.ServiceInstallationID), deliverySubscriptionKindFilter(params.SubscriptionKind, params.SubscriptionKinds), trim(params.DeliveryPolicy), trim(params.ServiceConnectionRef), now, limit)
	if err != nil {
		return nil, fmt.Errorf("list ready service event deliveries for claim: %w", err)
	}
	defer rows.Close()
	candidates := make([]readyDeliveryClaimCandidate, 0, limit)
	for rows.Next() {
		var candidate readyDeliveryClaimCandidate
		var nextRetryAt sql.NullTime
		if err := rows.Scan(&candidate.ID, &candidate.CreatedAt, &nextRetryAt); err != nil {
			return nil, fmt.Errorf("scan ready service event delivery claim candidate: %w", err)
		}
		candidate.ID = trim(candidate.ID)
		candidate.NextRetryAt = ptrFromNullTime(nextRetryAt)
		if candidate.ID != "" {
			candidates = append(candidates, candidate)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate ready service event delivery claim candidates: %w", err)
	}
	return candidates, nil
}

func mergeReadyDeliveryClaimCandidates(noRetryCandidates []readyDeliveryClaimCandidate, dueRetryCandidates []readyDeliveryClaimCandidate, limit int) []readyDeliveryClaimCandidate {
	if limit <= 0 || (len(noRetryCandidates) == 0 && len(dueRetryCandidates) == 0) {
		return nil
	}
	capacity := len(noRetryCandidates) + len(dueRetryCandidates)
	if capacity > limit {
		capacity = limit
	}
	merged := make([]readyDeliveryClaimCandidate, 0, capacity)
	noRetryIndex := 0
	dueRetryIndex := 0
	for len(merged) < limit && (noRetryIndex < len(noRetryCandidates) || dueRetryIndex < len(dueRetryCandidates)) {
		if dueRetryIndex >= len(dueRetryCandidates) || (noRetryIndex < len(noRetryCandidates) && readyDeliveryClaimCandidateBefore(noRetryCandidates[noRetryIndex], dueRetryCandidates[dueRetryIndex])) {
			merged = append(merged, noRetryCandidates[noRetryIndex])
			noRetryIndex++
			continue
		}
		merged = append(merged, dueRetryCandidates[dueRetryIndex])
		dueRetryIndex++
	}
	return merged
}

func readyDeliveryClaimCandidateBefore(left readyDeliveryClaimCandidate, right readyDeliveryClaimCandidate) bool {
	leftReadyAt := readyDeliveryClaimCandidateTime(left)
	rightReadyAt := readyDeliveryClaimCandidateTime(right)
	if leftReadyAt.Equal(rightReadyAt) {
		return trim(left.ID) < trim(right.ID)
	}
	return leftReadyAt.Before(rightReadyAt)
}

func readyDeliveryClaimCandidateTime(candidate readyDeliveryClaimCandidate) time.Time {
	if candidate.NextRetryAt != nil {
		return *candidate.NextRetryAt
	}
	return candidate.CreatedAt
}

func lockReadyDeliveryClaimsForUpdate(ctx context.Context, queryer coredb.Queryer, params deliveryReadyClaimQuery, now time.Time, candidates []readyDeliveryClaimCandidate) ([]EventDelivery, error) {
	if len(candidates) == 0 {
		return nil, nil
	}
	noRetryIDs := make([]string, 0, len(candidates))
	dueRetryIDs := make([]string, 0, len(candidates))
	for _, candidate := range candidates {
		if candidate.NextRetryAt == nil {
			noRetryIDs = append(noRetryIDs, candidate.ID)
			continue
		}
		dueRetryIDs = append(dueRetryIDs, candidate.ID)
	}
	lockedByID := make(map[string]EventDelivery, len(candidates))
	if err := lockReadyDeliveryClaimBranchForUpdate(ctx, queryer, "delivery.next_retry_at IS NULL", params, now, noRetryIDs, lockedByID); err != nil {
		return nil, err
	}
	if err := lockReadyDeliveryClaimBranchForUpdate(ctx, queryer, `delivery.next_retry_at IS NOT NULL
  AND delivery.next_retry_at <= $6`, params, now, dueRetryIDs, lockedByID); err != nil {
		return nil, err
	}
	deliveries := make([]EventDelivery, 0, len(candidates))
	for _, candidate := range candidates {
		if delivery, ok := lockedByID[candidate.ID]; ok {
			deliveries = append(deliveries, delivery)
		}
	}
	return deliveries, nil
}

func lockReadyDeliveryClaimBranchForUpdate(ctx context.Context, queryer coredb.Queryer, readyPredicate string, params deliveryReadyClaimQuery, now time.Time, ids []string, lockedByID map[string]EventDelivery) error {
	ids = trimSlice(ids)
	if len(ids) == 0 {
		return nil
	}
	placeholders := make([]string, len(ids))
	args := make([]any, 0, 6+len(ids))
	args = append(args, trim(params.WorkspaceID), trim(params.ServiceInstallationID), deliverySubscriptionKindFilter(params.SubscriptionKind, params.SubscriptionKinds), trim(params.DeliveryPolicy), trim(params.ServiceConnectionRef), now)
	for index, id := range ids {
		placeholders[index] = fmt.Sprintf("$%d", index+7)
		args = append(args, id)
	}
	rows, err := queryer.Query(ctx, readyDeliveryClaimLockSQL(readyPredicate, strings.Join(placeholders, ", ")), args...)
	if err != nil {
		return fmt.Errorf("lock ready service event deliveries for claim: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		delivery, err := scanEventDelivery(rows)
		if err != nil {
			return fmt.Errorf("scan locked ready service event delivery claim: %w", err)
		}
		delivery.ID = trim(delivery.ID)
		if delivery.ID != "" {
			lockedByID[delivery.ID] = delivery
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate locked ready service event delivery claims: %w", err)
	}
	return nil
}

func expireDeliveryAttemptLeases(ctx context.Context, execer coredb.QueryExecer, params DeliveryAttemptLeaseExpire) (int64, error) {
	workspaceID := trim(params.WorkspaceID)
	if workspaceID == "" {
		return 0, badRequest("delivery lease expiry workspace is required")
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	limit := normalizeDeliveryDispatchLimit(params.Limit)
	rows, err := execer.Query(ctx, `
SELECT attempt.id, attempt.service_event_delivery_id, attempt.lease_owner, attempt.lease_expires_at
FROM service_event_delivery_attempts attempt
JOIN service_event_deliveries delivery
  ON delivery.id = attempt.service_event_delivery_id
WHERE attempt.workspace_id = $1
  AND ($2 = '' OR attempt.service_installation_id = $2)
	AND ($3 = '' OR FIND_IN_SET(delivery.subscription_kind, $3) > 0)
  AND ($4 = '' OR delivery.delivery_policy = $4)
  AND ($5 = '' OR delivery.service_connection_ref = $5)
  AND attempt.status IN ('leased', 'delivered')
  AND attempt.lease_expires_at IS NOT NULL
  AND attempt.lease_expires_at <= $6
ORDER BY attempt.lease_expires_at ASC, attempt.id ASC
LIMIT $7
FOR UPDATE
`, workspaceID, trim(params.ServiceInstallationID), deliverySubscriptionKindFilter(params.SubscriptionKind, params.SubscriptionKinds), trim(params.DeliveryPolicy), trim(params.ServiceConnectionRef), now, limit)
	if err != nil {
		return 0, fmt.Errorf("list expired service event delivery attempts: %w", err)
	}
	defer rows.Close()
	type expiredAttempt struct {
		AttemptID      string
		DeliveryID     string
		LeaseOwner     string
		LeaseExpiresAt time.Time
	}
	expired := make([]expiredAttempt, 0, limit)
	for rows.Next() {
		var item expiredAttempt
		if err := rows.Scan(&item.AttemptID, &item.DeliveryID, &item.LeaseOwner, &item.LeaseExpiresAt); err != nil {
			return 0, fmt.Errorf("scan expired service event delivery attempt: %w", err)
		}
		item.AttemptID = trim(item.AttemptID)
		item.DeliveryID = trim(item.DeliveryID)
		item.LeaseOwner = trim(item.LeaseOwner)
		if item.AttemptID != "" && item.DeliveryID != "" {
			expired = append(expired, item)
		}
	}
	if err := rows.Err(); err != nil {
		return 0, fmt.Errorf("iterate expired service event delivery attempts: %w", err)
	}
	var changed int64
	for _, item := range expired {
		if _, err := setDeliveryAttemptStatus(ctx, execer, DeliveryAttemptStatusUpdate{
			WorkspaceID: workspaceID,
			ID:          item.AttemptID,
			Status:      DeliveryAttemptStatusExpired,
			Now:         now,
		}); err != nil {
			return changed, err
		}
		if err := releaseExpiredDeliveryLeaseIfCurrent(ctx, execer, workspaceID, item.DeliveryID, item.LeaseOwner, item.LeaseExpiresAt, now); err != nil {
			return changed, err
		}
		changed++
	}
	return changed, nil
}

func ensureDeliveryHasNoActiveLeaseAttempts(ctx context.Context, queryer coredb.Queryer, workspaceID string, deliveryID string) error {
	var activeCount int
	if err := queryer.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_event_delivery_attempts
WHERE workspace_id = $1
  AND service_event_delivery_id = $2
  AND status IN ('leased', 'delivered')
`, trim(workspaceID), trim(deliveryID)).Scan(&activeCount); err != nil {
		return fmt.Errorf("count active service event delivery attempts: %w", err)
	}
	if activeCount > 0 {
		return conflict("service event delivery already has an active attempt")
	}
	return nil
}

func releaseExpiredDeliveryLeaseIfCurrent(ctx context.Context, execer coredb.QueryExecer, workspaceID string, deliveryID string, leaseOwner string, leaseExpiresAt time.Time, now time.Time) error {
	if leaseExpiresAt.IsZero() {
		return nil
	}
	if _, err := execer.Exec(ctx, `
UPDATE service_event_deliveries
SET status = CASE WHEN status = 'retry_waiting' THEN status ELSE 'running' END,
    lease_owner = '',
    lease_expires_at = NULL,
    updated_at = $5
WHERE workspace_id = $1
  AND id = $2
  AND lease_owner = $3
  AND lease_expires_at = $4
`, trim(workspaceID), trim(deliveryID), trim(leaseOwner), leaseExpiresAt, now); err != nil {
		return fmt.Errorf("release expired service event delivery lease: %w", err)
	}
	return nil
}

func markDeliveryAttemptDelivered(ctx context.Context, execer coredb.QueryExecer, params DeliveryAttemptDeliveredMark) (DeliveryAttempt, error) {
	workspaceID := trim(params.WorkspaceID)
	attemptID := trim(params.AttemptID)
	if workspaceID == "" || attemptID == "" {
		return DeliveryAttempt{}, badRequest("delivery delivered mark workspace and attempt are required")
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	currentAttempt, err := loadDeliveryAttemptForUpdate(ctx, execer, workspaceID, attemptID)
	if err != nil {
		return DeliveryAttempt{}, err
	}
	if currentAttempt.Status != DeliveryAttemptStatusLeased {
		return DeliveryAttempt{}, conflict("service event delivery attempt is not leased")
	}
	currentDelivery, err := loadEventDeliveryForUpdate(ctx, execer, workspaceID, currentAttempt.ServiceEventDeliveryID)
	if err != nil {
		return DeliveryAttempt{}, err
	}
	if !deliveryLeaseMatchesAttempt(currentDelivery, currentAttempt) {
		return DeliveryAttempt{}, conflict("service event delivery attempt is no longer current")
	}
	return setDeliveryAttemptStatus(ctx, execer, DeliveryAttemptStatusUpdate{
		WorkspaceID:        workspaceID,
		ID:                 attemptID,
		Status:             DeliveryAttemptStatusDelivered,
		ResponseStatusCode: params.ResponseStatusCode,
		ResponseExcerpt:    params.ResponseExcerpt,
		Now:                now,
	})
}

func scheduleDeliveryAttemptRetry(ctx context.Context, execer coredb.QueryExecer, params DeliveryAttemptRetrySchedule) (DeliveryAttemptRetryResult, error) {
	workspaceID := trim(params.WorkspaceID)
	attemptID := trim(params.AttemptID)
	if workspaceID == "" || attemptID == "" {
		return DeliveryAttemptRetryResult{}, badRequest("delivery retry workspace and attempt are required")
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	currentAttempt, err := loadDeliveryAttemptForUpdate(ctx, execer, workspaceID, attemptID)
	if err != nil {
		return DeliveryAttemptRetryResult{}, err
	}
	if currentAttempt.Status != DeliveryAttemptStatusLeased && currentAttempt.Status != DeliveryAttemptStatusDelivered {
		return DeliveryAttemptRetryResult{}, conflict("service event delivery attempt is not retryable")
	}
	currentDelivery, err := loadEventDeliveryForUpdate(ctx, execer, workspaceID, currentAttempt.ServiceEventDeliveryID)
	if err != nil {
		return DeliveryAttemptRetryResult{}, err
	}
	if !deliveryLeaseMatchesAttempt(currentDelivery, currentAttempt) {
		return DeliveryAttemptRetryResult{}, conflict("service event delivery attempt is no longer current")
	}
	backoff := deliveryAttemptRetryBackoff(currentAttempt.AttemptNumber, params.BaseBackoff, params.MaxBackoff)
	nextRetryAt := now.Add(backoff)
	failedAttempt, err := setDeliveryAttemptStatus(ctx, execer, DeliveryAttemptStatusUpdate{
		WorkspaceID:        workspaceID,
		ID:                 attemptID,
		Status:             DeliveryAttemptStatusFailed,
		ResponseStatusCode: params.ResponseStatus,
		ResponseExcerpt:    params.ResponseBody,
		ErrorCode:          firstRuntimeNonEmpty(params.ErrorCode, "transport_failed"),
		ErrorDetail:        params.ErrorDetail,
		Now:                now,
	})
	if err != nil {
		return DeliveryAttemptRetryResult{}, err
	}
	if _, err := execer.Exec(ctx, `
UPDATE service_event_deliveries
SET status = 'retry_waiting',
    next_retry_at = $2,
    error_code = $3,
    error_detail = $4,
    lease_owner = '',
    lease_expires_at = NULL,
    updated_at = $5
WHERE workspace_id = $1
  AND id = $6
`, workspaceID, nextRetryAt, firstRuntimeNonEmpty(params.ErrorCode, "transport_failed"), trim(params.ErrorDetail), now, failedAttempt.ServiceEventDeliveryID); err != nil {
		return DeliveryAttemptRetryResult{}, fmt.Errorf("schedule service event delivery retry: %w", err)
	}
	delivery, err := loadEventDelivery(ctx, execer, failedAttempt.ServiceEventDeliveryID)
	if err != nil {
		return DeliveryAttemptRetryResult{}, err
	}
	if err := markEventSubscriptionFailure(ctx, execer, EventSubscriptionFailureUpdate{
		ID: delivery.EventSubscriptionID, Code: firstRuntimeNonEmpty(params.ErrorCode, "transport_failed"), Detail: params.ErrorDetail, Now: now,
	}); err != nil {
		return DeliveryAttemptRetryResult{}, err
	}
	return DeliveryAttemptRetryResult{Delivery: delivery, Attempt: failedAttempt, NextRetryAt: nextRetryAt}, nil
}

func markDeliveryAttemptLeased(ctx context.Context, execer coredb.QueryExecer, deliveryID string, attempt DeliveryAttempt, leaseOwner string, leaseExpiresAt time.Time, now time.Time) error {
	_, err := execer.Exec(ctx, `
UPDATE service_event_deliveries
SET status = 'running',
    attempted_at = $2,
    attempt_count = CASE WHEN attempt_count < $3 THEN $3 ELSE attempt_count END,
    next_retry_at = NULL,
    lease_owner = $4,
    lease_expires_at = $5,
    error_code = '',
    error_detail = '',
    updated_at = $2
WHERE id = $1
`, trim(deliveryID), now, attempt.AttemptNumber, trim(leaseOwner), leaseExpiresAt)
	if err != nil {
		return fmt.Errorf("mark service event delivery leased: %w", err)
	}
	return nil
}

func markRuntimeSessionClaimed(ctx context.Context, execer coredb.QueryExecer, workspaceID string, sessionID string, credentialID string, installationID string, now time.Time) error {
	result, err := execer.Exec(ctx, `
UPDATE service_runtime_sessions
SET last_claim_at = $3,
    updated_at = $3
WHERE workspace_id = $1
  AND id = $2
  AND service_runtime_credential_id = $4
  AND service_installation_id = $5
  AND status = 'online'
`, trim(workspaceID), trim(sessionID), now, trim(credentialID), trim(installationID))
	if err != nil {
		return fmt.Errorf("mark service runtime session claimed: %w", err)
	}
	if result.RowsAffected() == 0 {
		var exists int
		if err := execer.QueryRow(ctx, `
SELECT 1
FROM service_runtime_sessions
WHERE workspace_id = $1
  AND id = $2
  AND service_runtime_credential_id = $3
  AND service_installation_id = $4
  AND status = 'online'
LIMIT 1
`, trim(workspaceID), trim(sessionID), trim(credentialID), trim(installationID)).Scan(&exists); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return notFound("service runtime session not found")
			}
			return fmt.Errorf("verify service runtime session claim target: %w", err)
		}
	}
	return nil
}

func deliveryLeaseMatchesAttempt(delivery EventDelivery, attempt DeliveryAttempt) bool {
	if trim(delivery.ID) == "" || trim(delivery.ID) != trim(attempt.ServiceEventDeliveryID) {
		return false
	}
	if trim(delivery.LeaseOwner) != trim(attempt.LeaseOwner) {
		return false
	}
	if delivery.LeaseExpiresAt == nil || attempt.LeaseExpiresAt == nil {
		return delivery.LeaseExpiresAt == nil && attempt.LeaseExpiresAt == nil
	}
	return delivery.LeaseExpiresAt.Equal(*attempt.LeaseExpiresAt)
}

func deliveryLeaseOwner(params DeliveryDispatchClaim, transportKind string) string {
	for _, value := range []string{params.LeaseOwner, params.ServiceRuntimeSessionID, params.ServiceRuntimeCredentialID, transportKind} {
		if trimmed := trim(value); trimmed != "" {
			return trimmed
		}
	}
	return ""
}

func normalizeDeliveryDispatchLimit(limit int) int {
	if limit <= 0 {
		return deliveryDispatchDefaultLimit
	}
	if limit > deliveryDispatchMaxLimit {
		return deliveryDispatchMaxLimit
	}
	return limit
}

func normalizeDeliveryLeaseDuration(value time.Duration) time.Duration {
	if value <= 0 {
		return deliveryDispatchDefaultLease
	}
	if value > deliveryDispatchMaxLease {
		return deliveryDispatchMaxLease
	}
	return value
}

func deliverySubscriptionKindFilter(single string, multiple []string) string {
	values := make([]string, 0, len(multiple)+1)
	if value := trim(single); value != "" {
		values = append(values, value)
	}
	values = append(values, multiple...)
	seen := make(map[string]struct{}, len(values))
	normalized := make([]string, 0, len(values))
	for _, value := range values {
		value = trim(value)
		if value == "" {
			continue
		}
		if _, ok := seen[value]; ok {
			continue
		}
		seen[value] = struct{}{}
		normalized = append(normalized, value)
	}
	return strings.Join(normalized, ",")
}

func deliveryAttemptRetryBackoff(attemptNumber int, base time.Duration, max time.Duration) time.Duration {
	if base <= 0 {
		base = deliveryDispatchDefaultRetryBackoff
	}
	if max <= 0 {
		max = deliveryDispatchMaxRetryBackoff
	}
	if max < base {
		max = base
	}
	multiplier := 1
	for i := 1; i < attemptNumber && multiplier < 1024; i++ {
		multiplier *= 2
	}
	delay := time.Duration(multiplier) * base
	if delay > max {
		return max
	}
	return delay
}

func validRuntimeTransportKind(kind string) bool {
	switch trim(kind) {
	case RuntimeTransportKindCallback, RuntimeTransportKindRunnerStream, RuntimeTransportKindInternalHandler:
		return true
	default:
		return false
	}
}

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