552 lines
32 KiB
Go
552 lines
32 KiB
Go
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
|
|
}
|