package postgres

import (
	"context"
	"errors"
	"fmt"

	"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 OpenContinuity(ctx context.Context, config ContinuityConfig) (*ContinuityRepository, error) {
	dsn := fmt.Sprintf("host=%s port=%s dbname=%s user=%s password=%s", config.Host, config.Port, config.Database, config.User, config.Password)
	pool, err := pgxpool.New(ctx, dsn)
	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,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,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.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,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,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 > 10 {
		return nil, fmt.Errorf("invalid response candidate query")
	}
	rows, err := r.pool.Query(ctx, `SELECT
		a.attention_item_id,a.owner_principal_id,a.scope_id,a.source_kind,a.source_prospective_item_id,a.level,a.state,a.summary,
		a.due_at_unix_millis,a.created_at_unix_millis,a.delivered_at_unix_millis,a.action_intent_id,a.authorized_action_id,a.action_attempt_id,a.delivery_outcome,
		a.response_id,a.response_utterance_id,a.response_operation,a.response_outcome,a.responded_at_unix_millis,a.revision,
		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.attention_items a
		JOIN continuity.prospective_items p ON p.prospective_item_id=a.source_prospective_item_id
		WHERE a.owner_principal_id=$1
		  AND a.source_kind=$2
		  AND a.state IN ($3,$4,$5)
		  AND p.state IN ($6,$7)
		ORDER BY a.delivered_at_unix_millis DESC, a.due_at_unix_millis DESC, a.attention_item_id DESC
		LIMIT $8`,
		owner, continuity.AttentionSourceProspectiveReminder, continuity.AttentionDelivered, continuity.AttentionPending,
		continuity.AttentionFailed, continuity.ProspectiveDelivered, continuity.ProspectiveArmed, limit)
	if err != nil {
		return nil, fmt.Errorf("list human response candidates: %w", err)
	}
	defer rows.Close()
	var candidates []continuitydomain.ResponseCandidate
	for rows.Next() {
		var candidate continuitydomain.ResponseCandidate
		if err := rows.Scan(
			&candidate.Attention.ID, &candidate.Attention.OwnerPrincipalID, &candidate.Attention.ScopeID, &candidate.Attention.SourceKind,
			&candidate.Attention.SourceProspectiveID, &candidate.Attention.Level, &candidate.Attention.State, &candidate.Attention.Summary,
			&candidate.Attention.DueAtUnixMillis, &candidate.Attention.CreatedAtUnixMillis, &candidate.Attention.DeliveredAtUnixMillis,
			&candidate.Attention.ActionIntentID, &candidate.Attention.AuthorizedActionID, &candidate.Attention.ActionAttemptID,
			&candidate.Attention.DeliveryOutcome, &candidate.Attention.ResponseID, &candidate.Attention.ResponseUtteranceID,
			&candidate.Attention.ResponseOperation, &candidate.Attention.ResponseOutcome, &candidate.Attention.RespondedAtUnixMillis,
			&candidate.Attention.Revision,
			&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)
		}
		candidates = append(candidates, candidate)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list human response candidates: %w", err)
	}
	return candidates, 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 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 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 err = markAttentionResponded(ctx, tx, response, continuity.AttentionSuppressed); err != nil {
			return err
		}
	case continuity.ResponseAcknowledge, continuity.ResponseDismiss:
		if err = markAttentionResponded(ctx, tx, response, continuity.AttentionResolved); err != nil {
			return err
		}
	}
	return tx.Commit(ctx)
}

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,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.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,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.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,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.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.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
}