209 lines
10 KiB
Go
209 lines
10 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"reflect"
|
|
|
|
commitcontract "git.cin.su/m/Assistant/contracts/commit"
|
|
"git.cin.su/m/Assistant/contracts/continuity"
|
|
"git.cin.su/m/Assistant/contracts/identity"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
var (
|
|
ErrAuthorizedActionNotFound = errors.New("authorized action not found")
|
|
ErrAuthorizedActionConflict = errors.New("authorized action identity conflict")
|
|
ErrActionAttemptNotFound = errors.New("action attempt not found")
|
|
ErrActionAttemptConflict = errors.New("action attempt identity conflict")
|
|
)
|
|
|
|
type CommitRepository struct{ pool *pgxpool.Pool }
|
|
|
|
func OpenCommit(ctx context.Context, c ContinuityConfig) (*CommitRepository, error) {
|
|
p, e := openPool(ctx, c)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
return &CommitRepository{p}, nil
|
|
}
|
|
func (r *CommitRepository) Close() { r.pool.Close() }
|
|
func (r *CommitRepository) SaveAuthorized(ctx context.Context, a commitcontract.AuthorizedAction) error {
|
|
prospectiveItemID := a.Proposal.ContinuitySubject.ProspectiveItemID
|
|
expectedRevision := a.Proposal.ContinuitySubject.ExpectedRevision
|
|
if a.Proposal.Operation == commitcontract.OperationDeliverProspectiveNotification {
|
|
prospectiveItemID = a.Proposal.NotificationSubject.ProspectiveItemID
|
|
expectedRevision = a.Proposal.NotificationSubject.ExpectedRevision
|
|
}
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
if _, err = tx.Exec(ctx, `INSERT INTO commit.authorized_actions (authorized_action_id,action_intent_id,state,revision) VALUES ($1,$2,$3,1) ON CONFLICT (authorized_action_id) DO NOTHING`, a.ID, a.Proposal.IntentID, a.State); err != nil {
|
|
return err
|
|
}
|
|
result, err := tx.Exec(ctx, `INSERT INTO commit.authorized_action_envelopes (
|
|
authorized_action_id,action_intent_id,actor_id,principal_id,grant_id,grant_revision,scope_id,
|
|
task_id,goal_id,task_scope_id,task_revision,capability_id,operation,prospective_item_id,
|
|
expected_prospective_revision,prospective_from_state,prospective_to_state,evidence_ref,
|
|
proposal_state,action_state,authorized_at_unix_millis,envelope_revision,notification_template_code,
|
|
capability_binding_id,capability_binding_revision,notification_title,notification_message,request_digest)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,1,$22,$23,$24,$25,$26,$27)
|
|
ON CONFLICT DO NOTHING`,
|
|
a.ID, a.Proposal.IntentID, a.Proposal.ActorID, a.PrincipalID, a.GrantID, a.GrantRevision, a.ScopeID,
|
|
a.Proposal.Target.TaskID, a.Proposal.Target.GoalID, a.Proposal.Target.ScopeID, a.Proposal.Target.Revision,
|
|
a.Proposal.CapabilityID, a.Proposal.Operation, prospectiveItemID,
|
|
expectedRevision, a.Proposal.ContinuitySubject.FromState,
|
|
a.Proposal.ContinuitySubject.ToState, a.Proposal.ContinuitySubject.EvidenceRef,
|
|
a.Proposal.State, a.State, a.AuthorizedAtUnixMillis, a.Proposal.NotificationSubject.TemplateCode,
|
|
a.Proposal.NotificationSubject.BindingID, a.Proposal.NotificationSubject.BindingRevision,
|
|
a.Proposal.NotificationSubject.RenderedTitle, a.Proposal.NotificationSubject.RenderedMessage,
|
|
a.Proposal.NotificationSubject.RequestDigest)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if result.RowsAffected() == 0 {
|
|
existing, loadErr := loadAuthorized(ctx, tx, a.ID)
|
|
if loadErr != nil || !reflect.DeepEqual(existing, a) {
|
|
return ErrAuthorizedActionConflict
|
|
}
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
func (r *CommitRepository) SaveAttempt(ctx context.Context, a commitcontract.ActionAttempt) error {
|
|
result, err := r.pool.Exec(ctx, `INSERT INTO commit.action_attempts (action_attempt_id,authorized_action_id,action_intent_id,state) VALUES ($1,$2,$3,$4) ON CONFLICT (action_attempt_id) DO NOTHING`, a.ID, a.AuthorizedID, a.IntentID, a.State)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if result.RowsAffected() == 0 {
|
|
var existing commitcontract.ActionAttempt
|
|
err = r.pool.QueryRow(ctx, `SELECT action_attempt_id,authorized_action_id,action_intent_id,state FROM commit.action_attempts WHERE action_attempt_id=$1`, a.ID).Scan(&existing.ID, &existing.AuthorizedID, &existing.IntentID, &existing.State)
|
|
if err != nil || existing != a {
|
|
return ErrActionAttemptConflict
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *CommitRepository) BeginAttempt(ctx context.Context, a commitcontract.ActionAttempt) error {
|
|
if a.State != commitcontract.ActionExecuting {
|
|
return ErrActionAttemptConflict
|
|
}
|
|
return r.SaveAttempt(ctx, a)
|
|
}
|
|
|
|
func (r *CommitRepository) FinishAttempt(ctx context.Context, a commitcontract.ActionAttempt) error {
|
|
if a.State != commitcontract.ActionSucceeded && a.State != commitcontract.ActionFailed && a.State != commitcontract.ActionUnknownOutcome {
|
|
return ErrActionAttemptConflict
|
|
}
|
|
result, err := r.pool.Exec(ctx, `UPDATE commit.action_attempts SET state=$1,revision=revision+1,updated_at=CURRENT_TIMESTAMP
|
|
WHERE action_attempt_id=$2 AND authorized_action_id=$3 AND action_intent_id=$4 AND state=$5`,
|
|
a.State, a.ID, a.AuthorizedID, a.IntentID, commitcontract.ActionExecuting)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if result.RowsAffected() != 1 {
|
|
return ErrActionAttemptConflict
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *CommitRepository) LoadAuthorized(ctx context.Context, id commitcontract.AuthorizedActionID) (commitcontract.AuthorizedAction, error) {
|
|
return loadAuthorized(ctx, r.pool, id)
|
|
}
|
|
|
|
func (r *CommitRepository) LoadLatestAttempt(ctx context.Context, id commitcontract.AuthorizedActionID) (commitcontract.ActionAttempt, error) {
|
|
var attempt commitcontract.ActionAttempt
|
|
err := r.pool.QueryRow(ctx, `SELECT action_attempt_id,authorized_action_id,action_intent_id,state FROM commit.action_attempts WHERE authorized_action_id=$1 ORDER BY recorded_at DESC, action_attempt_id DESC LIMIT 1`, id).Scan(&attempt.ID, &attempt.AuthorizedID, &attempt.IntentID, &attempt.State)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return commitcontract.ActionAttempt{}, ErrActionAttemptNotFound
|
|
}
|
|
return attempt, err
|
|
}
|
|
|
|
func (r *CommitRepository) FindLatestAttempt(ctx context.Context, id commitcontract.AuthorizedActionID) (commitcontract.ActionAttempt, bool, error) {
|
|
attempt, err := r.LoadLatestAttempt(ctx, id)
|
|
if errors.Is(err, ErrActionAttemptNotFound) {
|
|
return commitcontract.ActionAttempt{}, false, nil
|
|
}
|
|
return attempt, err == nil, err
|
|
}
|
|
|
|
func (r *CommitRepository) ListAuthorizedByOwner(ctx context.Context, actorID identity.ActorID, principalID identity.PrincipalID, limit int) ([]commitcontract.AuthorizedAction, error) {
|
|
if actorID == "" || principalID == "" || limit < 1 || limit > 50 {
|
|
return nil, fmt.Errorf("invalid authorized action owner query")
|
|
}
|
|
rows, err := r.pool.Query(ctx, `SELECT authorized_action_id,action_intent_id,actor_id,principal_id,grant_id,grant_revision,scope_id,
|
|
task_id,goal_id,task_scope_id,task_revision,capability_id,operation,prospective_item_id,
|
|
expected_prospective_revision,prospective_from_state,prospective_to_state,evidence_ref,
|
|
proposal_state,action_state,authorized_at_unix_millis,notification_template_code,capability_binding_id,
|
|
capability_binding_revision,notification_title,notification_message,request_digest
|
|
FROM commit.authorized_action_envelopes
|
|
WHERE actor_id=$1 AND principal_id=$2
|
|
ORDER BY authorized_at_unix_millis DESC, authorized_action_id DESC LIMIT $3`, actorID, principalID, limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list authorized actions: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
actions := make([]commitcontract.AuthorizedAction, 0)
|
|
for rows.Next() {
|
|
var action commitcontract.AuthorizedAction
|
|
if err := scanAuthorized(rows, &action); err != nil {
|
|
return nil, fmt.Errorf("scan authorized action: %w", err)
|
|
}
|
|
actions = append(actions, action)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("list authorized actions: %w", err)
|
|
}
|
|
return actions, nil
|
|
}
|
|
|
|
type commitRowQuerier interface {
|
|
QueryRow(context.Context, string, ...any) pgx.Row
|
|
}
|
|
|
|
type commitRowScanner interface {
|
|
Scan(...any) error
|
|
}
|
|
|
|
func loadAuthorized(ctx context.Context, querier commitRowQuerier, id commitcontract.AuthorizedActionID) (commitcontract.AuthorizedAction, error) {
|
|
var action commitcontract.AuthorizedAction
|
|
err := scanAuthorized(querier.QueryRow(ctx, `SELECT authorized_action_id,action_intent_id,actor_id,principal_id,grant_id,grant_revision,scope_id,
|
|
task_id,goal_id,task_scope_id,task_revision,capability_id,operation,prospective_item_id,
|
|
expected_prospective_revision,prospective_from_state,prospective_to_state,evidence_ref,
|
|
proposal_state,action_state,authorized_at_unix_millis,notification_template_code,capability_binding_id,
|
|
capability_binding_revision,notification_title,notification_message,request_digest
|
|
FROM commit.authorized_action_envelopes WHERE authorized_action_id=$1`, id), &action)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return commitcontract.AuthorizedAction{}, ErrAuthorizedActionNotFound
|
|
}
|
|
if err != nil {
|
|
return commitcontract.AuthorizedAction{}, fmt.Errorf("load authorized action: %w", err)
|
|
}
|
|
return action, nil
|
|
}
|
|
|
|
func scanAuthorized(row commitRowScanner, action *commitcontract.AuthorizedAction) error {
|
|
err := row.Scan(
|
|
&action.ID, &action.Proposal.IntentID, &action.Proposal.ActorID, &action.PrincipalID,
|
|
&action.GrantID, &action.GrantRevision, &action.ScopeID, &action.Proposal.Target.TaskID,
|
|
&action.Proposal.Target.GoalID, &action.Proposal.Target.ScopeID, &action.Proposal.Target.Revision,
|
|
&action.Proposal.CapabilityID, &action.Proposal.Operation, &action.Proposal.ContinuitySubject.ProspectiveItemID,
|
|
&action.Proposal.ContinuitySubject.ExpectedRevision, &action.Proposal.ContinuitySubject.FromState,
|
|
&action.Proposal.ContinuitySubject.ToState, &action.Proposal.ContinuitySubject.EvidenceRef,
|
|
&action.Proposal.State, &action.State, &action.AuthorizedAtUnixMillis,
|
|
&action.Proposal.NotificationSubject.TemplateCode, &action.Proposal.NotificationSubject.BindingID,
|
|
&action.Proposal.NotificationSubject.BindingRevision, &action.Proposal.NotificationSubject.RenderedTitle,
|
|
&action.Proposal.NotificationSubject.RenderedMessage, &action.Proposal.NotificationSubject.RequestDigest)
|
|
if err == nil && action.Proposal.Operation == commitcontract.OperationDeliverProspectiveNotification {
|
|
action.Proposal.NotificationSubject.ProspectiveItemID = action.Proposal.ContinuitySubject.ProspectiveItemID
|
|
action.Proposal.NotificationSubject.ExpectedRevision = action.Proposal.ContinuitySubject.ExpectedRevision
|
|
action.Proposal.ContinuitySubject = continuity.SatisfyDeliveredSubject{}
|
|
}
|
|
return err
|
|
}
|