package postgres

import (
	"context"
	"errors"
	"fmt"
	"strconv"

	"git.cin.su/m/Assistant/contracts/authority"
	"git.cin.su/m/Assistant/contracts/continuity"
	"git.cin.su/m/Assistant/contracts/identity"
	continuitydomain "git.cin.su/m/Assistant/modules/continuity/domain"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"
)

var (
	ErrProspectiveNotFound         = errors.New("prospective item not found")
	ErrProspectiveOwnerConflict    = errors.New("prospective item owner conflict")
	ErrProspectiveStateConflict    = errors.New("prospective item state conflict")
	ErrProspectiveRevisionConflict = errors.New("prospective item revision conflict")
	ErrActionReceiptNotFound       = errors.New("continuity action receipt not found")
	ErrActionIntentConflict        = errors.New("action intent was already used for a different command")
	ErrAttentionNotFound           = errors.New("attention item not found")
	ErrAttentionStateConflict      = errors.New("attention item state conflict")
	ErrAttentionRevisionConflict   = errors.New("attention item revision conflict")
)

type ContinuityConfig struct{ Host, Port, Database, User, Password string }
type ContinuityRepository struct{ pool *pgxpool.Pool }

func openPool(ctx context.Context, config ContinuityConfig) (*pgxpool.Pool, error) {
	poolConfig, err := pgxpool.ParseConfig("")
	if err != nil {
		return nil, err
	}
	port, err := strconv.ParseUint(config.Port, 10, 16)
	if err != nil {
		return nil, fmt.Errorf("invalid postgres port: %w", err)
	}
	poolConfig.ConnConfig.Host = config.Host
	poolConfig.ConnConfig.Port = uint16(port)
	poolConfig.ConnConfig.Database = config.Database
	poolConfig.ConnConfig.User = config.User
	poolConfig.ConnConfig.Password = config.Password
	return pgxpool.NewWithConfig(ctx, poolConfig)
}

func OpenContinuity(ctx context.Context, config ContinuityConfig) (*ContinuityRepository, error) {
	pool, err := openPool(ctx, config)
	if err != nil {
		return nil, err
	}
	return &ContinuityRepository{pool: pool}, nil
}
func (r *ContinuityRepository) Close() { r.pool.Close() }
func (r *ContinuityRepository) Save(ctx context.Context, item continuity.ProspectiveItem) error {
	_, err := r.pool.Exec(ctx, `INSERT INTO continuity.prospective_items (prospective_item_id, commitment_id, owner_principal_id, scope_id, goal_id, source_intent_id, source_utterance_id, summary, trigger_at_unix_millis, state, revision) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) ON CONFLICT (prospective_item_id) DO UPDATE SET state=EXCLUDED.state, revision=EXCLUDED.revision`, item.ID, item.CommitmentID, item.OwnerPrincipalID, item.ScopeID, item.GoalID, item.SourceIntentID, item.SourceUtteranceID, item.Summary, item.TriggerAtUnixMillis, item.State, item.Revision)
	return err
}
func (r *ContinuityRepository) Load(ctx context.Context, id continuity.ProspectiveItemID) (continuity.ProspectiveItem, error) {
	var item continuity.ProspectiveItem
	err := r.pool.QueryRow(ctx, `SELECT prospective_item_id, commitment_id, owner_principal_id, scope_id, goal_id, source_intent_id, source_utterance_id, summary, trigger_at_unix_millis, state, revision FROM continuity.prospective_items WHERE prospective_item_id=$1`, id).Scan(&item.ID, &item.CommitmentID, &item.OwnerPrincipalID, &item.ScopeID, &item.GoalID, &item.SourceIntentID, &item.SourceUtteranceID, &item.Summary, &item.TriggerAtUnixMillis, &item.State, &item.Revision)
	if err != nil {
		return continuity.ProspectiveItem{}, fmt.Errorf("load prospective item: %w", err)
	}
	return item, nil
}

func (r *ContinuityRepository) ByOwner(ctx context.Context, owner identity.PrincipalID, limit int) ([]continuity.ProspectiveItem, error) {
	rows, err := r.pool.Query(ctx, `SELECT prospective_item_id, commitment_id, owner_principal_id, scope_id, goal_id, source_intent_id, source_utterance_id, summary, trigger_at_unix_millis, state, revision FROM continuity.prospective_items WHERE owner_principal_id=$1 ORDER BY trigger_at_unix_millis, prospective_item_id LIMIT $2`, owner, limit)
	if err != nil {
		return nil, fmt.Errorf("list prospective items by owner: %w", err)
	}
	defer rows.Close()
	items := make([]continuity.ProspectiveItem, 0)
	for rows.Next() {
		var item continuity.ProspectiveItem
		if err := rows.Scan(&item.ID, &item.CommitmentID, &item.OwnerPrincipalID, &item.ScopeID, &item.GoalID, &item.SourceIntentID, &item.SourceUtteranceID, &item.Summary, &item.TriggerAtUnixMillis, &item.State, &item.Revision); err != nil {
			return nil, fmt.Errorf("scan prospective item: %w", err)
		}
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list prospective items by owner: %w", err)
	}
	return items, nil
}

func (r *ContinuityRepository) DueArmed(ctx context.Context, nowUnixMillis int64, limit int) ([]continuity.ProspectiveItem, error) {
	if limit < 1 || limit > 100 {
		return nil, fmt.Errorf("invalid due reminder limit")
	}
	rows, err := r.pool.Query(ctx, `SELECT prospective_item_id, commitment_id, owner_principal_id, scope_id, goal_id, source_intent_id, source_utterance_id, summary, trigger_at_unix_millis, state, revision
		FROM continuity.prospective_items
		WHERE state=$1 AND trigger_at_unix_millis <= $2
		ORDER BY trigger_at_unix_millis, prospective_item_id LIMIT $3`, continuity.ProspectiveArmed, nowUnixMillis, limit)
	if err != nil {
		return nil, fmt.Errorf("list due prospective items: %w", err)
	}
	defer rows.Close()
	items := make([]continuity.ProspectiveItem, 0)
	for rows.Next() {
		var item continuity.ProspectiveItem
		if err := rows.Scan(&item.ID, &item.CommitmentID, &item.OwnerPrincipalID, &item.ScopeID, &item.GoalID, &item.SourceIntentID, &item.SourceUtteranceID, &item.Summary, &item.TriggerAtUnixMillis, &item.State, &item.Revision); err != nil {
			return nil, fmt.Errorf("scan due prospective item: %w", err)
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func (r *ContinuityRepository) ClaimDueReminder(ctx context.Context, id continuity.ProspectiveItemID, revision continuity.ProspectiveRevision, attentionID continuity.AttentionItemID, nowUnixMillis int64) (continuity.ProspectiveItem, continuity.AttentionItem, error) {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return continuity.ProspectiveItem{}, continuity.AttentionItem{}, err
	}
	defer func() { _ = tx.Rollback(ctx) }()

	var triggered continuity.ProspectiveItem
	err = tx.QueryRow(ctx, `UPDATE continuity.prospective_items
		SET state=$1, revision=revision+1
		WHERE prospective_item_id=$2 AND state=$3 AND revision=$4 AND trigger_at_unix_millis <= $5
		RETURNING prospective_item_id,commitment_id,owner_principal_id,scope_id,goal_id,source_intent_id,source_utterance_id,summary,trigger_at_unix_millis,state,revision`,
		continuity.ProspectiveTriggered, id, continuity.ProspectiveArmed, revision, nowUnixMillis).Scan(
		&triggered.ID, &triggered.CommitmentID, &triggered.OwnerPrincipalID, &triggered.ScopeID, &triggered.GoalID,
		&triggered.SourceIntentID, &triggered.SourceUtteranceID, &triggered.Summary, &triggered.TriggerAtUnixMillis, &triggered.State, &triggered.Revision)
	if errors.Is(err, pgx.ErrNoRows) {
		return continuity.ProspectiveItem{}, continuity.AttentionItem{}, ErrProspectiveRevisionConflict
	}
	if err != nil {
		return continuity.ProspectiveItem{}, continuity.AttentionItem{}, fmt.Errorf("trigger due reminder: %w", err)
	}

	var revalidating continuity.ProspectiveItem
	err = tx.QueryRow(ctx, `UPDATE continuity.prospective_items
		SET state=$1, revision=revision+1
		WHERE prospective_item_id=$2 AND state=$3 AND revision=$4
		RETURNING prospective_item_id,commitment_id,owner_principal_id,scope_id,goal_id,source_intent_id,source_utterance_id,summary,trigger_at_unix_millis,state,revision`,
		continuity.ProspectiveRevalidating, triggered.ID, continuity.ProspectiveTriggered, triggered.Revision).Scan(
		&revalidating.ID, &revalidating.CommitmentID, &revalidating.OwnerPrincipalID, &revalidating.ScopeID, &revalidating.GoalID,
		&revalidating.SourceIntentID, &revalidating.SourceUtteranceID, &revalidating.Summary, &revalidating.TriggerAtUnixMillis, &revalidating.State, &revalidating.Revision)
	if err != nil {
		return continuity.ProspectiveItem{}, continuity.AttentionItem{}, fmt.Errorf("begin due reminder revalidation: %w", err)
	}

	var delivered continuity.ProspectiveItem
	err = tx.QueryRow(ctx, `UPDATE continuity.prospective_items
		SET state=$1, revision=revision+1
		WHERE prospective_item_id=$2 AND state=$3 AND revision=$4
		RETURNING prospective_item_id,commitment_id,owner_principal_id,scope_id,goal_id,source_intent_id,source_utterance_id,summary,trigger_at_unix_millis,state,revision`,
		continuity.ProspectiveDelivered, revalidating.ID, continuity.ProspectiveRevalidating, revalidating.Revision).Scan(
		&delivered.ID, &delivered.CommitmentID, &delivered.OwnerPrincipalID, &delivered.ScopeID, &delivered.GoalID,
		&delivered.SourceIntentID, &delivered.SourceUtteranceID, &delivered.Summary, &delivered.TriggerAtUnixMillis, &delivered.State, &delivered.Revision)
	if err != nil {
		return continuity.ProspectiveItem{}, continuity.AttentionItem{}, fmt.Errorf("finish due reminder revalidation: %w", err)
	}

	attention, err := continuitydomain.BuildDueReminderAttention(delivered, attentionID, nowUnixMillis)
	if err != nil {
		return continuity.ProspectiveItem{}, continuity.AttentionItem{}, err
	}
	err = tx.QueryRow(ctx, `INSERT INTO continuity.attention_items (
		attention_item_id,owner_principal_id,scope_id,source_kind,source_prospective_item_id,source_external_ref,level,state,summary,
		due_at_unix_millis,created_at_unix_millis,delivered_at_unix_millis,action_intent_id,authorized_action_id,
		action_attempt_id,delivery_outcome,response_id,response_utterance_id,response_operation,response_outcome,responded_at_unix_millis,revision)
		VALUES ($1,$2,$3,$4,$5,'',$6,$7,$8,$9,$10,0,'','','','','','','','',0,1)
		ON CONFLICT (source_kind, source_prospective_item_id) DO UPDATE SET
			state=EXCLUDED.state,
			due_at_unix_millis=EXCLUDED.due_at_unix_millis,
			created_at_unix_millis=EXCLUDED.created_at_unix_millis,
			delivered_at_unix_millis=0,
			action_intent_id='',
			authorized_action_id='',
			action_attempt_id='',
			delivery_outcome='',
			response_id='',
			response_utterance_id='',
			response_operation='',
			response_outcome='',
			responded_at_unix_millis=0,
			revision=continuity.attention_items.revision+1
		RETURNING attention_item_id,owner_principal_id,scope_id,source_kind,source_prospective_item_id,source_external_ref,level,state,summary,due_at_unix_millis,
			created_at_unix_millis,delivered_at_unix_millis,action_intent_id,authorized_action_id,action_attempt_id,delivery_outcome,
			response_id,response_utterance_id,response_operation,response_outcome,responded_at_unix_millis,revision`,
		attention.ID, attention.OwnerPrincipalID, attention.ScopeID, attention.SourceKind, attention.SourceProspectiveID,
		attention.Level, attention.State, attention.Summary, attention.DueAtUnixMillis, attention.CreatedAtUnixMillis).Scan(
		&attention.ID, &attention.OwnerPrincipalID, &attention.ScopeID, &attention.SourceKind, &attention.SourceProspectiveID, &attention.SourceExternalRef,
		&attention.Level, &attention.State, &attention.Summary, &attention.DueAtUnixMillis, &attention.CreatedAtUnixMillis,
		&attention.DeliveredAtUnixMillis, &attention.ActionIntentID, &attention.AuthorizedActionID, &attention.ActionAttemptID,
		&attention.DeliveryOutcome, &attention.ResponseID, &attention.ResponseUtteranceID, &attention.ResponseOperation,
		&attention.ResponseOutcome, &attention.RespondedAtUnixMillis, &attention.Revision)
	if err != nil {
		return continuity.ProspectiveItem{}, continuity.AttentionItem{}, fmt.Errorf("create due reminder attention: %w", err)
	}
	if err = tx.Commit(ctx); err != nil {
		return continuity.ProspectiveItem{}, continuity.AttentionItem{}, err
	}
	return delivered, attention, nil
}

func (r *ContinuityRepository) PendingAttention(ctx context.Context, limit int) ([]continuity.AttentionItem, error) {
	if limit < 1 || limit > 100 {
		return nil, fmt.Errorf("invalid attention limit")
	}
	rows, err := r.pool.Query(ctx, `SELECT attention_item_id,owner_principal_id,scope_id,source_kind,source_prospective_item_id,source_external_ref,level,state,summary,
		due_at_unix_millis,created_at_unix_millis,delivered_at_unix_millis,action_intent_id,authorized_action_id,action_attempt_id,delivery_outcome,
		response_id,response_utterance_id,response_operation,response_outcome,responded_at_unix_millis,revision
		FROM continuity.attention_items WHERE state=$1 ORDER BY due_at_unix_millis, attention_item_id LIMIT $2`, continuity.AttentionPending, limit)
	if err != nil {
		return nil, fmt.Errorf("list pending attention: %w", err)
	}
	defer rows.Close()
	return scanAttentionRows(rows)
}

func (r *ContinuityRepository) AttentionByOwner(ctx context.Context, owner identity.PrincipalID, limit int) ([]continuity.AttentionItem, error) {
	if owner == "" || limit < 1 || limit > 100 {
		return nil, fmt.Errorf("invalid attention owner query")
	}
	rows, err := r.pool.Query(ctx, `SELECT attention_item_id,owner_principal_id,scope_id,source_kind,source_prospective_item_id,source_external_ref,level,state,summary,
		due_at_unix_millis,created_at_unix_millis,delivered_at_unix_millis,action_intent_id,authorized_action_id,action_attempt_id,delivery_outcome,
		response_id,response_utterance_id,response_operation,response_outcome,responded_at_unix_millis,revision
		FROM continuity.attention_items WHERE owner_principal_id=$1 ORDER BY due_at_unix_millis DESC, attention_item_id DESC LIMIT $2`, owner, limit)
	if err != nil {
		return nil, fmt.Errorf("list attention by owner: %w", err)
	}
	defer rows.Close()
	return scanAttentionRows(rows)
}

func (r *ContinuityRepository) ResponseCandidates(ctx context.Context, owner identity.PrincipalID, limit int) ([]continuitydomain.ResponseCandidate, error) {
	if owner == "" || limit < 1 || limit > 50 {
		return nil, fmt.Errorf("invalid response candidate query")
	}
	candidates := make([]continuitydomain.ResponseCandidate, 0, limit)
	attentionRows, err := r.pool.Query(ctx, `SELECT
		attention_item_id,owner_principal_id,scope_id,source_kind,source_prospective_item_id,source_external_ref,level,state,summary,
		due_at_unix_millis,created_at_unix_millis,delivered_at_unix_millis,action_intent_id,authorized_action_id,action_attempt_id,delivery_outcome,
		response_id,response_utterance_id,response_operation,response_outcome,responded_at_unix_millis,revision
		FROM continuity.attention_items
		WHERE owner_principal_id=$1
		  AND source_kind=$2
		  AND state IN ($3,$4,$5)
		ORDER BY created_at_unix_millis DESC, attention_item_id DESC
		LIMIT $6`,
		owner, continuity.AttentionSourceAgentDelegation, continuity.AttentionPending, continuity.AttentionDelivered, continuity.AttentionFailed, limit)
	if err != nil {
		return nil, fmt.Errorf("list delegation response candidates: %w", err)
	}
	defer attentionRows.Close()
	for attentionRows.Next() {
		item, err := scanAttentionItem(attentionRows)
		if err != nil {
			return nil, fmt.Errorf("scan delegation response candidate: %w", err)
		}
		candidates = append(candidates, continuitydomain.ResponseCandidate{Attention: item})
	}
	if err := attentionRows.Err(); err != nil {
		return nil, fmt.Errorf("list delegation response candidates: %w", err)
	}
	if len(candidates) >= limit {
		return candidates[:limit], nil
	}
	rows, err := r.pool.Query(ctx, `SELECT
		COALESCE(a.attention_item_id,''),COALESCE(a.owner_principal_id,''),COALESCE(a.scope_id,''),COALESCE(a.source_kind,''),COALESCE(a.source_prospective_item_id,''),COALESCE(a.source_external_ref,''),COALESCE(a.level,''),COALESCE(a.state,''),COALESCE(a.summary,''),
		COALESCE(a.due_at_unix_millis,0),COALESCE(a.created_at_unix_millis,0),COALESCE(a.delivered_at_unix_millis,0),COALESCE(a.action_intent_id,''),COALESCE(a.authorized_action_id,''),COALESCE(a.action_attempt_id,''),COALESCE(a.delivery_outcome,''),
		COALESCE(a.response_id,''),COALESCE(a.response_utterance_id,''),COALESCE(a.response_operation,''),COALESCE(a.response_outcome,''),COALESCE(a.responded_at_unix_millis,0),COALESCE(a.revision,0),
		p.prospective_item_id,p.commitment_id,p.owner_principal_id,p.scope_id,p.goal_id,p.source_intent_id,p.source_utterance_id,p.summary,p.trigger_at_unix_millis,p.state,p.revision
		FROM continuity.prospective_items p
		LEFT JOIN continuity.attention_items a ON a.source_kind=$2 AND a.source_prospective_item_id=p.prospective_item_id AND a.owner_principal_id=p.owner_principal_id
		WHERE p.owner_principal_id=$1
		  AND (a.attention_item_id IS NULL OR a.state IN ($3,$4,$5))
		  AND p.state IN ($6,$7)
		ORDER BY p.recorded_at DESC, COALESCE(a.delivered_at_unix_millis,0) DESC, p.prospective_item_id DESC
		LIMIT $8`,
		owner, continuity.AttentionSourceProspectiveReminder, continuity.AttentionDelivered, continuity.AttentionPending,
		continuity.AttentionFailed, continuity.ProspectiveDelivered, continuity.ProspectiveArmed, limit-len(candidates))
	if err != nil {
		return nil, fmt.Errorf("list human response candidates: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var candidate continuitydomain.ResponseCandidate
		var attentionID, attentionOwner, attentionScope, sourceKind, sourceProspective, sourceExternalRef, level, attentionState, attentionSummary string
		var actionIntent, authorizedAction, actionAttempt, deliveryOutcome, responseID, responseUtterance, responseOperation, responseOutcome string
		var dueAt, createdAt, deliveredAt, respondedAt int64
		var attentionRevision uint64
		if err := rows.Scan(
			&attentionID, &attentionOwner, &attentionScope, &sourceKind, &sourceProspective, &sourceExternalRef, &level, &attentionState, &attentionSummary,
			&dueAt, &createdAt, &deliveredAt, &actionIntent, &authorizedAction, &actionAttempt, &deliveryOutcome,
			&responseID, &responseUtterance, &responseOperation, &responseOutcome, &respondedAt, &attentionRevision,
			&candidate.Prospective.ID, &candidate.Prospective.CommitmentID, &candidate.Prospective.OwnerPrincipalID,
			&candidate.Prospective.ScopeID, &candidate.Prospective.GoalID, &candidate.Prospective.SourceIntentID,
			&candidate.Prospective.SourceUtteranceID, &candidate.Prospective.Summary, &candidate.Prospective.TriggerAtUnixMillis,
			&candidate.Prospective.State, &candidate.Prospective.Revision); err != nil {
			return nil, fmt.Errorf("scan human response candidate: %w", err)
		}
		if attentionID != "" {
			candidate.Attention = continuity.AttentionItem{
				ID:                    continuity.AttentionItemID(attentionID),
				OwnerPrincipalID:      identity.PrincipalID(attentionOwner),
				ScopeID:               authority.ScopeID(attentionScope),
				SourceKind:            continuity.AttentionSourceKind(sourceKind),
				SourceProspectiveID:   continuity.ProspectiveItemID(sourceProspective),
				SourceExternalRef:     sourceExternalRef,
				Level:                 continuity.AttentionLevel(level),
				State:                 continuity.AttentionState(attentionState),
				Summary:               attentionSummary,
				DueAtUnixMillis:       dueAt,
				CreatedAtUnixMillis:   createdAt,
				DeliveredAtUnixMillis: deliveredAt,
				ActionIntentID:        continuity.CommittedActionIntentID(actionIntent),
				AuthorizedActionID:    continuity.CommittedAuthorizedActionID(authorizedAction),
				ActionAttemptID:       continuity.CommittedActionAttemptID(actionAttempt),
				DeliveryOutcome:       deliveryOutcome,
				ResponseID:            continuity.HumanResponseID(responseID),
				ResponseUtteranceID:   responseUtterance,
				ResponseOperation:     continuity.HumanResponseOperation(responseOperation),
				ResponseOutcome:       continuity.ResponseOutcome(responseOutcome),
				RespondedAtUnixMillis: respondedAt,
				Revision:              continuity.AttentionRevision(attentionRevision),
			}
		}
		candidates = append(candidates, candidate)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list human response candidates: %w", err)
	}
	return candidates, nil
}

func scanAttentionItem(scanner interface {
	Scan(...any) error
}) (continuity.AttentionItem, error) {
	var item continuity.AttentionItem
	var attentionID, attentionOwner, attentionScope, sourceKind, sourceProspective, sourceExternalRef, level, attentionState, attentionSummary string
	var actionIntent, authorizedAction, actionAttempt, deliveryOutcome, responseID, responseUtterance, responseOperation, responseOutcome string
	var dueAt, createdAt, deliveredAt, respondedAt int64
	var attentionRevision uint64
	if err := scanner.Scan(
		&attentionID, &attentionOwner, &attentionScope, &sourceKind, &sourceProspective, &sourceExternalRef, &level, &attentionState, &attentionSummary,
		&dueAt, &createdAt, &deliveredAt, &actionIntent, &authorizedAction, &actionAttempt, &deliveryOutcome,
		&responseID, &responseUtterance, &responseOperation, &responseOutcome, &respondedAt, &attentionRevision); err != nil {
		return continuity.AttentionItem{}, err
	}
	item = continuity.AttentionItem{
		ID:                    continuity.AttentionItemID(attentionID),
		OwnerPrincipalID:      identity.PrincipalID(attentionOwner),
		ScopeID:               authority.ScopeID(attentionScope),
		SourceKind:            continuity.AttentionSourceKind(sourceKind),
		SourceProspectiveID:   continuity.ProspectiveItemID(sourceProspective),
		SourceExternalRef:     sourceExternalRef,
		Level:                 continuity.AttentionLevel(level),
		State:                 continuity.AttentionState(attentionState),
		Summary:               attentionSummary,
		DueAtUnixMillis:       dueAt,
		CreatedAtUnixMillis:   createdAt,
		DeliveredAtUnixMillis: deliveredAt,
		ActionIntentID:        continuity.CommittedActionIntentID(actionIntent),
		AuthorizedActionID:    continuity.CommittedAuthorizedActionID(authorizedAction),
		ActionAttemptID:       continuity.CommittedActionAttemptID(actionAttempt),
		DeliveryOutcome:       deliveryOutcome,
		ResponseID:            continuity.HumanResponseID(responseID),
		ResponseUtteranceID:   responseUtterance,
		ResponseOperation:     continuity.HumanResponseOperation(responseOperation),
		ResponseOutcome:       continuity.ResponseOutcome(responseOutcome),
		RespondedAtUnixMillis: respondedAt,
		Revision:              continuity.AttentionRevision(attentionRevision),
	}
	return item, nil
}

func (r *ContinuityRepository) ApplyHumanResponse(ctx context.Context, response continuity.HumanResponse) error {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	if _, err = tx.Exec(ctx, `INSERT INTO continuity.human_responses (
		human_response_id,owner_principal_id,scope_id,source_utterance_id,source_intent_id,operation,target_kind,
		target_attention_id,target_prospective_item_id,target_external_ref,resolution_state,outcome,clarification,
		snooze_until_unix_millis,user_reported_not_verified,recorded_at_unix_millis,revision)
		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17)`,
		response.ID, response.OwnerPrincipalID, response.ScopeID, response.SourceUtteranceID, response.SourceIntentID,
		response.Operation, response.TargetKind, response.TargetAttentionID, response.TargetProspectiveID, response.TargetExternalRef,
		response.ResolutionState, response.Outcome, response.Clarification, response.SnoozeUntilUnixMillis,
		response.UserReportedNotVerified, response.RecordedAtUnixMillis, response.Revision); err != nil {
		return fmt.Errorf("save human response: %w", err)
	}
	if response.ResolutionState != continuity.ResponseResolved {
		return tx.Commit(ctx)
	}
	switch response.Operation {
	case continuity.ResponseComplete:
		tag, err := tx.Exec(ctx, `UPDATE continuity.prospective_items SET state=$1, revision=revision+1
			WHERE prospective_item_id=$2 AND owner_principal_id=$3 AND state IN ($4,$5)`,
			continuity.ProspectiveSatisfied, response.TargetProspectiveID, response.OwnerPrincipalID,
			continuity.ProspectiveDelivered, continuity.ProspectiveArmed)
		if err != nil {
			return fmt.Errorf("complete response prospective: %w", err)
		}
		if tag.RowsAffected() != 1 {
			return ErrProspectiveStateConflict
		}
		if response.TargetAttentionID != "" {
			if err = markAttentionResponded(ctx, tx, response, continuity.AttentionResolved); err != nil {
				return err
			}
		}
	case continuity.ResponseSnooze:
		tag, err := tx.Exec(ctx, `UPDATE continuity.prospective_items SET state=$1, trigger_at_unix_millis=$2, revision=revision+1
			WHERE prospective_item_id=$3 AND owner_principal_id=$4 AND state IN ($5,$6)`,
			continuity.ProspectiveArmed, response.SnoozeUntilUnixMillis, response.TargetProspectiveID,
			response.OwnerPrincipalID, continuity.ProspectiveDelivered, continuity.ProspectiveArmed)
		if err != nil {
			return fmt.Errorf("snooze response prospective: %w", err)
		}
		if tag.RowsAffected() != 1 {
			return ErrProspectiveStateConflict
		}
		if response.TargetAttentionID != "" {
			if err = markAttentionResponded(ctx, tx, response, continuity.AttentionResolved); err != nil {
				return err
			}
		}
	case continuity.ResponseCancel:
		tag, err := tx.Exec(ctx, `UPDATE continuity.prospective_items SET state=$1, revision=revision+1
			WHERE prospective_item_id=$2 AND owner_principal_id=$3 AND state IN ($4,$5)`,
			continuity.ProspectiveSuppressed, response.TargetProspectiveID, response.OwnerPrincipalID,
			continuity.ProspectiveDelivered, continuity.ProspectiveArmed)
		if err != nil {
			return fmt.Errorf("cancel response prospective: %w", err)
		}
		if tag.RowsAffected() != 1 {
			return ErrProspectiveStateConflict
		}
		if response.TargetAttentionID != "" {
			if err = markAttentionResponded(ctx, tx, response, continuity.AttentionSuppressed); err != nil {
				return err
			}
		}
	case continuity.ResponseAcknowledge, continuity.ResponseDismiss:
		if response.TargetAttentionID == "" {
			return ErrAttentionNotFound
		}
		if err = markAttentionResponded(ctx, tx, response, continuity.AttentionResolved); err != nil {
			return err
		}
	case continuity.ResponseContinue, continuity.ResponseReject:
		if response.TargetAttentionID == "" {
			return ErrAttentionNotFound
		}
		if err = markAttentionResponded(ctx, tx, response, continuity.AttentionResolved); err != nil {
			return err
		}
	}
	return tx.Commit(ctx)
}

func (r *ContinuityRepository) CreateAttentionItem(ctx context.Context, item continuity.AttentionItem) (continuity.AttentionItem, error) {
	err := r.pool.QueryRow(ctx, `INSERT INTO continuity.attention_items(
		attention_item_id,owner_principal_id,scope_id,source_kind,source_prospective_item_id,source_external_ref,level,state,summary,
		due_at_unix_millis,created_at_unix_millis,delivered_at_unix_millis,action_intent_id,authorized_action_id,action_attempt_id,delivery_outcome,
		response_id,response_utterance_id,response_operation,response_outcome,responded_at_unix_millis,revision)
		VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22)
		ON CONFLICT (source_kind, source_external_ref) WHERE source_external_ref <> '' DO UPDATE SET
			level=EXCLUDED.level,
			state=EXCLUDED.state,
			summary=EXCLUDED.summary,
			due_at_unix_millis=EXCLUDED.due_at_unix_millis,
			created_at_unix_millis=EXCLUDED.created_at_unix_millis,
			delivered_at_unix_millis=0,
			action_intent_id='',
			authorized_action_id='',
			action_attempt_id='',
			delivery_outcome='',
			response_id='',
			response_utterance_id='',
			response_operation='',
			response_outcome='',
			responded_at_unix_millis=0,
			revision=continuity.attention_items.revision+1
		RETURNING attention_item_id,owner_principal_id,scope_id,source_kind,source_prospective_item_id,source_external_ref,level,state,summary,due_at_unix_millis,
			created_at_unix_millis,delivered_at_unix_millis,action_intent_id,authorized_action_id,action_attempt_id,delivery_outcome,
			response_id,response_utterance_id,response_operation,response_outcome,responded_at_unix_millis,revision`,
		item.ID, item.OwnerPrincipalID, item.ScopeID, item.SourceKind, item.SourceProspectiveID, item.SourceExternalRef, item.Level, item.State,
		item.Summary, item.DueAtUnixMillis, item.CreatedAtUnixMillis, item.DeliveredAtUnixMillis, item.ActionIntentID, item.AuthorizedActionID,
		item.ActionAttemptID, item.DeliveryOutcome, item.ResponseID, item.ResponseUtteranceID, item.ResponseOperation, item.ResponseOutcome,
		item.RespondedAtUnixMillis, item.Revision).Scan(
		&item.ID, &item.OwnerPrincipalID, &item.ScopeID, &item.SourceKind, &item.SourceProspectiveID, &item.SourceExternalRef, &item.Level, &item.State,
		&item.Summary, &item.DueAtUnixMillis, &item.CreatedAtUnixMillis, &item.DeliveredAtUnixMillis, &item.ActionIntentID, &item.AuthorizedActionID,
		&item.ActionAttemptID, &item.DeliveryOutcome, &item.ResponseID, &item.ResponseUtteranceID, &item.ResponseOperation, &item.ResponseOutcome,
		&item.RespondedAtUnixMillis, &item.Revision)
	if err != nil {
		return continuity.AttentionItem{}, fmt.Errorf("create attention item: %w", err)
	}
	return item, nil
}

func (r *ContinuityRepository) MarkAttentionDelivering(ctx context.Context, id continuity.AttentionItemID, revision continuity.AttentionRevision) (continuity.AttentionItem, error) {
	var item continuity.AttentionItem
	err := r.pool.QueryRow(ctx, `UPDATE continuity.attention_items SET state=$1, revision=revision+1
		WHERE attention_item_id=$2 AND state=$3 AND revision=$4
		RETURNING attention_item_id,owner_principal_id,scope_id,source_kind,source_prospective_item_id,source_external_ref,level,state,summary,due_at_unix_millis,
			created_at_unix_millis,delivered_at_unix_millis,action_intent_id,authorized_action_id,action_attempt_id,delivery_outcome,
			response_id,response_utterance_id,response_operation,response_outcome,responded_at_unix_millis,revision`,
		continuity.AttentionDelivering, id, continuity.AttentionPending, revision).Scan(
		&item.ID, &item.OwnerPrincipalID, &item.ScopeID, &item.SourceKind, &item.SourceProspectiveID, &item.SourceExternalRef, &item.Level, &item.State,
		&item.Summary, &item.DueAtUnixMillis, &item.CreatedAtUnixMillis, &item.DeliveredAtUnixMillis, &item.ActionIntentID,
		&item.AuthorizedActionID, &item.ActionAttemptID, &item.DeliveryOutcome, &item.ResponseID, &item.ResponseUtteranceID,
		&item.ResponseOperation, &item.ResponseOutcome, &item.RespondedAtUnixMillis, &item.Revision)
	if errors.Is(err, pgx.ErrNoRows) {
		return continuity.AttentionItem{}, ErrAttentionRevisionConflict
	}
	return item, err
}

func (r *ContinuityRepository) MarkAttentionDelivered(ctx context.Context, id continuity.AttentionItemID, revision continuity.AttentionRevision, intentID continuity.CommittedActionIntentID, authorizedID continuity.CommittedAuthorizedActionID, attemptID continuity.CommittedActionAttemptID, outcome string, deliveredAtUnixMillis int64) (continuity.AttentionItem, error) {
	var item continuity.AttentionItem
	err := r.pool.QueryRow(ctx, `UPDATE continuity.attention_items
		SET state=$1, revision=revision+1, delivered_at_unix_millis=$2, action_intent_id=$3, authorized_action_id=$4,
			action_attempt_id=$5, delivery_outcome=$6
		WHERE attention_item_id=$7 AND state=$8 AND revision=$9
		RETURNING attention_item_id,owner_principal_id,scope_id,source_kind,source_prospective_item_id,source_external_ref,level,state,summary,due_at_unix_millis,
			created_at_unix_millis,delivered_at_unix_millis,action_intent_id,authorized_action_id,action_attempt_id,delivery_outcome,
			response_id,response_utterance_id,response_operation,response_outcome,responded_at_unix_millis,revision`,
		continuity.AttentionDelivered, deliveredAtUnixMillis, intentID, authorizedID, attemptID, outcome,
		id, continuity.AttentionDelivering, revision).Scan(
		&item.ID, &item.OwnerPrincipalID, &item.ScopeID, &item.SourceKind, &item.SourceProspectiveID, &item.SourceExternalRef, &item.Level, &item.State,
		&item.Summary, &item.DueAtUnixMillis, &item.CreatedAtUnixMillis, &item.DeliveredAtUnixMillis, &item.ActionIntentID,
		&item.AuthorizedActionID, &item.ActionAttemptID, &item.DeliveryOutcome, &item.ResponseID, &item.ResponseUtteranceID,
		&item.ResponseOperation, &item.ResponseOutcome, &item.RespondedAtUnixMillis, &item.Revision)
	if errors.Is(err, pgx.ErrNoRows) {
		return continuity.AttentionItem{}, ErrAttentionRevisionConflict
	}
	return item, err
}

func (r *ContinuityRepository) MarkAttentionFailed(ctx context.Context, id continuity.AttentionItemID, revision continuity.AttentionRevision, intentID continuity.CommittedActionIntentID, authorizedID continuity.CommittedAuthorizedActionID, attemptID continuity.CommittedActionAttemptID, outcome string) (continuity.AttentionItem, error) {
	var item continuity.AttentionItem
	err := r.pool.QueryRow(ctx, `UPDATE continuity.attention_items
		SET state=$1, revision=revision+1, action_intent_id=$2, authorized_action_id=$3,
			action_attempt_id=$4, delivery_outcome=$5
		WHERE attention_item_id=$6 AND state=$7 AND revision=$8
		RETURNING attention_item_id,owner_principal_id,scope_id,source_kind,source_prospective_item_id,source_external_ref,level,state,summary,due_at_unix_millis,
			created_at_unix_millis,delivered_at_unix_millis,action_intent_id,authorized_action_id,action_attempt_id,delivery_outcome,
			response_id,response_utterance_id,response_operation,response_outcome,responded_at_unix_millis,revision`,
		continuity.AttentionFailed, intentID, authorizedID, attemptID, outcome,
		id, continuity.AttentionDelivering, revision).Scan(
		&item.ID, &item.OwnerPrincipalID, &item.ScopeID, &item.SourceKind, &item.SourceProspectiveID, &item.SourceExternalRef, &item.Level, &item.State,
		&item.Summary, &item.DueAtUnixMillis, &item.CreatedAtUnixMillis, &item.DeliveredAtUnixMillis, &item.ActionIntentID,
		&item.AuthorizedActionID, &item.ActionAttemptID, &item.DeliveryOutcome, &item.ResponseID, &item.ResponseUtteranceID,
		&item.ResponseOperation, &item.ResponseOutcome, &item.RespondedAtUnixMillis, &item.Revision)
	if errors.Is(err, pgx.ErrNoRows) {
		return continuity.AttentionItem{}, ErrAttentionRevisionConflict
	}
	return item, err
}

// SatisfyDelivered owns both the Prospective CAS and its idempotency receipt in
// one Continuity-local transaction. It never reaches Commit storage.
func (r *ContinuityRepository) SatisfyDelivered(ctx context.Context, command continuity.SatisfyDeliveredCommand) (continuity.ActionReceipt, error) {
	if err := continuitydomain.ValidateSatisfyDelivered(command); err != nil {
		return continuity.ActionReceipt{}, err
	}
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return continuity.ActionReceipt{}, err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	// Serialize deliveries of one ActionIntent inside the canonical owner. The
	// receipt remains the durable correctness source; the lock is transaction-local.
	if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, command.ActionIntentID); err != nil {
		return continuity.ActionReceipt{}, err
	}
	if existing, loadErr := loadActionReceipt(ctx, tx, command.ActionIntentID); loadErr == nil {
		if receiptMatchesCommand(existing, command) {
			return existing, tx.Commit(ctx)
		}
		return continuity.ActionReceipt{}, ErrActionIntentConflict
	} else if !errors.Is(loadErr, ErrActionReceiptNotFound) {
		return continuity.ActionReceipt{}, loadErr
	}

	var item continuity.ProspectiveItem
	err = tx.QueryRow(ctx, `UPDATE continuity.prospective_items
		SET state=$1, revision=revision+1
		WHERE prospective_item_id=$2 AND owner_principal_id=$3 AND state=$4 AND revision=$5
		RETURNING prospective_item_id,commitment_id,owner_principal_id,scope_id,goal_id,source_intent_id,source_utterance_id,summary,trigger_at_unix_millis,state,revision`,
		continuity.ProspectiveSatisfied, command.Subject.ProspectiveItemID, command.OwnerPrincipalID,
		continuity.ProspectiveDelivered, command.Subject.ExpectedRevision).Scan(
		&item.ID, &item.CommitmentID, &item.OwnerPrincipalID, &item.ScopeID, &item.GoalID, &item.SourceIntentID, &item.SourceUtteranceID, &item.Summary,
		&item.TriggerAtUnixMillis, &item.State, &item.Revision)
	if errors.Is(err, pgx.ErrNoRows) {
		return continuity.ActionReceipt{}, diagnoseProspectiveConflict(ctx, tx, command)
	}
	if err != nil {
		return continuity.ActionReceipt{}, fmt.Errorf("satisfy delivered prospective item: %w", err)
	}
	receipt := continuity.ActionReceipt{
		ActionIntentID: command.ActionIntentID, AuthorizedActionID: command.AuthorizedActionID,
		OwnerPrincipalID: command.OwnerPrincipalID, Subject: command.Subject,
		ResultState: item.State, ResultRevision: item.Revision,
	}
	_, err = tx.Exec(ctx, `INSERT INTO continuity.action_receipts (
		action_intent_id,authorized_action_id,owner_principal_id,prospective_item_id,expected_revision,
		from_state,to_state,evidence_ref,result_state,result_revision)
		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)`, receipt.ActionIntentID, receipt.AuthorizedActionID,
		receipt.OwnerPrincipalID, receipt.Subject.ProspectiveItemID, receipt.Subject.ExpectedRevision,
		receipt.Subject.FromState, receipt.Subject.ToState, receipt.Subject.EvidenceRef,
		receipt.ResultState, receipt.ResultRevision)
	if err != nil {
		return continuity.ActionReceipt{}, fmt.Errorf("save continuity action receipt: %w", err)
	}
	if err = tx.Commit(ctx); err != nil {
		return continuity.ActionReceipt{}, err
	}
	return receipt, nil
}

func (r *ContinuityRepository) LoadActionReceipt(ctx context.Context, intentID continuity.CommittedActionIntentID) (continuity.ActionReceipt, error) {
	return loadActionReceipt(ctx, r.pool, intentID)
}

type continuityRowQuerier interface {
	QueryRow(context.Context, string, ...any) pgx.Row
}

type continuityExecutor interface {
	Exec(context.Context, string, ...any) (pgconn.CommandTag, error)
}

type continuityRows interface {
	Next() bool
	Scan(...any) error
	Err() error
}

func markAttentionResponded(ctx context.Context, executor continuityExecutor, response continuity.HumanResponse, state continuity.AttentionState) error {
	tag, err := executor.Exec(ctx, `UPDATE continuity.attention_items
		SET state=$1, response_id=$2, response_utterance_id=$3, response_operation=$4, response_outcome=$5,
			responded_at_unix_millis=$6, revision=revision+1
		WHERE attention_item_id=$7 AND owner_principal_id=$8 AND state IN ($9,$10,$11)`,
		state, response.ID, response.SourceUtteranceID, response.Operation, response.Outcome, response.RecordedAtUnixMillis,
		response.TargetAttentionID, response.OwnerPrincipalID, continuity.AttentionDelivered, continuity.AttentionPending, continuity.AttentionFailed)
	if err != nil {
		return fmt.Errorf("mark attention responded: %w", err)
	}
	if tag.RowsAffected() != 1 {
		return ErrAttentionStateConflict
	}
	return nil
}

func scanAttentionRows(rows continuityRows) ([]continuity.AttentionItem, error) {
	items := make([]continuity.AttentionItem, 0)
	for rows.Next() {
		var item continuity.AttentionItem
		if err := rows.Scan(&item.ID, &item.OwnerPrincipalID, &item.ScopeID, &item.SourceKind, &item.SourceProspectiveID, &item.SourceExternalRef,
			&item.Level, &item.State, &item.Summary, &item.DueAtUnixMillis, &item.CreatedAtUnixMillis,
			&item.DeliveredAtUnixMillis, &item.ActionIntentID, &item.AuthorizedActionID, &item.ActionAttemptID,
			&item.DeliveryOutcome, &item.ResponseID, &item.ResponseUtteranceID, &item.ResponseOperation,
			&item.ResponseOutcome, &item.RespondedAtUnixMillis, &item.Revision); err != nil {
			return nil, fmt.Errorf("scan attention item: %w", err)
		}
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("scan attention items: %w", err)
	}
	return items, nil
}

func loadActionReceipt(ctx context.Context, querier continuityRowQuerier, intentID continuity.CommittedActionIntentID) (continuity.ActionReceipt, error) {
	var receipt continuity.ActionReceipt
	err := querier.QueryRow(ctx, `SELECT action_intent_id,authorized_action_id,owner_principal_id,prospective_item_id,
		expected_revision,from_state,to_state,evidence_ref,result_state,result_revision
		FROM continuity.action_receipts WHERE action_intent_id=$1`, intentID).Scan(
		&receipt.ActionIntentID, &receipt.AuthorizedActionID, &receipt.OwnerPrincipalID,
		&receipt.Subject.ProspectiveItemID, &receipt.Subject.ExpectedRevision, &receipt.Subject.FromState,
		&receipt.Subject.ToState, &receipt.Subject.EvidenceRef, &receipt.ResultState, &receipt.ResultRevision)
	if errors.Is(err, pgx.ErrNoRows) {
		return continuity.ActionReceipt{}, ErrActionReceiptNotFound
	}
	if err != nil {
		return continuity.ActionReceipt{}, fmt.Errorf("load continuity action receipt: %w", err)
	}
	return receipt, nil
}

func receiptMatchesCommand(receipt continuity.ActionReceipt, command continuity.SatisfyDeliveredCommand) bool {
	return receipt.ActionIntentID == command.ActionIntentID && receipt.AuthorizedActionID == command.AuthorizedActionID &&
		receipt.OwnerPrincipalID == command.OwnerPrincipalID && receipt.Subject == command.Subject
}

func diagnoseProspectiveConflict(ctx context.Context, querier continuityRowQuerier, command continuity.SatisfyDeliveredCommand) error {
	var owner identity.PrincipalID
	var state continuity.ProspectiveState
	var revision continuity.ProspectiveRevision
	err := querier.QueryRow(ctx, `SELECT owner_principal_id,state,revision FROM continuity.prospective_items WHERE prospective_item_id=$1`, command.Subject.ProspectiveItemID).Scan(&owner, &state, &revision)
	if errors.Is(err, pgx.ErrNoRows) {
		return ErrProspectiveNotFound
	}
	if err != nil {
		return err
	}
	if owner != command.OwnerPrincipalID {
		return ErrProspectiveOwnerConflict
	}
	if state != continuity.ProspectiveDelivered {
		return ErrProspectiveStateConflict
	}
	if revision != command.Subject.ExpectedRevision {
		return ErrProspectiveRevisionConflict
	}
	return ErrProspectiveRevisionConflict
}