182 lines
9.0 KiB
Go
182 lines
9.0 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/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")
|
|
)
|
|
|
|
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, trigger_at_unix_millis, state, revision) VALUES ($1,$2,$3,$4,$5,$6,$7,$8) 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.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, 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.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, 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.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
|
|
}
|
|
|
|
// 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,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.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
|
|
}
|
|
|
|
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
|
|
}
|