package serviceactor

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

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

const (
	RuntimeCredentialKindRunnerStreamToken = "runner_stream_token"

	RuntimeTransportKindCallback        = "callback"
	RuntimeTransportKindRunnerStream    = "runner_stream"
	RuntimeTransportKindInternalHandler = "internal_handler"

	RuntimeCredentialStatusActive   = StatusActive
	RuntimeCredentialStatusDisabled = StatusDisabled
	RuntimeCredentialStatusRotated  = "rotated"
	RuntimeCredentialStatusExpired  = "expired"

	RuntimeSessionStatusOnline   = "online"
	RuntimeSessionStatusStale    = "stale"
	RuntimeSessionStatusOffline  = "offline"
	RuntimeSessionStatusDisabled = StatusDisabled

	DeliveryAttemptStatusLeased    = "leased"
	DeliveryAttemptStatusDelivered = "delivered"
	DeliveryAttemptStatusAcked     = "acked"
	DeliveryAttemptStatusFailed    = "failed"
	DeliveryAttemptStatusExpired   = "expired"
	DeliveryAttemptStatusCanceled  = "canceled"

	DeliveryEffectStatusCreated  = "created"
	DeliveryEffectStatusFailed   = "failed"
	DeliveryEffectStatusReverted = "reverted"

	runtimeRetentionDeleteBatchDefault = 500
	runtimeRetentionDeleteBatchMax     = 1000
	runtimeListLimitDefault            = 100
	runtimeListLimitMax                = 500
)

// RuntimeCredential stores hash-only credentials that let a service runtime
// authenticate to AgentTeam runtime APIs.
type RuntimeCredential struct {
	ID                    string
	WorkspaceID           string
	ServiceDefinitionID   string
	ServiceInstallationID string
	ServiceMemberID       string
	ServiceKind           string
	CredentialKind        string
	TransportKind         string
	Name                  string
	TokenHash             string
	TokenHint             string
	Status                string
	CreatedByMemberID     string
	CreatedAt             time.Time
	UpdatedAt             time.Time
	LastUsedAt            *time.Time
	DisabledAt            *time.Time
	ExpiresAt             *time.Time
	Metadata              []byte
}

// RuntimeCredentialCreate describes a runtime credential row. The caller passes
// a token hash; CreateRuntimeCredentialToken can generate a token and hash for
// user-facing token setup flows.
type RuntimeCredentialCreate struct {
	ID                    string
	WorkspaceID           string
	ServiceDefinitionID   string
	ServiceInstallationID string
	ServiceMemberID       string
	ServiceKind           string
	CredentialKind        string
	TransportKind         string
	Name                  string
	TokenHash             string
	TokenHint             string
	Status                string
	CreatedByMemberID     string
	Now                   time.Time
	ExpiresAt             *time.Time
	Metadata              []byte
}

// RuntimeCredentialTokenCreate describes a runtime credential creation that
// returns the plaintext token exactly once to the caller.
type RuntimeCredentialTokenCreate struct {
	RuntimeCredentialCreate
	TokenByteLength int
}

// RuntimeCredentialToken is the one-time plaintext token plus persisted row.
type RuntimeCredentialToken struct {
	Credential RuntimeCredential
	Token      string
}

// RuntimeSession records one connected runner runtime instance.
type RuntimeSession struct {
	ID                         string
	WorkspaceID                string
	ServiceDefinitionID        string
	ServiceInstallationID      string
	ServiceMemberID            string
	ServiceRuntimeCredentialID string
	RunnerID                   string
	RunnerVersion              string
	Status                     string
	CapabilitiesJSON           []byte
	LastHeartbeatAt            *time.Time
	LastClaimAt                *time.Time
	CreatedAt                  time.Time
	UpdatedAt                  time.Time
	Metadata                   []byte
}

// RuntimeSessionUpsert describes a runner session heartbeat/update.
type RuntimeSessionUpsert struct {
	ID                         string
	WorkspaceID                string
	ServiceDefinitionID        string
	ServiceInstallationID      string
	ServiceMemberID            string
	ServiceRuntimeCredentialID string
	RunnerID                   string
	RunnerVersion              string
	Status                     string
	CapabilitiesJSON           []byte
	LastHeartbeatAt            *time.Time
	LastClaimAt                *time.Time
	Now                        time.Time
	Metadata                   []byte
}

// DeliveryAttempt records one delivery lease/transport attempt.
type DeliveryAttempt struct {
	ID                         string
	WorkspaceID                string
	ServiceEventDeliveryID     string
	ServiceInstallationID      string
	ServiceMemberID            string
	TransportKind              string
	ServiceRuntimeCredentialID string
	ServiceRuntimeSessionID    string
	AttemptNumber              int
	Status                     string
	LeaseOwner                 string
	LeaseExpiresAt             *time.Time
	DeliveryTokenHash          string
	DeliveryTokenExpiresAt     *time.Time
	CallbackURLSnapshot        string
	ResponseStatusCode         *int
	ResponseExcerpt            string
	ClaimedAt                  *time.Time
	DeliveredAt                *time.Time
	AckedAt                    *time.Time
	FailedAt                   *time.Time
	ExpiredAt                  *time.Time
	ErrorCode                  string
	ErrorDetail                string
	ResultJSON                 []byte
	CreatedAt                  time.Time
	UpdatedAt                  time.Time
}

// DeliveryAttemptCreate describes a new delivery attempt. When AttemptNumber is
// zero, the store allocates the next number while holding the delivery row lock.
type DeliveryAttemptCreate struct {
	ID                         string
	WorkspaceID                string
	ServiceEventDeliveryID     string
	TransportKind              string
	ServiceRuntimeCredentialID string
	ServiceRuntimeSessionID    string
	AttemptNumber              int
	Status                     string
	LeaseOwner                 string
	LeaseExpiresAt             *time.Time
	DeliveryTokenHash          string
	DeliveryTokenExpiresAt     *time.Time
	CallbackURLSnapshot        string
	ResponseStatusCode         *int
	ResponseExcerpt            string
	ClaimedAt                  *time.Time
	DeliveredAt                *time.Time
	Now                        time.Time
}

// DeliveryAttemptStatusUpdate describes an attempt lifecycle transition.
type DeliveryAttemptStatusUpdate struct {
	WorkspaceID        string
	ID                 string
	Status             string
	ResponseStatusCode *int
	ResponseExcerpt    string
	ErrorCode          string
	ErrorDetail        string
	ResultJSON         []byte
	Now                time.Time
}

// DeliveryAttemptDeliveredMark marks the current leased transport attempt as
// delivered after a transport envelope reaches the runtime endpoint.
type DeliveryAttemptDeliveredMark struct {
	WorkspaceID        string
	AttemptID          string
	ResponseStatusCode *int
	ResponseExcerpt    string
	Now                time.Time
}

// DeliveryAttemptCandidateQuery selects deliveries ready for transport attempt
// creation.
type DeliveryAttemptCandidateQuery struct {
	WorkspaceID           string
	ServiceInstallationID string
	SubscriptionKind      string
	Now                   time.Time
	Limit                 int
}

// DeliveryEffect records one AgentTeam-visible or domain effect for a delivery.
type DeliveryEffect struct {
	ID                     string
	WorkspaceID            string
	ServiceEventDeliveryID string
	DeliveryAttemptID      string
	ServiceInstallationID  string
	ServiceMemberID        string
	EffectKind             string
	EffectRef              string
	TargetKind             string
	TargetRef              string
	MessageID              string
	Status                 string
	Metadata               []byte
	CreatedAt              time.Time
	UpdatedAt              time.Time
}

// DeliveryEffectAppend describes a delivery effect write.
type DeliveryEffectAppend struct {
	ID                string
	WorkspaceID       string
	DeliveryAttemptID string
	EffectKind        string
	EffectRef         string
	TargetKind        string
	TargetRef         string
	MessageID         string
	Status            string
	Metadata          []byte
	Now               time.Time
}

// DeliveryEffectListQuery selects delivery effects for diagnostics, replay, and
// UI activity.
type DeliveryEffectListQuery struct {
	WorkspaceID            string
	ServiceEventDeliveryID string
	Limit                  int
}

// RuntimeRetentionFilter bounds cleanup of runtime ledger rows.
type RuntimeRetentionFilter struct {
	Status string
	Cutoff time.Time
	Limit  int
}

// CreateRuntimeCredential inserts a hash-only runtime credential row.
func (store *Store) CreateRuntimeCredential(ctx context.Context, params RuntimeCredentialCreate) (RuntimeCredential, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return RuntimeCredential{}, err
	}
	return createRuntimeCredential(ctx, execer, params)
}

// CreateRuntimeCredential creates a runtime credential row in a transaction.
func (store *TxStore) CreateRuntimeCredential(ctx context.Context, params RuntimeCredentialCreate) (RuntimeCredential, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return RuntimeCredential{}, err
	}
	return createRuntimeCredential(ctx, execer, params)
}

// CreateRuntimeCredentialToken creates a runtime token and persists only its hash.
func (store *Store) CreateRuntimeCredentialToken(ctx context.Context, params RuntimeCredentialTokenCreate) (RuntimeCredentialToken, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return RuntimeCredentialToken{}, err
	}
	return createRuntimeCredentialToken(ctx, execer, params)
}

// CreateRuntimeCredentialToken creates a runtime token in a transaction and persists only its hash.
func (store *TxStore) CreateRuntimeCredentialToken(ctx context.Context, params RuntimeCredentialTokenCreate) (RuntimeCredentialToken, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return RuntimeCredentialToken{}, err
	}
	return createRuntimeCredentialToken(ctx, execer, params)
}

// AuthenticateRuntimeCredential loads an active credential by bearer token and updates last_used_at.
func (store *Store) AuthenticateRuntimeCredential(ctx context.Context, token string) (RuntimeCredential, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return RuntimeCredential{}, err
	}
	return authenticateRuntimeCredential(ctx, execer, token, time.Now().UTC())
}

// AuthenticateRuntimeCredential loads an active credential by bearer token in a
// transaction and updates last_used_at.
func (store *TxStore) AuthenticateRuntimeCredential(ctx context.Context, token string) (RuntimeCredential, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return RuntimeCredential{}, err
	}
	return authenticateRuntimeCredential(ctx, execer, token, time.Now().UTC())
}

// LoadRuntimeCredential loads one runtime credential by id.
func (store *Store) LoadRuntimeCredential(ctx context.Context, workspaceID string, credentialID string) (RuntimeCredential, error) {
	queryer, err := queryerFromStore(store)
	if err != nil {
		return RuntimeCredential{}, err
	}
	return loadRuntimeCredential(ctx, queryer, workspaceID, credentialID)
}

// LoadRuntimeCredential loads one runtime credential by id in a transaction.
func (store *TxStore) LoadRuntimeCredential(ctx context.Context, workspaceID string, credentialID string) (RuntimeCredential, error) {
	queryer, err := queryerFromTxStore(store)
	if err != nil {
		return RuntimeCredential{}, err
	}
	return loadRuntimeCredential(ctx, queryer, workspaceID, credentialID)
}

// DisableRuntimeCredential disables one runtime credential.
func (store *Store) DisableRuntimeCredential(ctx context.Context, workspaceID string, credentialID string, now time.Time) (RuntimeCredential, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return RuntimeCredential{}, err
	}
	return setRuntimeCredentialStatus(ctx, execer, workspaceID, credentialID, RuntimeCredentialStatusDisabled, now)
}

// DisableRuntimeCredential disables one runtime credential in a transaction.
func (store *TxStore) DisableRuntimeCredential(ctx context.Context, workspaceID string, credentialID string, now time.Time) (RuntimeCredential, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return RuntimeCredential{}, err
	}
	return setRuntimeCredentialStatus(ctx, execer, workspaceID, credentialID, RuntimeCredentialStatusDisabled, now)
}

// UpsertRuntimeSession records a runner session heartbeat/update.
func (store *Store) UpsertRuntimeSession(ctx context.Context, params RuntimeSessionUpsert) (RuntimeSession, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return RuntimeSession{}, err
	}
	return upsertRuntimeSession(ctx, execer, params)
}

// UpsertRuntimeSession records a runner session heartbeat/update in a transaction.
func (store *TxStore) UpsertRuntimeSession(ctx context.Context, params RuntimeSessionUpsert) (RuntimeSession, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return RuntimeSession{}, err
	}
	return upsertRuntimeSession(ctx, execer, params)
}

// LoadRuntimeSession loads one runner session by id.
func (store *Store) LoadRuntimeSession(ctx context.Context, workspaceID string, sessionID string) (RuntimeSession, error) {
	queryer, err := queryerFromStore(store)
	if err != nil {
		return RuntimeSession{}, err
	}
	return loadRuntimeSession(ctx, queryer, workspaceID, sessionID)
}

// LoadRuntimeSession loads one runner session by id in a transaction.
func (store *TxStore) LoadRuntimeSession(ctx context.Context, workspaceID string, sessionID string) (RuntimeSession, error) {
	queryer, err := queryerFromTxStore(store)
	if err != nil {
		return RuntimeSession{}, err
	}
	return loadRuntimeSession(ctx, queryer, workspaceID, sessionID)
}

// SetRuntimeSessionStatus updates one runner session status.
func (store *Store) SetRuntimeSessionStatus(ctx context.Context, workspaceID string, sessionID string, status string, now time.Time) (RuntimeSession, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return RuntimeSession{}, err
	}
	return setRuntimeSessionStatus(ctx, execer, workspaceID, sessionID, status, now)
}

// SetRuntimeSessionStatus updates one runner session status in a transaction.
func (store *TxStore) SetRuntimeSessionStatus(ctx context.Context, workspaceID string, sessionID string, status string, now time.Time) (RuntimeSession, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return RuntimeSession{}, err
	}
	return setRuntimeSessionStatus(ctx, execer, workspaceID, sessionID, status, now)
}

// SetRuntimeSessionsStatusByCredential updates runner sessions that are bound
// to one runtime credential.
func (store *Store) SetRuntimeSessionsStatusByCredential(ctx context.Context, workspaceID string, credentialID string, status string, now time.Time) (int64, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return 0, err
	}
	return setRuntimeSessionsStatusByCredential(ctx, execer, workspaceID, credentialID, status, now)
}

// SetRuntimeSessionsStatusByCredential updates runner sessions that are bound
// to one runtime credential in a transaction.
func (store *TxStore) SetRuntimeSessionsStatusByCredential(ctx context.Context, workspaceID string, credentialID string, status string, now time.Time) (int64, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return 0, err
	}
	return setRuntimeSessionsStatusByCredential(ctx, execer, workspaceID, credentialID, status, now)
}

// CreateDeliveryAttempt creates a delivery attempt.
func (store *Store) CreateDeliveryAttempt(ctx context.Context, params DeliveryAttemptCreate) (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 = createDeliveryAttempt(ctx, tx, params)
		return err
	}); err != nil {
		return DeliveryAttempt{}, err
	}
	return attempt, nil
}

// CreateDeliveryAttempt creates a delivery attempt in a transaction.
func (store *TxStore) CreateDeliveryAttempt(ctx context.Context, params DeliveryAttemptCreate) (DeliveryAttempt, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return DeliveryAttempt{}, err
	}
	return createDeliveryAttempt(ctx, execer, params)
}

// SetDeliveryAttemptStatus records a delivery attempt lifecycle transition.
func (store *Store) SetDeliveryAttemptStatus(ctx context.Context, params DeliveryAttemptStatusUpdate) (DeliveryAttempt, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return DeliveryAttempt{}, err
	}
	return setDeliveryAttemptStatus(ctx, execer, params)
}

// SetDeliveryAttemptStatus records a delivery attempt lifecycle transition in a transaction.
func (store *TxStore) SetDeliveryAttemptStatus(ctx context.Context, params DeliveryAttemptStatusUpdate) (DeliveryAttempt, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return DeliveryAttempt{}, err
	}
	return setDeliveryAttemptStatus(ctx, execer, params)
}

// LoadDeliveryAttempt loads one delivery attempt by id.
func (store *Store) LoadDeliveryAttempt(ctx context.Context, workspaceID string, attemptID string) (DeliveryAttempt, error) {
	queryer, err := queryerFromStore(store)
	if err != nil {
		return DeliveryAttempt{}, err
	}
	return loadDeliveryAttempt(ctx, queryer, workspaceID, attemptID)
}

// LoadDeliveryAttempt loads one delivery attempt by id in a transaction.
func (store *TxStore) LoadDeliveryAttempt(ctx context.Context, workspaceID string, attemptID string) (DeliveryAttempt, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return DeliveryAttempt{}, err
	}
	return loadDeliveryAttempt(ctx, execer, workspaceID, attemptID)
}

// LoadDeliveryAttemptForUpdate locks one delivery attempt by id. Owning-domain
// adapters use it when their mutation must remain atomic with a leased service
// event delivery.
func (store *TxStore) LoadDeliveryAttemptForUpdate(ctx context.Context, workspaceID string, attemptID string) (DeliveryAttempt, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return DeliveryAttempt{}, err
	}
	return loadDeliveryAttemptForUpdate(ctx, execer, workspaceID, attemptID)
}

// AuthenticateDeliveryAttemptToken loads the current leased or delivered
// attempt addressed by a delivery-scoped runtime token.
func (store *TxStore) AuthenticateDeliveryAttemptToken(ctx context.Context, token string, deliveryID string, attemptID string, now time.Time) (DeliveryAttempt, error) {
	queryer, err := queryerFromTxStore(store)
	if err != nil {
		return DeliveryAttempt{}, err
	}
	return authenticateDeliveryAttemptToken(ctx, queryer, token, deliveryID, attemptID, now, true)
}

// ListDeliveryAttemptCandidates lists deliveries ready for transport attempt
// creation.
func (store *Store) ListDeliveryAttemptCandidates(ctx context.Context, params DeliveryAttemptCandidateQuery) ([]EventDelivery, error) {
	queryer, err := queryerFromStore(store)
	if err != nil {
		return nil, err
	}
	return listDeliveryAttemptCandidates(ctx, queryer, params)
}

// ListDeliveryAttemptCandidates lists delivery attempt candidates in a transaction.
func (store *TxStore) ListDeliveryAttemptCandidates(ctx context.Context, params DeliveryAttemptCandidateQuery) ([]EventDelivery, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return nil, err
	}
	return listDeliveryAttemptCandidates(ctx, execer, params)
}

// AppendDeliveryEffect appends one effect row and updates the parent delivery summary.
func (store *Store) AppendDeliveryEffect(ctx context.Context, params DeliveryEffectAppend) (DeliveryEffect, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return DeliveryEffect{}, err
	}
	return appendDeliveryEffect(ctx, execer, params)
}

// AppendDeliveryEffect appends one effect row and updates the parent delivery summary in a transaction.
func (store *TxStore) AppendDeliveryEffect(ctx context.Context, params DeliveryEffectAppend) (DeliveryEffect, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return DeliveryEffect{}, err
	}
	return appendDeliveryEffect(ctx, execer, params)
}

// ListDeliveryEffectsByDelivery lists the effects for one delivery in creation order.
func (store *Store) ListDeliveryEffectsByDelivery(ctx context.Context, params DeliveryEffectListQuery) ([]DeliveryEffect, error) {
	queryer, err := queryerFromStore(store)
	if err != nil {
		return nil, err
	}
	return listDeliveryEffectsByDelivery(ctx, queryer, params)
}

// ListDeliveryEffectsByDelivery lists the effects for one delivery in creation order.
func (store *TxStore) ListDeliveryEffectsByDelivery(ctx context.Context, params DeliveryEffectListQuery) ([]DeliveryEffect, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return nil, err
	}
	return listDeliveryEffectsByDelivery(ctx, execer, params)
}

// DeleteRuntimeSessionsBefore deletes old runtime sessions in bounded batches.
func (store *Store) DeleteRuntimeSessionsBefore(ctx context.Context, filter RuntimeRetentionFilter) (int64, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return 0, err
	}
	return deleteRuntimeRowsBefore(ctx, execer, runtimeRetentionDeleteSpec{
		Table:       "service_runtime_sessions",
		TimeColumn:  "updated_at",
		Status:      filter.Status,
		Cutoff:      filter.Cutoff,
		Limit:       filter.Limit,
		Description: "runtime sessions",
	})
}

// DeleteDeliveryAttemptsBefore deletes old delivery attempts in bounded batches.
func (store *Store) DeleteDeliveryAttemptsBefore(ctx context.Context, filter RuntimeRetentionFilter) (int64, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return 0, err
	}
	return deleteRuntimeRowsBefore(ctx, execer, runtimeRetentionDeleteSpec{
		Table:       "service_event_delivery_attempts",
		TimeColumn:  "updated_at",
		Status:      filter.Status,
		Cutoff:      filter.Cutoff,
		Limit:       filter.Limit,
		Description: "delivery attempts",
	})
}

// DeleteDeliveryEffectsBefore deletes old delivery effects in bounded batches.
func (store *Store) DeleteDeliveryEffectsBefore(ctx context.Context, filter RuntimeRetentionFilter) (int64, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return 0, err
	}
	return deleteRuntimeRowsBefore(ctx, execer, runtimeRetentionDeleteSpec{
		Table:       "service_event_delivery_effects",
		TimeColumn:  "updated_at",
		Status:      filter.Status,
		Cutoff:      filter.Cutoff,
		Limit:       filter.Limit,
		Description: "delivery effects",
	})
}

func createRuntimeCredentialToken(ctx context.Context, execer coredb.QueryExecer, params RuntimeCredentialTokenCreate) (RuntimeCredentialToken, error) {
	byteLength := params.TokenByteLength
	if byteLength <= 0 {
		byteLength = 32
	}
	token, err := secrets.NewToken(byteLength)
	if err != nil {
		return RuntimeCredentialToken{}, fmt.Errorf("generate service runtime token: %w", err)
	}
	create := params.RuntimeCredentialCreate
	create.TokenHash = secrets.HashToken(token)
	if strings.TrimSpace(create.TokenHint) == "" {
		create.TokenHint = runtimeTokenHint(token)
	}
	credential, err := createRuntimeCredential(ctx, execer, create)
	if err != nil {
		return RuntimeCredentialToken{}, err
	}
	return RuntimeCredentialToken{Credential: credential, Token: token}, nil
}

func createRuntimeCredential(ctx context.Context, execer coredb.QueryExecer, params RuntimeCredentialCreate) (RuntimeCredential, error) {
	workspaceID := trim(params.WorkspaceID)
	definitionID := trim(params.ServiceDefinitionID)
	installationID := trim(params.ServiceInstallationID)
	memberID := trim(params.ServiceMemberID)
	serviceKind := trim(params.ServiceKind)
	credentialKind := trim(params.CredentialKind)
	transportKind := trim(params.TransportKind)
	tokenHash := trim(params.TokenHash)
	createdByMemberID := trim(params.CreatedByMemberID)
	if workspaceID == "" || definitionID == "" || installationID == "" || memberID == "" || serviceKind == "" || credentialKind == "" || transportKind == "" || tokenHash == "" || createdByMemberID == "" {
		return RuntimeCredential{}, badRequest("runtime credential workspace, service, kind, token hash, and creator are required")
	}
	status := trim(params.Status)
	if status == "" {
		status = RuntimeCredentialStatusActive
	}
	if !validRuntimeCredentialStatus(status) {
		return RuntimeCredential{}, badRequest("runtime credential status is invalid")
	}
	credentialID := trim(params.ID)
	var err error
	if credentialID == "" {
		credentialID, err = identifiers.NewEntityID("svcrc")
		if err != nil {
			return RuntimeCredential{}, fmt.Errorf("generate service runtime credential id: %w", err)
		}
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	var disabledAt any
	if status == RuntimeCredentialStatusDisabled {
		disabledAt = now
	}
	_, err = execer.Exec(ctx, `
INSERT INTO service_runtime_credentials (
    id, workspace_id, service_definition_id, service_installation_id,
    service_member_id, service_kind, credential_kind, transport_kind,
    name, token_hash, token_hint, status, created_by_member_id,
    created_at, updated_at, disabled_at, expires_at, metadata
) VALUES (
    $1, $2, $3, $4,
    $5, $6, $7, $8,
    $9, $10, $11, $12, $13,
    $14, $14, $15, $16, $17
)
`, credentialID, workspaceID, definitionID, installationID, memberID, serviceKind, credentialKind, transportKind,
		trim(params.Name), tokenHash, trim(params.TokenHint), status, createdByMemberID,
		now, disabledAt, nullTimePtr(params.ExpiresAt), nullableBytes(params.Metadata))
	if err != nil {
		return RuntimeCredential{}, fmt.Errorf("insert service runtime credential: %w", mapRuntimeCredentialWriteError(err))
	}
	return loadRuntimeCredential(ctx, execer, workspaceID, credentialID)
}

func authenticateRuntimeCredential(ctx context.Context, execer coredb.QueryExecer, token string, now time.Time) (RuntimeCredential, error) {
	trimmedToken := trim(token)
	if trimmedToken == "" {
		return RuntimeCredential{}, badRequest("runtime token is required")
	}
	tokenHash := secrets.HashToken(trimmedToken)
	if now.IsZero() {
		now = time.Now().UTC()
	}
	credential, err := loadActiveRuntimeCredentialByTokenHash(ctx, execer, tokenHash, now)
	if err != nil {
		return RuntimeCredential{}, err
	}
	if _, err := execer.Exec(ctx, `
UPDATE service_runtime_credentials
SET last_used_at = $2,
    updated_at = $2
WHERE id = $1
`, credential.ID, now.UTC()); err != nil {
		return RuntimeCredential{}, fmt.Errorf("mark service runtime credential used: %w", err)
	}
	credential.LastUsedAt = &now
	credential.UpdatedAt = now
	return credential, nil
}

func loadRuntimeCredentialForUpdate(ctx context.Context, queryer coredb.QueryRower, workspaceID string, credentialID string) (RuntimeCredential, error) {
	row := queryer.QueryRow(ctx, runtimeCredentialSelectSQL()+`
WHERE workspace_id = $1
  AND id = $2
FOR UPDATE
`, trim(workspaceID), trim(credentialID))
	credential, err := scanRuntimeCredential(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return RuntimeCredential{}, notFound("service runtime credential not found")
		}
		return RuntimeCredential{}, fmt.Errorf("load service runtime credential for update: %w", err)
	}
	return credential, nil
}

func loadRuntimeCredential(ctx context.Context, queryer coredb.QueryRower, workspaceID string, credentialID string) (RuntimeCredential, error) {
	row := queryer.QueryRow(ctx, runtimeCredentialSelectSQL()+` WHERE workspace_id = $1 AND id = $2`, trim(workspaceID), trim(credentialID))
	credential, err := scanRuntimeCredential(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return RuntimeCredential{}, notFound("service runtime credential not found")
		}
		return RuntimeCredential{}, fmt.Errorf("load service runtime credential: %w", err)
	}
	return credential, nil
}

func loadActiveRuntimeCredentialByTokenHash(ctx context.Context, queryer coredb.QueryRower, tokenHash string, now time.Time) (RuntimeCredential, error) {
	row := queryer.QueryRow(ctx, runtimeCredentialSelectSQL()+`
WHERE token_hash = $1
  AND status = $2
  AND (expires_at IS NULL OR expires_at > $3)
`, trim(tokenHash), RuntimeCredentialStatusActive, now.UTC())
	credential, err := scanRuntimeCredential(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return RuntimeCredential{}, notFound("service runtime credential not found")
		}
		return RuntimeCredential{}, fmt.Errorf("load active service runtime credential: %w", err)
	}
	return credential, nil
}

func setRuntimeCredentialStatus(ctx context.Context, execer coredb.QueryExecer, workspaceID string, credentialID string, status string, now time.Time) (RuntimeCredential, error) {
	workspaceID = trim(workspaceID)
	credentialID = trim(credentialID)
	status = trim(status)
	if workspaceID == "" || credentialID == "" || status == "" {
		return RuntimeCredential{}, badRequest("runtime credential workspace, id, and status are required")
	}
	if !validRuntimeCredentialStatus(status) {
		return RuntimeCredential{}, badRequest("runtime credential status is invalid")
	}
	if now.IsZero() {
		now = time.Now().UTC()
	}
	var disabledAt any
	if status == RuntimeCredentialStatusDisabled {
		disabledAt = now.UTC()
	}
	result, err := execer.Exec(ctx, `
UPDATE service_runtime_credentials
SET status = $3,
    disabled_at = $4,
    updated_at = $5
WHERE workspace_id = $1
  AND id = $2
`, workspaceID, credentialID, status, disabledAt, now.UTC())
	if err != nil {
		return RuntimeCredential{}, fmt.Errorf("update service runtime credential status: %w", err)
	}
	if result.RowsAffected() == 0 {
		return RuntimeCredential{}, notFound("service runtime credential not found")
	}
	return loadRuntimeCredential(ctx, execer, workspaceID, credentialID)
}

func upsertRuntimeSession(ctx context.Context, execer coredb.QueryExecer, params RuntimeSessionUpsert) (RuntimeSession, error) {
	workspaceID := trim(params.WorkspaceID)
	definitionID := trim(params.ServiceDefinitionID)
	installationID := trim(params.ServiceInstallationID)
	memberID := trim(params.ServiceMemberID)
	credentialID := trim(params.ServiceRuntimeCredentialID)
	runnerID := trim(params.RunnerID)
	if workspaceID == "" || definitionID == "" || installationID == "" || memberID == "" || credentialID == "" || runnerID == "" {
		return RuntimeSession{}, badRequest("runtime session workspace, service, credential, and runner id are required")
	}
	status := trim(params.Status)
	if status == "" {
		status = RuntimeSessionStatusOnline
	}
	if !validRuntimeSessionStatus(status) {
		return RuntimeSession{}, badRequest("runtime session status is invalid")
	}
	sessionID := trim(params.ID)
	var err error
	if sessionID == "" {
		sessionID, err = identifiers.NewEntityID("svcrs")
		if err != nil {
			return RuntimeSession{}, fmt.Errorf("generate service runtime session id: %w", err)
		}
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	heartbeatAt := params.LastHeartbeatAt
	if heartbeatAt == nil && status == RuntimeSessionStatusOnline {
		heartbeatAt = &now
	}
	_, err = execer.Exec(ctx, `
INSERT INTO service_runtime_sessions (
    id, workspace_id, service_definition_id, service_installation_id,
    service_member_id, service_runtime_credential_id, runner_id,
    runner_version, status, capabilities_json, last_heartbeat_at,
    last_claim_at, created_at, updated_at, metadata
) VALUES (
    $1, $2, $3, $4,
    $5, $6, $7,
    $8, $9, $10, $11,
    $12, $13, $13, $14
)
ON DUPLICATE KEY UPDATE
    service_definition_id = VALUES(service_definition_id),
    service_member_id = VALUES(service_member_id),
    service_runtime_credential_id = VALUES(service_runtime_credential_id),
    runner_version = VALUES(runner_version),
    status = VALUES(status),
    capabilities_json = VALUES(capabilities_json),
    last_heartbeat_at = VALUES(last_heartbeat_at),
    last_claim_at = COALESCE(VALUES(last_claim_at), last_claim_at),
    updated_at = VALUES(updated_at),
    metadata = VALUES(metadata)
`, sessionID, workspaceID, definitionID, installationID, memberID, credentialID, runnerID,
		trim(params.RunnerVersion), status, nullableBytes(params.CapabilitiesJSON), nullTimePtr(heartbeatAt), nullTimePtr(params.LastClaimAt), now, nullableBytes(params.Metadata))
	if err != nil {
		return RuntimeSession{}, fmt.Errorf("upsert service runtime session: %w", mapRuntimeSessionWriteError(err))
	}
	return loadRuntimeSessionByInstallationRunner(ctx, execer, installationID, runnerID)
}

func loadRuntimeSession(ctx context.Context, queryer coredb.QueryRower, workspaceID string, sessionID string) (RuntimeSession, error) {
	row := queryer.QueryRow(ctx, runtimeSessionSelectSQL()+` WHERE workspace_id = $1 AND id = $2`, trim(workspaceID), trim(sessionID))
	session, err := scanRuntimeSession(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return RuntimeSession{}, notFound("service runtime session not found")
		}
		return RuntimeSession{}, fmt.Errorf("load service runtime session: %w", err)
	}
	return session, nil
}

func loadRuntimeSessionForUpdate(ctx context.Context, queryer coredb.QueryRower, workspaceID string, sessionID string) (RuntimeSession, error) {
	row := queryer.QueryRow(ctx, runtimeSessionSelectSQL()+`
WHERE workspace_id = $1
  AND id = $2
FOR UPDATE
`, trim(workspaceID), trim(sessionID))
	session, err := scanRuntimeSession(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return RuntimeSession{}, notFound("service runtime session not found")
		}
		return RuntimeSession{}, fmt.Errorf("load service runtime session for update: %w", err)
	}
	return session, nil
}

func loadRuntimeSessionByInstallationRunner(ctx context.Context, queryer coredb.QueryRower, installationID string, runnerID string) (RuntimeSession, error) {
	row := queryer.QueryRow(ctx, runtimeSessionSelectSQL()+` WHERE service_installation_id = $1 AND runner_id = $2`, trim(installationID), trim(runnerID))
	session, err := scanRuntimeSession(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return RuntimeSession{}, notFound("service runtime session not found")
		}
		return RuntimeSession{}, fmt.Errorf("load service runtime session by runner: %w", err)
	}
	return session, nil
}

func setRuntimeSessionStatus(ctx context.Context, execer coredb.QueryExecer, workspaceID string, sessionID string, status string, now time.Time) (RuntimeSession, error) {
	workspaceID = trim(workspaceID)
	sessionID = trim(sessionID)
	status = trim(status)
	if workspaceID == "" || sessionID == "" || status == "" {
		return RuntimeSession{}, badRequest("runtime session workspace, id, and status are required")
	}
	if !validRuntimeSessionStatus(status) {
		return RuntimeSession{}, badRequest("runtime session status is invalid")
	}
	if now.IsZero() {
		now = time.Now().UTC()
	}
	result, err := execer.Exec(ctx, `
UPDATE service_runtime_sessions
SET status = $3,
    updated_at = $4
WHERE workspace_id = $1
  AND id = $2
`, workspaceID, sessionID, status, now.UTC())
	if err != nil {
		return RuntimeSession{}, fmt.Errorf("update service runtime session status: %w", err)
	}
	if result.RowsAffected() == 0 {
		return RuntimeSession{}, notFound("service runtime session not found")
	}
	return loadRuntimeSession(ctx, execer, workspaceID, sessionID)
}

func setRuntimeSessionsStatusByCredential(ctx context.Context, execer coredb.QueryExecer, workspaceID string, credentialID string, status string, now time.Time) (int64, error) {
	workspaceID = trim(workspaceID)
	credentialID = trim(credentialID)
	status = trim(status)
	if workspaceID == "" || credentialID == "" || status == "" {
		return 0, badRequest("runtime session workspace, credential, and status are required")
	}
	if !validRuntimeSessionStatus(status) {
		return 0, badRequest("runtime session status is invalid")
	}
	if now.IsZero() {
		now = time.Now().UTC()
	}
	result, err := execer.Exec(ctx, `
UPDATE service_runtime_sessions
SET status = $3,
    updated_at = $4
WHERE workspace_id = $1
  AND service_runtime_credential_id = $2
  AND status <> $3
`, workspaceID, credentialID, status, now.UTC())
	if err != nil {
		return 0, fmt.Errorf("update service runtime sessions by credential: %w", err)
	}
	return result.RowsAffected(), nil
}

func createDeliveryAttempt(ctx context.Context, execer coredb.QueryExecer, params DeliveryAttemptCreate) (DeliveryAttempt, error) {
	workspaceID := trim(params.WorkspaceID)
	deliveryID := trim(params.ServiceEventDeliveryID)
	transportKind := trim(params.TransportKind)
	if workspaceID == "" || deliveryID == "" || transportKind == "" {
		return DeliveryAttempt{}, badRequest("delivery attempt workspace, delivery, and transport are required")
	}
	status := trim(params.Status)
	if status == "" {
		status = DeliveryAttemptStatusLeased
	}
	if !validDeliveryAttemptStatus(status) {
		return DeliveryAttempt{}, badRequest("delivery attempt status is invalid")
	}
	attemptID := trim(params.ID)
	var err error
	if attemptID == "" {
		attemptID, err = identifiers.NewEntityID("svcea")
		if err != nil {
			return DeliveryAttempt{}, fmt.Errorf("generate service event delivery attempt id: %w", err)
		}
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	if err := lockDeliveryForAttempt(ctx, execer, workspaceID, deliveryID); err != nil {
		return DeliveryAttempt{}, err
	}
	attemptNumber := params.AttemptNumber
	if attemptNumber <= 0 {
		var nextNumber int
		if err := execer.QueryRow(ctx, `
SELECT COALESCE(MAX(attempt_number), 0) + 1
FROM service_event_delivery_attempts
WHERE service_event_delivery_id = $1
`, deliveryID).Scan(&nextNumber); err != nil {
			return DeliveryAttempt{}, fmt.Errorf("allocate service event delivery attempt number: %w", err)
		}
		attemptNumber = nextNumber
	}
	claimedAt := params.ClaimedAt
	if claimedAt == nil && status == DeliveryAttemptStatusLeased {
		claimedAt = &now
	}
	result, err := execer.Exec(ctx, `
INSERT INTO service_event_delivery_attempts (
    id, workspace_id, service_event_delivery_id, service_installation_id,
    service_member_id, transport_kind, service_runtime_credential_id,
    service_runtime_session_id, attempt_number, status, lease_owner,
    lease_expires_at, delivery_token_hash, delivery_token_expires_at,
    callback_url_snapshot, response_status_code, response_excerpt,
    error_code, error_detail, claimed_at, delivered_at, created_at, updated_at
)
SELECT
    $1, delivery.workspace_id, delivery.id, delivery.service_installation_id,
    delivery.service_member_id, $2, $3,
    $4, $5, $6, $7,
    $8, $9, $10,
    $11, $12, $13,
    '', '', $14, $15, $16, $16
FROM service_event_deliveries delivery
WHERE delivery.workspace_id = $17
  AND delivery.id = $18
`, attemptID, transportKind, nullableString(trim(params.ServiceRuntimeCredentialID)), nullableString(trim(params.ServiceRuntimeSessionID)), attemptNumber, status, trim(params.LeaseOwner), nullTimePtr(params.LeaseExpiresAt), nullableString(trim(params.DeliveryTokenHash)), nullTimePtr(params.DeliveryTokenExpiresAt), trim(params.CallbackURLSnapshot), nullableIntPtr(params.ResponseStatusCode), trim(params.ResponseExcerpt), nullTimePtr(claimedAt), nullTimePtr(params.DeliveredAt), now, workspaceID, deliveryID)
	if err != nil {
		return DeliveryAttempt{}, fmt.Errorf("insert service event delivery attempt: %w", mapDeliveryAttemptWriteError(err))
	}
	if result.RowsAffected() == 0 {
		return DeliveryAttempt{}, notFound("service event delivery not found")
	}
	return loadDeliveryAttempt(ctx, execer, workspaceID, attemptID)
}

func lockDeliveryForAttempt(ctx context.Context, queryer coredb.QueryRower, workspaceID string, deliveryID string) error {
	var lockedID string
	if err := queryer.QueryRow(ctx, `
SELECT id
FROM service_event_deliveries
WHERE workspace_id = $1
  AND id = $2
FOR UPDATE
`, workspaceID, deliveryID).Scan(&lockedID); err != nil {
		if coredb.IsNotFound(err) {
			return notFound("service event delivery not found")
		}
		return fmt.Errorf("lock service event delivery for attempt: %w", err)
	}
	return nil
}

func setDeliveryAttemptStatus(ctx context.Context, execer coredb.QueryExecer, params DeliveryAttemptStatusUpdate) (DeliveryAttempt, error) {
	workspaceID := trim(params.WorkspaceID)
	attemptID := trim(params.ID)
	status := trim(params.Status)
	if workspaceID == "" || attemptID == "" || status == "" {
		return DeliveryAttempt{}, badRequest("delivery attempt workspace, id, and status are required")
	}
	if !validDeliveryAttemptStatus(status) {
		return DeliveryAttempt{}, badRequest("delivery attempt status is invalid")
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	result, err := execer.Exec(ctx, `
UPDATE service_event_delivery_attempts
SET status = $3,
    response_status_code = COALESCE($4, response_status_code),
    response_excerpt = CASE WHEN $5 <> '' THEN $5 ELSE response_excerpt END,
    error_code = CASE WHEN $6 <> '' THEN $6 ELSE error_code END,
    error_detail = CASE WHEN $7 <> '' THEN $7 ELSE error_detail END,
    result_json = COALESCE($8, result_json),
    delivered_at = CASE WHEN $3 = 'delivered' THEN $9 ELSE delivered_at END,
    acked_at = CASE WHEN $3 = 'acked' THEN $9 ELSE acked_at END,
    failed_at = CASE WHEN $3 = 'failed' THEN $9 ELSE failed_at END,
    expired_at = CASE WHEN $3 = 'expired' THEN $9 ELSE expired_at END,
    updated_at = $9
WHERE workspace_id = $1
  AND id = $2
`, workspaceID, attemptID, status, nullableIntPtr(params.ResponseStatusCode), trim(params.ResponseExcerpt), trim(params.ErrorCode), trim(params.ErrorDetail), nullableBytes(params.ResultJSON), now)
	if err != nil {
		return DeliveryAttempt{}, fmt.Errorf("update service event delivery attempt status: %w", err)
	}
	if result.RowsAffected() == 0 {
		return DeliveryAttempt{}, notFound("service event delivery attempt not found")
	}
	return loadDeliveryAttempt(ctx, execer, workspaceID, attemptID)
}

func listDeliveryAttemptCandidates(ctx context.Context, queryer coredb.Queryer, params DeliveryAttemptCandidateQuery) ([]EventDelivery, error) {
	workspaceID := trim(params.WorkspaceID)
	if workspaceID == "" {
		return nil, badRequest("delivery attempt candidate workspace is required")
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	limit := normalizeRuntimeListLimit(params.Limit)
	rows, err := queryer.Query(ctx, eventDeliverySelectFromSQL("service_event_deliveries delivery", "delivery")+`
WHERE delivery.workspace_id = $1
  AND ($2 = '' OR delivery.service_installation_id = $2)
  AND ($3 = '' OR delivery.subscription_kind = $3)
  AND delivery.status IN ('running', 'pending', 'retry_waiting')
  AND (delivery.next_retry_at IS NULL OR delivery.next_retry_at <= $4)
  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')
      AND (attempt.lease_expires_at IS NULL OR attempt.lease_expires_at > $4)
  )
ORDER BY COALESCE(delivery.next_retry_at, delivery.created_at), delivery.id
LIMIT $5
`, workspaceID, trim(params.ServiceInstallationID), trim(params.SubscriptionKind), now, limit)
	if err != nil {
		return nil, fmt.Errorf("list service event delivery attempt candidates: %w", err)
	}
	defer rows.Close()

	candidates := make([]EventDelivery, 0)
	for rows.Next() {
		delivery, err := scanEventDelivery(rows)
		if err != nil {
			return nil, fmt.Errorf("scan service event delivery attempt candidate: %w", err)
		}
		candidates = append(candidates, delivery)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate service event delivery attempt candidates: %w", err)
	}
	return candidates, nil
}

func appendDeliveryEffect(ctx context.Context, execer coredb.QueryExecer, params DeliveryEffectAppend) (DeliveryEffect, error) {
	workspaceID := trim(params.WorkspaceID)
	attemptID := trim(params.DeliveryAttemptID)
	effectKind := trim(params.EffectKind)
	targetKind := trim(params.TargetKind)
	targetRef := trim(params.TargetRef)
	if workspaceID == "" || attemptID == "" || effectKind == "" || targetKind == "" || targetRef == "" {
		return DeliveryEffect{}, badRequest("delivery effect workspace, attempt, effect, and target are required")
	}
	if _, err := runeffect.NormalizeEffectKind(effectKind); err != nil {
		return DeliveryEffect{}, badRequest("delivery effect kind is invalid")
	}
	status := trim(params.Status)
	if status == "" {
		status = DeliveryEffectStatusCreated
	}
	if !validDeliveryEffectStatus(status) {
		return DeliveryEffect{}, badRequest("delivery effect status is invalid")
	}
	effectID := trim(params.ID)
	var err error
	if effectID == "" {
		effectID, err = identifiers.NewEntityID("svceff")
		if err != nil {
			return DeliveryEffect{}, fmt.Errorf("generate service event delivery effect id: %w", err)
		}
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	messageID := nullableString(trim(params.MessageID))
	if err := (runeffect.Effect{Kind: runeffect.EffectKind(effectKind), Ref: trim(params.EffectRef), MessageID: trim(params.MessageID)}).Validate(); err != nil {
		return DeliveryEffect{}, badRequest("delivery effect contract is invalid")
	}
	result, err := execer.Exec(ctx, `
INSERT INTO service_event_delivery_effects (
    id, workspace_id, service_event_delivery_id, service_event_delivery_attempt_id,
    service_installation_id, service_member_id, effect_kind, effect_ref,
    target_kind, target_ref, message_id, status, metadata, created_at, updated_at
)
SELECT
    $1, attempt.workspace_id, attempt.service_event_delivery_id, attempt.id,
    attempt.service_installation_id, attempt.service_member_id, $2, $3,
    $4, $5, $6, $7, $8, $9, $9
FROM service_event_delivery_attempts attempt
WHERE attempt.workspace_id = $10
  AND attempt.id = $11
`, effectID, effectKind, trim(params.EffectRef), targetKind, targetRef, messageID, status, nullableBytes(params.Metadata), now, workspaceID, attemptID)
	if err != nil {
		return DeliveryEffect{}, fmt.Errorf("insert service event delivery effect: %w", err)
	}
	if result.RowsAffected() == 0 {
		return DeliveryEffect{}, notFound("service event delivery attempt not found")
	}
	effect, err := loadDeliveryEffect(ctx, execer, workspaceID, effectID)
	if err != nil {
		return DeliveryEffect{}, err
	}
	if effect.Status == DeliveryEffectStatusCreated {
		if _, err := execer.Exec(ctx, `
UPDATE service_event_deliveries
SET effect_kind = $2,
    effect_ref = $3,
    message_id = $4,
    updated_at = $5
WHERE workspace_id = $1
  AND id = $6
`, workspaceID, effect.EffectKind, effect.EffectRef, nullableString(effect.MessageID), now, effect.ServiceEventDeliveryID); err != nil {
			return DeliveryEffect{}, fmt.Errorf("update service event delivery effect summary: %w", err)
		}
	}
	return effect, nil
}

func listDeliveryEffectsByDelivery(ctx context.Context, queryer coredb.Queryer, params DeliveryEffectListQuery) ([]DeliveryEffect, error) {
	workspaceID := trim(params.WorkspaceID)
	deliveryID := trim(params.ServiceEventDeliveryID)
	if workspaceID == "" || deliveryID == "" {
		return nil, badRequest("delivery effect workspace and delivery are required")
	}
	limit := normalizeRuntimeListLimit(params.Limit)
	rows, err := queryer.Query(ctx, deliveryEffectSelectSQL()+`
WHERE workspace_id = $1
  AND service_event_delivery_id = $2
ORDER BY created_at ASC, id ASC
LIMIT $3
`, workspaceID, deliveryID, limit)
	if err != nil {
		return nil, fmt.Errorf("list service event delivery effects: %w", err)
	}
	defer rows.Close()

	effects := make([]DeliveryEffect, 0)
	for rows.Next() {
		effect, err := scanDeliveryEffect(rows)
		if err != nil {
			return nil, fmt.Errorf("scan service event delivery effect: %w", err)
		}
		effects = append(effects, effect)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate service event delivery effects: %w", err)
	}
	return effects, nil
}

func loadDeliveryAttempt(ctx context.Context, queryer coredb.QueryRower, workspaceID string, attemptID string) (DeliveryAttempt, error) {
	row := queryer.QueryRow(ctx, deliveryAttemptSelectSQL()+` WHERE workspace_id = $1 AND id = $2`, trim(workspaceID), trim(attemptID))
	attempt, err := scanDeliveryAttempt(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return DeliveryAttempt{}, notFound("service event delivery attempt not found")
		}
		return DeliveryAttempt{}, fmt.Errorf("load service event delivery attempt: %w", err)
	}
	return attempt, nil
}

func loadDeliveryAttemptForUpdate(ctx context.Context, queryer coredb.QueryRower, workspaceID string, attemptID string) (DeliveryAttempt, error) {
	row := queryer.QueryRow(ctx, deliveryAttemptSelectSQL()+`
WHERE workspace_id = $1
  AND id = $2
FOR UPDATE
`, trim(workspaceID), trim(attemptID))
	attempt, err := scanDeliveryAttempt(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return DeliveryAttempt{}, notFound("service event delivery attempt not found")
		}
		return DeliveryAttempt{}, fmt.Errorf("load service event delivery attempt for update: %w", err)
	}
	return attempt, nil
}

func authenticateDeliveryAttemptToken(ctx context.Context, queryer coredb.QueryRower, token string, deliveryID string, attemptID string, now time.Time, forUpdate bool) (DeliveryAttempt, error) {
	trimmedToken := trim(token)
	trimmedDeliveryID := trim(deliveryID)
	trimmedAttemptID := trim(attemptID)
	if trimmedToken == "" || trimmedDeliveryID == "" || trimmedAttemptID == "" {
		return DeliveryAttempt{}, badRequest("delivery token, delivery id, and attempt id are required")
	}
	if now.IsZero() {
		now = time.Now().UTC()
	}
	query := deliveryAttemptSelectSQL() + `
WHERE delivery_token_hash = $1
  AND service_event_delivery_id = $2
  AND id = $3
  AND status IN ($4, $5, $6, $7)
  AND (delivery_token_expires_at IS NULL OR delivery_token_expires_at > $8)
`
	if forUpdate {
		query += "FOR UPDATE\n"
	}
	row := queryer.QueryRow(ctx, query, secrets.HashToken(trimmedToken), trimmedDeliveryID, trimmedAttemptID, DeliveryAttemptStatusLeased, DeliveryAttemptStatusDelivered, DeliveryAttemptStatusAcked, DeliveryAttemptStatusFailed, now.UTC())
	attempt, err := scanDeliveryAttempt(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return DeliveryAttempt{}, notFound("service event delivery attempt token not found")
		}
		return DeliveryAttempt{}, fmt.Errorf("authenticate service event delivery attempt token: %w", err)
	}
	return attempt, nil
}

func loadDeliveryEffect(ctx context.Context, queryer coredb.QueryRower, workspaceID string, effectID string) (DeliveryEffect, error) {
	row := queryer.QueryRow(ctx, deliveryEffectSelectSQL()+` WHERE workspace_id = $1 AND id = $2`, trim(workspaceID), trim(effectID))
	effect, err := scanDeliveryEffect(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return DeliveryEffect{}, notFound("service event delivery effect not found")
		}
		return DeliveryEffect{}, fmt.Errorf("load service event delivery effect: %w", err)
	}
	return effect, nil
}

func runtimeCredentialSelectSQL() string {
	return `
SELECT id, workspace_id, service_definition_id, service_installation_id,
       service_member_id, service_kind, credential_kind, transport_kind,
       name, token_hash, token_hint, status, created_by_member_id,
       created_at, updated_at, last_used_at, disabled_at, expires_at, metadata
FROM service_runtime_credentials`
}

func runtimeSessionSelectSQL() string {
	return `
SELECT id, workspace_id, service_definition_id, service_installation_id,
       service_member_id, service_runtime_credential_id, runner_id,
       runner_version, status, capabilities_json, last_heartbeat_at,
       last_claim_at, created_at, updated_at, metadata
FROM service_runtime_sessions`
}

func deliveryAttemptSelectSQL() string {
	return `
SELECT id, workspace_id, service_event_delivery_id, service_installation_id,
       service_member_id, transport_kind, COALESCE(service_runtime_credential_id, ''),
       COALESCE(service_runtime_session_id, ''), attempt_number, status,
       lease_owner, lease_expires_at, COALESCE(delivery_token_hash, ''),
       delivery_token_expires_at, callback_url_snapshot, response_status_code,
       response_excerpt, claimed_at, delivered_at, acked_at, failed_at,
       expired_at, error_code, error_detail, result_json, created_at, updated_at
FROM service_event_delivery_attempts`
}

func deliveryEffectSelectSQL() string {
	return `
SELECT id, workspace_id, service_event_delivery_id, service_event_delivery_attempt_id,
       service_installation_id, service_member_id, effect_kind, effect_ref,
       target_kind, target_ref, COALESCE(message_id, ''), status, metadata,
       created_at, updated_at
FROM service_event_delivery_effects`
}

func scanRuntimeCredential(row coredb.Row) (RuntimeCredential, error) {
	var credential RuntimeCredential
	var lastUsedAt, disabledAt, expiresAt sql.NullTime
	var metadata []byte
	if err := row.Scan(
		&credential.ID,
		&credential.WorkspaceID,
		&credential.ServiceDefinitionID,
		&credential.ServiceInstallationID,
		&credential.ServiceMemberID,
		&credential.ServiceKind,
		&credential.CredentialKind,
		&credential.TransportKind,
		&credential.Name,
		&credential.TokenHash,
		&credential.TokenHint,
		&credential.Status,
		&credential.CreatedByMemberID,
		&credential.CreatedAt,
		&credential.UpdatedAt,
		&lastUsedAt,
		&disabledAt,
		&expiresAt,
		&metadata,
	); err != nil {
		return RuntimeCredential{}, err
	}
	credential.ID = trim(credential.ID)
	credential.WorkspaceID = trim(credential.WorkspaceID)
	credential.ServiceDefinitionID = trim(credential.ServiceDefinitionID)
	credential.ServiceInstallationID = trim(credential.ServiceInstallationID)
	credential.ServiceMemberID = trim(credential.ServiceMemberID)
	credential.ServiceKind = trim(credential.ServiceKind)
	credential.CredentialKind = trim(credential.CredentialKind)
	credential.TransportKind = trim(credential.TransportKind)
	credential.Status = trim(credential.Status)
	credential.LastUsedAt = ptrFromNullTime(lastUsedAt)
	credential.DisabledAt = ptrFromNullTime(disabledAt)
	credential.ExpiresAt = ptrFromNullTime(expiresAt)
	credential.Metadata = metadata
	return credential, nil
}

func scanRuntimeSession(row coredb.Row) (RuntimeSession, error) {
	var session RuntimeSession
	var lastHeartbeatAt, lastClaimAt sql.NullTime
	var capabilitiesJSON, metadata []byte
	if err := row.Scan(
		&session.ID,
		&session.WorkspaceID,
		&session.ServiceDefinitionID,
		&session.ServiceInstallationID,
		&session.ServiceMemberID,
		&session.ServiceRuntimeCredentialID,
		&session.RunnerID,
		&session.RunnerVersion,
		&session.Status,
		&capabilitiesJSON,
		&lastHeartbeatAt,
		&lastClaimAt,
		&session.CreatedAt,
		&session.UpdatedAt,
		&metadata,
	); err != nil {
		return RuntimeSession{}, err
	}
	session.ID = trim(session.ID)
	session.WorkspaceID = trim(session.WorkspaceID)
	session.ServiceDefinitionID = trim(session.ServiceDefinitionID)
	session.ServiceInstallationID = trim(session.ServiceInstallationID)
	session.ServiceMemberID = trim(session.ServiceMemberID)
	session.ServiceRuntimeCredentialID = trim(session.ServiceRuntimeCredentialID)
	session.RunnerID = trim(session.RunnerID)
	session.Status = trim(session.Status)
	session.CapabilitiesJSON = capabilitiesJSON
	session.LastHeartbeatAt = ptrFromNullTime(lastHeartbeatAt)
	session.LastClaimAt = ptrFromNullTime(lastClaimAt)
	session.Metadata = metadata
	return session, nil
}

func scanDeliveryAttempt(row coredb.Row) (DeliveryAttempt, error) {
	var attempt DeliveryAttempt
	var attemptNumber int64
	var leaseExpiresAt, tokenExpiresAt, claimedAt, deliveredAt, ackedAt, failedAt, expiredAt sql.NullTime
	var responseStatusCode sql.NullInt64
	var resultJSON []byte
	if err := row.Scan(
		&attempt.ID,
		&attempt.WorkspaceID,
		&attempt.ServiceEventDeliveryID,
		&attempt.ServiceInstallationID,
		&attempt.ServiceMemberID,
		&attempt.TransportKind,
		&attempt.ServiceRuntimeCredentialID,
		&attempt.ServiceRuntimeSessionID,
		&attemptNumber,
		&attempt.Status,
		&attempt.LeaseOwner,
		&leaseExpiresAt,
		&attempt.DeliveryTokenHash,
		&tokenExpiresAt,
		&attempt.CallbackURLSnapshot,
		&responseStatusCode,
		&attempt.ResponseExcerpt,
		&claimedAt,
		&deliveredAt,
		&ackedAt,
		&failedAt,
		&expiredAt,
		&attempt.ErrorCode,
		&attempt.ErrorDetail,
		&resultJSON,
		&attempt.CreatedAt,
		&attempt.UpdatedAt,
	); err != nil {
		return DeliveryAttempt{}, err
	}
	attempt.ID = trim(attempt.ID)
	attempt.WorkspaceID = trim(attempt.WorkspaceID)
	attempt.ServiceEventDeliveryID = trim(attempt.ServiceEventDeliveryID)
	attempt.ServiceInstallationID = trim(attempt.ServiceInstallationID)
	attempt.ServiceMemberID = trim(attempt.ServiceMemberID)
	attempt.TransportKind = trim(attempt.TransportKind)
	attempt.ServiceRuntimeCredentialID = trim(attempt.ServiceRuntimeCredentialID)
	attempt.ServiceRuntimeSessionID = trim(attempt.ServiceRuntimeSessionID)
	attempt.AttemptNumber = int(attemptNumber)
	attempt.Status = trim(attempt.Status)
	attempt.LeaseOwner = trim(attempt.LeaseOwner)
	attempt.LeaseExpiresAt = ptrFromNullTime(leaseExpiresAt)
	attempt.DeliveryTokenExpiresAt = ptrFromNullTime(tokenExpiresAt)
	attempt.ClaimedAt = ptrFromNullTime(claimedAt)
	attempt.DeliveredAt = ptrFromNullTime(deliveredAt)
	attempt.AckedAt = ptrFromNullTime(ackedAt)
	attempt.FailedAt = ptrFromNullTime(failedAt)
	attempt.ExpiredAt = ptrFromNullTime(expiredAt)
	if responseStatusCode.Valid {
		value := int(responseStatusCode.Int64)
		attempt.ResponseStatusCode = &value
	}
	attempt.ResultJSON = resultJSON
	return attempt, nil
}

func scanDeliveryEffect(row coredb.Row) (DeliveryEffect, error) {
	var effect DeliveryEffect
	var metadata []byte
	if err := row.Scan(
		&effect.ID,
		&effect.WorkspaceID,
		&effect.ServiceEventDeliveryID,
		&effect.DeliveryAttemptID,
		&effect.ServiceInstallationID,
		&effect.ServiceMemberID,
		&effect.EffectKind,
		&effect.EffectRef,
		&effect.TargetKind,
		&effect.TargetRef,
		&effect.MessageID,
		&effect.Status,
		&metadata,
		&effect.CreatedAt,
		&effect.UpdatedAt,
	); err != nil {
		return DeliveryEffect{}, err
	}
	effect.ID = trim(effect.ID)
	effect.WorkspaceID = trim(effect.WorkspaceID)
	effect.ServiceEventDeliveryID = trim(effect.ServiceEventDeliveryID)
	effect.DeliveryAttemptID = trim(effect.DeliveryAttemptID)
	effect.ServiceInstallationID = trim(effect.ServiceInstallationID)
	effect.ServiceMemberID = trim(effect.ServiceMemberID)
	effect.EffectKind = trim(effect.EffectKind)
	effect.EffectRef = trim(effect.EffectRef)
	effect.TargetKind = trim(effect.TargetKind)
	effect.TargetRef = trim(effect.TargetRef)
	effect.MessageID = trim(effect.MessageID)
	effect.Status = trim(effect.Status)
	effect.Metadata = metadata
	return effect, nil
}

func runtimeTokenHint(token string) string {
	trimmed := trim(token)
	if len(trimmed) <= 8 {
		return trimmed
	}
	return trimmed[:8]
}

func validRuntimeCredentialStatus(status string) bool {
	switch trim(status) {
	case RuntimeCredentialStatusActive, RuntimeCredentialStatusDisabled, RuntimeCredentialStatusRotated, RuntimeCredentialStatusExpired:
		return true
	default:
		return false
	}
}

func validRuntimeSessionStatus(status string) bool {
	switch trim(status) {
	case RuntimeSessionStatusOnline, RuntimeSessionStatusStale, RuntimeSessionStatusOffline, RuntimeSessionStatusDisabled:
		return true
	default:
		return false
	}
}

func validDeliveryAttemptStatus(status string) bool {
	switch trim(status) {
	case DeliveryAttemptStatusLeased, DeliveryAttemptStatusDelivered, DeliveryAttemptStatusAcked, DeliveryAttemptStatusFailed, DeliveryAttemptStatusExpired, DeliveryAttemptStatusCanceled:
		return true
	default:
		return false
	}
}

func validDeliveryEffectStatus(status string) bool {
	switch trim(status) {
	case DeliveryEffectStatusCreated, DeliveryEffectStatusFailed, DeliveryEffectStatusReverted:
		return true
	default:
		return false
	}
}

func nullableIntPtr(value *int) any {
	if value == nil {
		return nil
	}
	return *value
}

func mapRuntimeCredentialWriteError(err error) error {
	if constraint, ok := coredb.UniqueViolationConstraint(err); ok {
		switch constraint {
		case "service_runtime_credentials_token_hash_key":
			return conflict("service runtime credential token already exists")
		}
	}
	return err
}

func mapRuntimeSessionWriteError(err error) error {
	if constraint, ok := coredb.UniqueViolationConstraint(err); ok {
		switch constraint {
		case "service_runtime_sessions_installation_runner_key":
			return conflict("service runtime session already exists")
		}
	}
	return err
}

func mapDeliveryAttemptWriteError(err error) error {
	if constraint, ok := coredb.UniqueViolationConstraint(err); ok {
		switch constraint {
		case "service_event_delivery_attempts_delivery_number_key":
			return conflict("service event delivery attempt number already exists")
		case "service_event_delivery_attempts_token_hash_key":
			return conflict("service event delivery attempt token already exists")
		}
	}
	return err
}

type runtimeRetentionDeleteSpec struct {
	Table       string
	TimeColumn  string
	Status      string
	Cutoff      time.Time
	Limit       int
	Description string
}

func deleteRuntimeRowsBefore(ctx context.Context, execer coredb.QueryExecer, spec runtimeRetentionDeleteSpec) (int64, error) {
	status := trim(spec.Status)
	if status == "" || spec.Cutoff.IsZero() {
		return 0, badRequest("runtime retention status and cutoff are required")
	}
	limit := normalizeRuntimeRetentionLimit(spec.Limit)
	rows, err := execer.Query(ctx, fmt.Sprintf(`
SELECT id
FROM %s
WHERE status = $1
  AND %s < $2
ORDER BY %s ASC, id ASC
LIMIT $3
`, spec.Table, spec.TimeColumn, spec.TimeColumn), status, spec.Cutoff.UTC(), limit)
	if err != nil {
		return 0, fmt.Errorf("select old %s: %w", spec.Description, err)
	}
	defer rows.Close()

	ids := make([]string, 0, limit)
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return 0, fmt.Errorf("scan old %s: %w", spec.Description, err)
		}
		if trimmed := trim(id); trimmed != "" {
			ids = append(ids, trimmed)
		}
	}
	if err := rows.Err(); err != nil {
		return 0, fmt.Errorf("iterate old %s: %w", spec.Description, err)
	}

	var deleted int64
	if len(ids) == 0 {
		return deleted, nil
	}
	placeholders := make([]string, len(ids))
	args := make([]any, len(ids))
	for index, id := range ids {
		placeholders[index] = fmt.Sprintf("$%d", index+1)
		args[index] = id
	}
	result, err := execer.Exec(ctx, fmt.Sprintf(`DELETE FROM %s WHERE id IN (%s)`, spec.Table, strings.Join(placeholders, ", ")), args...)
	if err != nil {
		return deleted, fmt.Errorf("delete old %s: %w", spec.Description, err)
	}
	deleted += result.RowsAffected()
	return deleted, nil
}

func normalizeRuntimeRetentionLimit(limit int) int {
	if limit <= 0 {
		return runtimeRetentionDeleteBatchDefault
	}
	if limit > runtimeRetentionDeleteBatchMax {
		return runtimeRetentionDeleteBatchMax
	}
	return limit
}

func normalizeRuntimeListLimit(limit int) int {
	if limit <= 0 {
		return runtimeListLimitDefault
	}
	if limit > runtimeListLimitMax {
		return runtimeListLimitMax
	}
	return limit
}