package serviceactor

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

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

const (
	// SubscriptionKindPullRequest identifies service event subscriptions for pull request resources.
	SubscriptionKindPullRequest = "pull_request"
	// SubscriptionKindRepositoryPush identifies internal repository push subscriptions.
	SubscriptionKindRepositoryPush = "repository_push"
	// SubscriptionKindAppMention identifies App direct-mention subscriptions bound to AgentTeam targets.
	SubscriptionKindAppMention = "app_mention"
	// SubscriptionKindAppNotificationAction identifies App interactive-card
	// action subscriptions. The Human action is durably captured before a
	// delivery of this kind is reserved.
	SubscriptionKindAppNotificationAction = "app_notification_action"

	// AppRuntimeEventMessageMention is the pinned manifest event required for
	// direct-mention subscriptions and deliveries.
	AppRuntimeEventMessageMention = "agentteam.message.mention"
	// AppRuntimeEventNotificationAction is the pinned manifest event required
	// for interactive notification action subscriptions and deliveries.
	AppRuntimeEventNotificationAction = "agentteam.app.notification.action"

	// EventSubscriptionStatusActive is the normal delivery-ready state.
	EventSubscriptionStatusActive = "active"
	// EventSubscriptionStatusPaused keeps a subscription visible but suppresses delivery.
	EventSubscriptionStatusPaused = "paused"
	// EventSubscriptionStatusBlocked marks a subscription unavailable due to a service connection problem.
	EventSubscriptionStatusBlocked = "blocked"
)

// AppRuntimeSubscriptionKinds returns the closed allowlist of subscription
// kinds accepted by App callback and runner transports.
func AppRuntimeSubscriptionKinds() []string {
	return []string{SubscriptionKindAppMention, SubscriptionKindAppNotificationAction}
}

// IsAppRuntimeSubscriptionKind reports whether a subscription kind belongs to
// the App runtime transport contract.
func IsAppRuntimeSubscriptionKind(value string) bool {
	switch trim(value) {
	case SubscriptionKindAppMention, SubscriptionKindAppNotificationAction:
		return true
	default:
		return false
	}
}

// EventSubscription is the canonical service event subscription record.
type EventSubscription struct {
	ID                     string
	WorkspaceID            string
	ServiceDefinitionID    string
	ServiceInstallationID  string
	ServiceMemberID        string
	ServiceConnectionID    string
	ServiceConnectionRef   string
	ServiceTargetBindingID string
	ServiceKind            string
	SubscriptionKind       string
	TargetKind             string
	TargetRef              string
	DeliveryScopeID        string
	PermissionCheckScopeID string
	OwnerMemberID          string
	AuthorizerMemberID     string
	CreatedByActorType     string
	CreatedByActorID       string
	CreatedVia             string
	ExternalResourceType   string
	ExternalResourceKey    string
	ExternalResourceRef    string
	ExternalResourceURL    string
	EventFilterKey         string
	EventFilters           []byte
	DeliveryPolicy         string
	NotifyPrincipal        bool
	Status                 string
	ActiveSubscriptionKey  string
	LastDeliveryAt         *time.Time
	LastSuccessAt          *time.Time
	LastFailureAt          *time.Time
	LastFailureCode        string
	LastFailureDetail      string
	BlockedAt              *time.Time
	BlockedReason          string
	PausedAt               *time.Time
	PausedReason           string
	RemovedAt              *time.Time
	Metadata               []byte
	CreatedAt              time.Time
	UpdatedAt              time.Time
}

// EventSubscriptionUpsert describes a service event subscription write.
type EventSubscriptionUpsert struct {
	ID                     string
	WorkspaceID            string
	ServiceDefinitionID    string
	ServiceInstallationID  string
	ServiceMemberID        string
	ServiceConnectionID    string
	ServiceConnectionRef   string
	ServiceTargetBindingID string
	ServiceKind            string
	SubscriptionKind       string
	TargetKind             string
	TargetRef              string
	DeliveryScopeID        string
	PermissionCheckScopeID string
	OwnerMemberID          string
	AuthorizerMemberID     string
	CreatedByActorType     string
	CreatedByActorID       string
	CreatedVia             string
	ExternalResourceType   string
	ExternalResourceKey    string
	ExternalResourceRef    string
	ExternalResourceURL    string
	EventFilterKey         string
	EventFilters           []byte
	DeliveryPolicy         string
	NotifyPrincipal        bool
	Status                 string
	ActiveSubscriptionKey  string
	PausedReason           string
	Metadata               []byte
	Now                    time.Time
}

// EventSubscriptionListQuery bounds service event subscription list queries.
type EventSubscriptionListQuery struct {
	WorkspaceID           string
	ServiceDefinitionIDs  []string
	ServiceInstallationID string
	SubscriptionKind      string
	TargetKind            string
	TargetRef             string
	Status                string
	Statuses              []string
	ExternalResourceKey   string
	OwnerMemberID         string
	Limit                 int
}

// EventSubscriptionStatusUpdate describes a status transition.
type EventSubscriptionStatusUpdate struct {
	WorkspaceID string
	ID          string
	Status      string
	Reason      string
	Now         time.Time
}

// EventSubscriptionFailureUpdate describes a failure state update.
type EventSubscriptionFailureUpdate struct {
	ID     string
	Code   string
	Detail string
	Now    time.Time
}

// UpsertEventSubscription creates or reuses an active service event subscription.
func (store *TxStore) UpsertEventSubscription(ctx context.Context, params EventSubscriptionUpsert) (EventSubscription, bool, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return EventSubscription{}, false, err
	}
	return upsertEventSubscription(ctx, execer, params)
}

// ListEventSubscriptions lists service event subscriptions.
func (store *Store) ListEventSubscriptions(ctx context.Context, params EventSubscriptionListQuery) ([]EventSubscription, error) {
	queryer, err := queryerFromStore(store)
	if err != nil {
		return nil, err
	}
	return listEventSubscriptions(ctx, queryer, params, false)
}

// ListEventSubscriptions lists service event subscriptions through an
// existing transaction without locking them. Callers that subsequently touch
// deliveries can preserve the global attempt -> delivery -> subscription lock
// order and lock each subscription only at its final transition.
func (store *TxStore) ListEventSubscriptions(ctx context.Context, params EventSubscriptionListQuery) ([]EventSubscription, error) {
	queryer, err := queryerFromTxStore(store)
	if err != nil {
		return nil, err
	}
	return listEventSubscriptions(ctx, queryer, params, false)
}

// ListEventSubscriptionsForUpdate lists and locks service event subscriptions in a transaction.
func (store *TxStore) ListEventSubscriptionsForUpdate(ctx context.Context, params EventSubscriptionListQuery) ([]EventSubscription, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return nil, err
	}
	return listEventSubscriptions(ctx, execer, params, true)
}

// LoadEventSubscription loads one service event subscription by id.
func (store *Store) LoadEventSubscription(ctx context.Context, workspaceID string, subscriptionID string) (EventSubscription, error) {
	queryer, err := queryerFromStore(store)
	if err != nil {
		return EventSubscription{}, err
	}
	return loadEventSubscription(ctx, queryer, workspaceID, subscriptionID)
}

// LoadEventSubscriptionForUpdate locks one service event subscription by id.
func (store *TxStore) LoadEventSubscriptionForUpdate(ctx context.Context, workspaceID string, subscriptionID string) (EventSubscription, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return EventSubscription{}, err
	}
	return loadEventSubscriptionForUpdate(ctx, execer, workspaceID, subscriptionID)
}

// SetEventSubscriptionStatus updates active/paused/blocked state.
func (store *TxStore) SetEventSubscriptionStatus(ctx context.Context, params EventSubscriptionStatusUpdate) (EventSubscription, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return EventSubscription{}, err
	}
	return setEventSubscriptionStatus(ctx, execer, params)
}

// RemoveEventSubscription soft-removes a subscription and clears its active key.
func (store *TxStore) RemoveEventSubscription(ctx context.Context, workspaceID string, subscriptionID string, now time.Time) (EventSubscription, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return EventSubscription{}, err
	}
	return removeEventSubscription(ctx, execer, workspaceID, subscriptionID, now)
}

// MarkEventSubscriptionsDelivered updates success/failure counters after a delivery effect.
func (store *TxStore) MarkEventSubscriptionsDelivered(ctx context.Context, subscriptionIDs []string, now time.Time) error {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return err
	}
	return markEventSubscriptionsDelivered(ctx, execer, subscriptionIDs, now)
}

// MarkEventSubscriptionFailure records the latest delivery failure.
func (store *TxStore) MarkEventSubscriptionFailure(ctx context.Context, params EventSubscriptionFailureUpdate) error {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return err
	}
	return markEventSubscriptionFailure(ctx, execer, params)
}

func upsertEventSubscription(ctx context.Context, execer coredb.QueryExecer, params EventSubscriptionUpsert) (EventSubscription, bool, error) {
	workspaceID := trim(params.WorkspaceID)
	definitionID := trim(params.ServiceDefinitionID)
	installationID := trim(params.ServiceInstallationID)
	serviceMemberID := trim(params.ServiceMemberID)
	connectionRef := trim(params.ServiceConnectionRef)
	bindingID := trim(params.ServiceTargetBindingID)
	serviceKind := trim(params.ServiceKind)
	subscriptionKind := trim(params.SubscriptionKind)
	targetKind := trim(params.TargetKind)
	targetRef := trim(params.TargetRef)
	deliveryScopeID := trim(params.DeliveryScopeID)
	permissionScopeID := trim(params.PermissionCheckScopeID)
	ownerMemberID := trim(params.OwnerMemberID)
	authorizerMemberID := trim(params.AuthorizerMemberID)
	createdByActorType := trim(params.CreatedByActorType)
	createdByActorID := trim(params.CreatedByActorID)
	createdVia := trim(params.CreatedVia)
	externalResourceType := trim(params.ExternalResourceType)
	externalResourceKey := trim(params.ExternalResourceKey)
	eventFilterKey := trim(params.EventFilterKey)
	activeKey := trim(params.ActiveSubscriptionKey)
	if workspaceID == "" || definitionID == "" || installationID == "" || serviceMemberID == "" || bindingID == "" ||
		connectionRef == "" || serviceKind == "" || subscriptionKind == "" || targetKind == "" || targetRef == "" || deliveryScopeID == "" ||
		permissionScopeID == "" || ownerMemberID == "" || authorizerMemberID == "" || createdByActorType == "" ||
		createdByActorID == "" || createdVia == "" || externalResourceType == "" || externalResourceKey == "" ||
		eventFilterKey == "" || activeKey == "" {
		return EventSubscription{}, false, badRequest("service event subscription identity, target, owner, resource, filter, and active key are required")
	}
	if existing, found, err := loadEventSubscriptionByActiveKeyForUpdate(ctx, execer, activeKey); err != nil {
		return EventSubscription{}, false, err
	} else if found {
		return existing, true, nil
	}

	subscriptionID := trim(params.ID)
	var err error
	if subscriptionID == "" {
		subscriptionID, err = identifiers.NewEntityID("svcesub")
		if err != nil {
			return EventSubscription{}, false, fmt.Errorf("generate service event subscription id: %w", err)
		}
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	status := trim(params.Status)
	if status == "" {
		status = EventSubscriptionStatusActive
	}
	pausedAt := any(nil)
	if status == EventSubscriptionStatusPaused {
		pausedAt = now
	}
	deliveryPolicy := trim(params.DeliveryPolicy)
	if deliveryPolicy == "" {
		deliveryPolicy = "default"
	}
	eventFilters := params.EventFilters
	if len(eventFilters) == 0 {
		eventFilters = []byte(`{}`)
	}
	_, err = execer.Exec(ctx, `
INSERT INTO service_event_subscriptions (
    id, workspace_id, service_definition_id, service_installation_id,
    service_member_id, service_connection_id, service_connection_ref,
    service_target_binding_id,
    service_kind, subscription_kind, target_kind, target_ref,
    delivery_scope_id, permission_check_scope_id, owner_member_id,
    authorizer_member_id, created_by_actor_type, created_by_actor_id,
    created_via, external_resource_type, external_resource_key,
    external_repository_key, external_resource_ref, external_resource_url, event_filter_key,
    event_filters, delivery_policy, notify_principal, status,
    active_subscription_key, last_failure_detail, paused_at, paused_reason, metadata,
    base_branch_key, target_branch_key, head_sha_key, base_sha_key,
    created_at, updated_at
) VALUES (
    $1, $2, $3, $4,
    $5, $6, $7,
    $8,
    $9, $10, $11, $12,
    $13, $14, $15,
    $16, $17, $18,
    $19, $20, $21,
    LOWER(SUBSTRING_INDEX($21, '#', 1)), $22, $23, $24,
    $25, $26, $27, $28,
    $29, '', $30, $31, $32,
    LOWER(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(CASE WHEN JSON_VALID($32) THEN $32 ELSE JSON_OBJECT() END, '$.base_branch')), '')),
    LOWER(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(CASE WHEN JSON_VALID($32) THEN $32 ELSE JSON_OBJECT() END, '$.target_branch')), '')),
    LOWER(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(CASE WHEN JSON_VALID($32) THEN $32 ELSE JSON_OBJECT() END, '$.head_sha')), '')),
    LOWER(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(CASE WHEN JSON_VALID($32) THEN $32 ELSE JSON_OBJECT() END, '$.base_sha')), '')),
    $33, $33
)
`, subscriptionID, workspaceID, definitionID, installationID,
		serviceMemberID, nullableString(trim(params.ServiceConnectionID)), connectionRef,
		bindingID,
		serviceKind, subscriptionKind, targetKind, targetRef,
		deliveryScopeID, permissionScopeID, ownerMemberID,
		authorizerMemberID, createdByActorType, createdByActorID,
		createdVia, externalResourceType, externalResourceKey,
		trim(params.ExternalResourceRef), trim(params.ExternalResourceURL), eventFilterKey,
		string(eventFilters), deliveryPolicy, params.NotifyPrincipal, status,
		activeKey, pausedAt, trim(params.PausedReason), nullableString(string(params.Metadata)),
		now)
	if err != nil {
		if coredb.IsUniqueViolation(err) {
			if existing, found, loadErr := loadEventSubscriptionByActiveKeyForUpdate(ctx, execer, activeKey); loadErr != nil {
				return EventSubscription{}, false, loadErr
			} else if found {
				return existing, true, nil
			}
		}
		return EventSubscription{}, false, fmt.Errorf("insert service event subscription: %w", err)
	}
	subscription, err := loadEventSubscription(ctx, execer, workspaceID, subscriptionID)
	return subscription, false, err
}

func loadEventSubscriptionByActiveKeyForUpdate(ctx context.Context, queryer coredb.QueryRower, activeKey string) (EventSubscription, bool, error) {
	row := queryer.QueryRow(ctx, eventSubscriptionSelectSQL()+`
WHERE active_subscription_key = $1
  AND removed_at IS NULL
LIMIT 1
FOR UPDATE
`, trim(activeKey))
	subscription, err := scanEventSubscription(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return EventSubscription{}, false, nil
		}
		return EventSubscription{}, false, fmt.Errorf("load service event subscription by active key: %w", err)
	}
	return subscription, true, nil
}

func loadEventSubscription(ctx context.Context, queryer coredb.QueryRower, workspaceID string, subscriptionID string) (EventSubscription, error) {
	row := queryer.QueryRow(ctx, eventSubscriptionSelectSQL()+`
WHERE workspace_id = $1
  AND id = $2
`, trim(workspaceID), trim(subscriptionID))
	subscription, err := scanEventSubscription(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return EventSubscription{}, notFound("service event subscription not found")
		}
		return EventSubscription{}, fmt.Errorf("load service event subscription: %w", err)
	}
	return subscription, nil
}

func loadEventSubscriptionForUpdate(ctx context.Context, queryer coredb.QueryRower, workspaceID string, subscriptionID string) (EventSubscription, error) {
	row := queryer.QueryRow(ctx, eventSubscriptionSelectSQL()+`
WHERE workspace_id = $1
  AND id = $2
  AND removed_at IS NULL
FOR UPDATE
`, trim(workspaceID), trim(subscriptionID))
	subscription, err := scanEventSubscription(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return EventSubscription{}, notFound("service event subscription not found")
		}
		return EventSubscription{}, fmt.Errorf("load service event subscription for update: %w", err)
	}
	return subscription, nil
}

func listEventSubscriptions(ctx context.Context, queryer coredb.Queryer, params EventSubscriptionListQuery, forUpdate bool) ([]EventSubscription, error) {
	workspaceID := trim(params.WorkspaceID)
	if workspaceID == "" {
		return nil, badRequest("workspaceId is required")
	}
	conditions := []string{"workspace_id = $1", "removed_at IS NULL"}
	args := []any{workspaceID}
	if len(params.ServiceDefinitionIDs) > 0 {
		conditions = append(conditions, stringInCondition("service_definition_id", params.ServiceDefinitionIDs, &args))
	}
	if trim(params.ServiceInstallationID) != "" {
		args = append(args, trim(params.ServiceInstallationID))
		conditions = append(conditions, fmt.Sprintf("service_installation_id = $%d", len(args)))
	}
	if trim(params.SubscriptionKind) != "" {
		args = append(args, trim(params.SubscriptionKind))
		conditions = append(conditions, fmt.Sprintf("subscription_kind = $%d", len(args)))
	}
	if trim(params.TargetKind) != "" {
		args = append(args, trim(params.TargetKind))
		conditions = append(conditions, fmt.Sprintf("target_kind = $%d", len(args)))
	}
	if trim(params.TargetRef) != "" {
		args = append(args, trim(params.TargetRef))
		conditions = append(conditions, fmt.Sprintf("target_ref = $%d", len(args)))
	}
	if trim(params.Status) != "" {
		args = append(args, trim(params.Status))
		conditions = append(conditions, fmt.Sprintf("status = $%d", len(args)))
	} else if len(params.Statuses) > 0 {
		conditions = append(conditions, stringInCondition("status", params.Statuses, &args))
	}
	if trim(params.ExternalResourceKey) != "" {
		args = append(args, trim(params.ExternalResourceKey))
		conditions = append(conditions, fmt.Sprintf("external_resource_key = $%d", len(args)))
	}
	if trim(params.OwnerMemberID) != "" {
		args = append(args, trim(params.OwnerMemberID))
		conditions = append(conditions, fmt.Sprintf("owner_member_id = $%d", len(args)))
	}
	query := eventSubscriptionSelectSQL() + `
WHERE ` + strings.Join(conditions, " AND ") + `
ORDER BY updated_at DESC, id ASC
`
	if params.Limit > 0 {
		args = append(args, params.Limit)
		query += fmt.Sprintf("LIMIT $%d\n", len(args))
	}
	if forUpdate {
		query += "FOR UPDATE\n"
	}
	rows, err := queryer.Query(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("list service event subscriptions: %w", err)
	}
	defer rows.Close()
	subscriptions := make([]EventSubscription, 0)
	for rows.Next() {
		subscription, err := scanEventSubscription(rows)
		if err != nil {
			return nil, fmt.Errorf("scan service event subscription: %w", err)
		}
		subscriptions = append(subscriptions, subscription)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate service event subscriptions: %w", err)
	}
	return subscriptions, nil
}

func setEventSubscriptionStatus(ctx context.Context, execer coredb.QueryExecer, params EventSubscriptionStatusUpdate) (EventSubscription, error) {
	workspaceID := trim(params.WorkspaceID)
	subscriptionID := trim(params.ID)
	status := trim(params.Status)
	if workspaceID == "" || subscriptionID == "" || status == "" {
		return EventSubscription{}, badRequest("workspaceId, subscriptionId, and status are required")
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	var pausedAt any
	if status == EventSubscriptionStatusPaused {
		pausedAt = now
	}
	var blockedAt any
	blockedReason := ""
	if status == EventSubscriptionStatusBlocked {
		blockedAt = now
		blockedReason = trim(params.Reason)
	}
	if _, err := execer.Exec(ctx, `
UPDATE service_event_subscriptions
SET status = $3,
    paused_reason = $4,
    paused_at = $5,
    blocked_reason = $6,
    blocked_at = $7,
    updated_at = $8
WHERE workspace_id = $1
  AND id = $2
  AND removed_at IS NULL
`, workspaceID, subscriptionID, status, trim(params.Reason), pausedAt, blockedReason, blockedAt, now); err != nil {
		return EventSubscription{}, fmt.Errorf("update service event subscription status: %w", err)
	}
	return loadEventSubscriptionForUpdate(ctx, execer, workspaceID, subscriptionID)
}

func removeEventSubscription(ctx context.Context, execer coredb.QueryExecer, workspaceID string, subscriptionID string, now time.Time) (EventSubscription, error) {
	trimmedWorkspaceID := trim(workspaceID)
	trimmedID := trim(subscriptionID)
	if trimmedWorkspaceID == "" || trimmedID == "" {
		return EventSubscription{}, badRequest("workspaceId and subscriptionId are required")
	}
	if now.IsZero() {
		now = time.Now().UTC()
	}
	subscription, err := loadEventSubscriptionForUpdate(ctx, execer, trimmedWorkspaceID, trimmedID)
	if err != nil {
		return EventSubscription{}, err
	}
	if _, err := execer.Exec(ctx, `
UPDATE service_event_subscriptions
SET active_subscription_key = NULL,
    removed_at = $3,
    updated_at = $3
WHERE workspace_id = $1
  AND id = $2
  AND removed_at IS NULL
`, trimmedWorkspaceID, trimmedID, now); err != nil {
		return EventSubscription{}, fmt.Errorf("remove service event subscription: %w", err)
	}
	subscription.ActiveSubscriptionKey = ""
	subscription.RemovedAt = &now
	subscription.UpdatedAt = now
	return subscription, nil
}

func markEventSubscriptionsDelivered(ctx context.Context, execer coredb.QueryExecer, subscriptionIDs []string, now time.Time) error {
	ids := trimSlice(subscriptionIDs)
	if len(ids) == 0 {
		return nil
	}
	if now.IsZero() {
		now = time.Now().UTC()
	}
	requested, args, _, err := coredb.BuildStringValuesCTE("delivered_service_event_subscriptions", "subscription_id", ids, 2)
	if err != nil {
		return fmt.Errorf("build delivered service event subscription ids: %w", err)
	}
	query := fmt.Sprintf(`
WITH %s
UPDATE service_event_subscriptions
SET last_delivery_at = $1,
    last_success_at = $1,
    last_failure_at = NULL,
    last_failure_code = '',
    last_failure_detail = '',
    updated_at = $1
WHERE EXISTS (
      SELECT 1
      FROM delivered_service_event_subscriptions delivered
      WHERE delivered.subscription_id = id
)
`, requested)
	_, err = execer.Exec(ctx, query, append([]any{now}, args...)...)
	if err != nil {
		return fmt.Errorf("mark service event subscriptions delivered: %w", err)
	}
	return nil
}

func markEventSubscriptionFailure(ctx context.Context, execer coredb.QueryExecer, params EventSubscriptionFailureUpdate) error {
	subscriptionID := trim(params.ID)
	if subscriptionID == "" {
		return nil
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	detail := trim(params.Detail)
	if len(detail) > 4096 {
		detail = detail[:4096]
	}
	_, err := execer.Exec(ctx, `
UPDATE service_event_subscriptions
SET last_failure_at = $1,
    last_failure_code = $2,
    last_failure_detail = $3,
    updated_at = $1
WHERE id = $4
  AND removed_at IS NULL
`, now, trim(params.Code), detail, subscriptionID)
	if err != nil {
		return fmt.Errorf("mark service event subscription failure: %w", err)
	}
	return nil
}

func eventSubscriptionSelectSQL() string {
	return `
SELECT id, workspace_id, service_definition_id, service_installation_id,
       service_member_id, COALESCE(service_connection_id, ''), service_connection_ref,
       service_target_binding_id,
       service_kind, subscription_kind, target_kind, target_ref,
       delivery_scope_id, permission_check_scope_id, owner_member_id,
       authorizer_member_id, created_by_actor_type, created_by_actor_id,
       created_via, external_resource_type, external_resource_key,
       external_resource_ref, external_resource_url, event_filter_key, event_filters,
       delivery_policy, notify_principal, status, COALESCE(active_subscription_key, ''),
       last_delivery_at, last_success_at, last_failure_at,
       COALESCE(last_failure_code, ''), COALESCE(last_failure_detail, ''),
       blocked_at, COALESCE(blocked_reason, ''), paused_at, COALESCE(paused_reason, ''),
       removed_at, metadata, created_at, updated_at
FROM service_event_subscriptions`
}

func scanEventSubscription(row coredb.Row) (EventSubscription, error) {
	var subscription EventSubscription
	var lastDeliveryAt, lastSuccessAt, lastFailureAt sql.NullTime
	var blockedAt, pausedAt, removedAt sql.NullTime
	var metadata []byte
	if err := row.Scan(
		&subscription.ID,
		&subscription.WorkspaceID,
		&subscription.ServiceDefinitionID,
		&subscription.ServiceInstallationID,
		&subscription.ServiceMemberID,
		&subscription.ServiceConnectionID,
		&subscription.ServiceConnectionRef,
		&subscription.ServiceTargetBindingID,
		&subscription.ServiceKind,
		&subscription.SubscriptionKind,
		&subscription.TargetKind,
		&subscription.TargetRef,
		&subscription.DeliveryScopeID,
		&subscription.PermissionCheckScopeID,
		&subscription.OwnerMemberID,
		&subscription.AuthorizerMemberID,
		&subscription.CreatedByActorType,
		&subscription.CreatedByActorID,
		&subscription.CreatedVia,
		&subscription.ExternalResourceType,
		&subscription.ExternalResourceKey,
		&subscription.ExternalResourceRef,
		&subscription.ExternalResourceURL,
		&subscription.EventFilterKey,
		&subscription.EventFilters,
		&subscription.DeliveryPolicy,
		&subscription.NotifyPrincipal,
		&subscription.Status,
		&subscription.ActiveSubscriptionKey,
		&lastDeliveryAt,
		&lastSuccessAt,
		&lastFailureAt,
		&subscription.LastFailureCode,
		&subscription.LastFailureDetail,
		&blockedAt,
		&subscription.BlockedReason,
		&pausedAt,
		&subscription.PausedReason,
		&removedAt,
		&metadata,
		&subscription.CreatedAt,
		&subscription.UpdatedAt,
	); err != nil {
		return EventSubscription{}, err
	}
	subscription.ID = trim(subscription.ID)
	subscription.WorkspaceID = trim(subscription.WorkspaceID)
	subscription.ServiceDefinitionID = trim(subscription.ServiceDefinitionID)
	subscription.ServiceInstallationID = trim(subscription.ServiceInstallationID)
	subscription.ServiceMemberID = trim(subscription.ServiceMemberID)
	subscription.ServiceConnectionID = trim(subscription.ServiceConnectionID)
	subscription.ServiceConnectionRef = trim(subscription.ServiceConnectionRef)
	subscription.ServiceTargetBindingID = trim(subscription.ServiceTargetBindingID)
	subscription.ServiceKind = trim(subscription.ServiceKind)
	subscription.SubscriptionKind = trim(subscription.SubscriptionKind)
	subscription.Status = trim(subscription.Status)
	subscription.LastDeliveryAt = ptrFromNullTime(lastDeliveryAt)
	subscription.LastSuccessAt = ptrFromNullTime(lastSuccessAt)
	subscription.LastFailureAt = ptrFromNullTime(lastFailureAt)
	subscription.BlockedAt = ptrFromNullTime(blockedAt)
	subscription.PausedAt = ptrFromNullTime(pausedAt)
	subscription.RemovedAt = ptrFromNullTime(removedAt)
	subscription.Metadata = metadata
	if len(subscription.EventFilters) == 0 {
		subscription.EventFilters = []byte(`{}`)
	}
	return subscription, nil
}

func stringInCondition(column string, values []string, args *[]any) string {
	placeholders := make([]string, 0, len(values))
	for _, value := range trimSlice(values) {
		*args = append(*args, value)
		placeholders = append(placeholders, fmt.Sprintf("$%d", len(*args)))
	}
	if len(placeholders) == 0 {
		return "1 = 0"
	}
	return fmt.Sprintf("%s IN (%s)", column, strings.Join(placeholders, ", "))
}

func nullableBytes(value []byte) any {
	if len(value) == 0 {
		return nil
	}
	return value
}