818 lines
32 KiB
Go
818 lines
32 KiB
Go
package phasea
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"database/sql"
|
|
_ "embed"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"regexp"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
_ "github.com/jackc/pgx/v5/stdlib"
|
|
)
|
|
|
|
//go:embed migrations/001_phase_a.sql
|
|
var migration001 string
|
|
|
|
var (
|
|
ErrConflict = errors.New("resource conflicts with existing state")
|
|
ErrInvalid = errors.New("invalid phase A input")
|
|
ErrNotFound = errors.New("resource not found")
|
|
idPattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,31}$`)
|
|
refPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$`)
|
|
platformKeyPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,127}$`)
|
|
credentialKeyPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,63}/[A-Za-z0-9][A-Za-z0-9._/-]{0,126}$`)
|
|
)
|
|
|
|
type Store struct{ db *sql.DB }
|
|
|
|
type Account struct {
|
|
ID string `json:"id"`
|
|
Platform string `json:"platform"`
|
|
PlatformAccountKey string `json:"platform_account_key"`
|
|
AuthorizationKind string `json:"authorization_kind"`
|
|
AuthorizationStatus string `json:"authorization_status"`
|
|
RuntimeStatus string `json:"runtime_status"`
|
|
CredentialReference CredentialReference `json:"credential_reference"`
|
|
CredentialKey string `json:"-"`
|
|
Version int64 `json:"version"`
|
|
}
|
|
|
|
type CredentialReference struct {
|
|
ID string `json:"id"`
|
|
Provider string `json:"provider"`
|
|
}
|
|
|
|
type RuntimeBinding struct {
|
|
ID string `json:"id"`
|
|
AccountID string `json:"account_id"`
|
|
RuntimeID string `json:"runtime_id"`
|
|
LeaseUntil time.Time `json:"lease_until"`
|
|
}
|
|
|
|
type Draft struct {
|
|
ID string `json:"id"`
|
|
AccountID string `json:"account_id"`
|
|
Version int64 `json:"version"`
|
|
Content string `json:"content"`
|
|
}
|
|
|
|
type Confirmation struct {
|
|
ID string `json:"id"`
|
|
AccountID string `json:"account_id"`
|
|
AccountVersion int64 `json:"account_version"`
|
|
DraftID string `json:"draft_id"`
|
|
DraftVersion int64 `json:"draft_version"`
|
|
Version int64 `json:"version"`
|
|
}
|
|
|
|
type Task struct {
|
|
ID string `json:"id"`
|
|
IdempotencyKey string `json:"idempotency_key"`
|
|
AccountID string `json:"account_id"`
|
|
AccountVersion int64 `json:"account_version"`
|
|
DraftID string `json:"draft_id"`
|
|
DraftVersion int64 `json:"draft_version"`
|
|
ConfirmationID string `json:"confirmation_id"`
|
|
ConfirmationVersion int64 `json:"confirmation_version"`
|
|
State string `json:"state"`
|
|
}
|
|
|
|
type Execution struct {
|
|
TaskID string `json:"task_id"`
|
|
AttemptID string `json:"attempt_id"`
|
|
AccountID string `json:"account_id"`
|
|
ConfirmationID string `json:"confirmation_id"`
|
|
ConfirmationVersion int64 `json:"confirmation_version"`
|
|
State string `json:"state"`
|
|
WasClaimed bool `json:"was_claimed"`
|
|
}
|
|
|
|
type AuditEvent struct {
|
|
ID int64 `json:"id"`
|
|
EventType string `json:"event_type"`
|
|
AccountID string `json:"account_id,omitempty"`
|
|
ConfirmationID string `json:"confirmation_id,omitempty"`
|
|
ConfirmationVersion int64 `json:"confirmation_version,omitempty"`
|
|
AttemptID string `json:"attempt_id,omitempty"`
|
|
TaskID string `json:"task_id,omitempty"`
|
|
BrowserEnvAlias string `json:"browser_env_alias,omitempty"`
|
|
NetworkExitID string `json:"network_exit_id,omitempty"`
|
|
RuntimeInstanceID string `json:"runtime_instance_id,omitempty"`
|
|
BindingVersion int64 `json:"binding_version,omitempty"`
|
|
Actor string `json:"actor,omitempty"`
|
|
ReasonCode string `json:"reason_code,omitempty"`
|
|
OperationID string `json:"operation_id,omitempty"`
|
|
Action string `json:"action,omitempty"`
|
|
Outcome string `json:"outcome,omitempty"`
|
|
OldImageVersion string `json:"old_image_version,omitempty"`
|
|
NewImageVersion string `json:"new_image_version,omitempty"`
|
|
Details json.RawMessage `json:"details"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
}
|
|
|
|
func Open(ctx context.Context, databaseURL string) (*Store, error) {
|
|
db, err := sql.Open("pgx", databaseURL)
|
|
if err != nil {
|
|
return nil, errors.New("open phase A database")
|
|
}
|
|
db.SetMaxOpenConns(10)
|
|
db.SetMaxIdleConns(2)
|
|
db.SetConnMaxIdleTime(5 * time.Minute)
|
|
if err := db.PingContext(ctx); err != nil {
|
|
db.Close()
|
|
return nil, errors.New("connect to phase A database")
|
|
}
|
|
store := &Store{db: db}
|
|
if err := store.migrate(ctx); err != nil {
|
|
db.Close()
|
|
return nil, err
|
|
}
|
|
return store, nil
|
|
}
|
|
|
|
func (s *Store) Close() error { return s.db.Close() }
|
|
|
|
func (s *Store) migrate(ctx context.Context) error {
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return errors.New("begin schema migration")
|
|
}
|
|
defer tx.Rollback()
|
|
if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(1542738011)`); err != nil {
|
|
return errors.New("lock schema migration")
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_migration (version integer PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`); err != nil {
|
|
return errors.New("create schema migration table")
|
|
}
|
|
var applied bool
|
|
if err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM schema_migration WHERE version = 1)`).Scan(&applied); err != nil {
|
|
return errors.New("read schema migration state")
|
|
}
|
|
if !applied {
|
|
if _, err := tx.ExecContext(ctx, migration001); err != nil {
|
|
return fmt.Errorf("apply schema migration 1: %w", err)
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `INSERT INTO schema_migration (version) VALUES (1)`); err != nil {
|
|
return errors.New("record schema migration 1")
|
|
}
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return errors.New("commit schema migration")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) CreateAccount(ctx context.Context, account Account) error {
|
|
if !idPattern.MatchString(account.ID) || !idPattern.MatchString(account.Platform) ||
|
|
!platformKeyPattern.MatchString(account.PlatformAccountKey) ||
|
|
(account.AuthorizationKind != "owned" && account.AuthorizationKind != "authorized") ||
|
|
!refPattern.MatchString(account.CredentialReference.ID) || !credentialKeyPattern.MatchString(account.CredentialKey) ||
|
|
(account.CredentialReference.Provider != "os_keyring" && account.CredentialReference.Provider != "secret_manager") {
|
|
return ErrInvalid
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return errors.New("begin account transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
if _, err := tx.ExecContext(ctx, `INSERT INTO credential_reference (id, provider, reference_key) VALUES ($1, $2, $3)`,
|
|
account.CredentialReference.ID, account.CredentialReference.Provider, account.CredentialKey); err != nil {
|
|
return publicDatabaseError(err)
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
INSERT INTO social_account
|
|
(id, credential_reference_id, platform, platform_account_key, authorization_kind, authorization_status, status)
|
|
VALUES ($1, $2, $3, $4, $5, 'authorized', 'paused')`, account.ID, account.CredentialReference.ID,
|
|
account.Platform, account.PlatformAccountKey, account.AuthorizationKind); err != nil {
|
|
return publicDatabaseError(err)
|
|
}
|
|
if err := appendAudit(ctx, tx, "account_created", account.ID, "", 0, "", "", map[string]string{"platform": account.Platform}); err != nil {
|
|
return err
|
|
}
|
|
return commit(tx)
|
|
}
|
|
|
|
func (s *Store) ListAccounts(ctx context.Context) ([]Account, error) {
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT account.id, account.platform, account.platform_account_key, account.authorization_kind,
|
|
account.authorization_status, account.status, reference.id, reference.provider, account.version
|
|
FROM social_account account
|
|
JOIN credential_reference reference ON reference.id = account.credential_reference_id
|
|
ORDER BY account.created_at, account.id`)
|
|
if err != nil {
|
|
return nil, errors.New("read accounts")
|
|
}
|
|
defer rows.Close()
|
|
accounts := []Account{}
|
|
for rows.Next() {
|
|
account, err := scanAccount(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
accounts = append(accounts, account)
|
|
}
|
|
return accounts, rows.Err()
|
|
}
|
|
|
|
func (s *Store) GetAccount(ctx context.Context, id string) (Account, error) {
|
|
if !idPattern.MatchString(id) {
|
|
return Account{}, ErrInvalid
|
|
}
|
|
return scanAccount(s.db.QueryRowContext(ctx, `
|
|
SELECT account.id, account.platform, account.platform_account_key, account.authorization_kind,
|
|
account.authorization_status, account.status, reference.id, reference.provider, account.version
|
|
FROM social_account account
|
|
JOIN credential_reference reference ON reference.id = account.credential_reference_id
|
|
WHERE account.id = $1`, id))
|
|
}
|
|
|
|
type accountScanner interface{ Scan(...any) error }
|
|
|
|
func scanAccount(row accountScanner) (Account, error) {
|
|
var account Account
|
|
if err := row.Scan(&account.ID, &account.Platform, &account.PlatformAccountKey, &account.AuthorizationKind,
|
|
&account.AuthorizationStatus, &account.RuntimeStatus, &account.CredentialReference.ID,
|
|
&account.CredentialReference.Provider, &account.Version); err != nil {
|
|
return Account{}, rowError(err)
|
|
}
|
|
return account, nil
|
|
}
|
|
|
|
func (s *Store) BindRuntime(ctx context.Context, binding RuntimeBinding) error {
|
|
if !refPattern.MatchString(binding.ID) || !idPattern.MatchString(binding.AccountID) ||
|
|
!refPattern.MatchString(binding.RuntimeID) || !binding.LeaseUntil.After(time.Now()) {
|
|
return ErrInvalid
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return errors.New("begin runtime transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
if _, err := tx.ExecContext(ctx, `UPDATE runtime_instance SET released_at = now() WHERE released_at IS NULL AND lease_until < now()`); err != nil {
|
|
return errors.New("expire runtime bindings")
|
|
}
|
|
var accountID string
|
|
if err := tx.QueryRowContext(ctx, `
|
|
INSERT INTO runtime_instance (id, account_id, binding_id, runtime_id, lease_until)
|
|
SELECT $1, account.id, binding.id, $3, $4
|
|
FROM social_account account
|
|
JOIN environment_binding binding ON binding.account_id = account.id
|
|
WHERE account.id = $2 AND account.status = 'active'
|
|
RETURNING account_id`, binding.ID, binding.AccountID, binding.RuntimeID, binding.LeaseUntil).Scan(&accountID); err != nil {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return ErrConflict
|
|
}
|
|
return rowError(err)
|
|
}
|
|
if err := appendAudit(ctx, tx, "runtime_bound", accountID, "", 0, "", "", map[string]string{"runtime_id": binding.RuntimeID}); err != nil {
|
|
return err
|
|
}
|
|
return commit(tx)
|
|
}
|
|
|
|
func (s *Store) ReleaseRuntime(ctx context.Context, id string) error {
|
|
if !refPattern.MatchString(id) {
|
|
return ErrInvalid
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return errors.New("begin runtime release")
|
|
}
|
|
defer tx.Rollback()
|
|
var accountID string
|
|
if err := tx.QueryRowContext(ctx, `UPDATE runtime_instance SET released_at = now() WHERE id = $1 AND released_at IS NULL RETURNING account_id`, id).Scan(&accountID); err != nil {
|
|
return rowError(err)
|
|
}
|
|
if err := appendAudit(ctx, tx, "runtime_released", accountID, "", 0, "", "", map[string]string{"binding_id": id}); err != nil {
|
|
return err
|
|
}
|
|
return commit(tx)
|
|
}
|
|
|
|
func (s *Store) CreateDraft(ctx context.Context, draft Draft) error {
|
|
if !refPattern.MatchString(draft.ID) || !idPattern.MatchString(draft.AccountID) || draft.Version < 1 || strings.TrimSpace(draft.Content) == "" {
|
|
return ErrInvalid
|
|
}
|
|
_, err := s.db.ExecContext(ctx, `INSERT INTO content_draft (id, account_id, version, content) VALUES ($1, $2, $3, $4)`,
|
|
draft.ID, draft.AccountID, draft.Version, draft.Content)
|
|
return publicDatabaseError(err)
|
|
}
|
|
|
|
func (s *Store) Confirm(ctx context.Context, confirmation Confirmation) error {
|
|
if !refPattern.MatchString(confirmation.ID) || !idPattern.MatchString(confirmation.AccountID) ||
|
|
!refPattern.MatchString(confirmation.DraftID) || confirmation.AccountVersion < 1 || confirmation.DraftVersion < 1 || confirmation.Version < 1 {
|
|
return ErrInvalid
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return errors.New("begin confirmation transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
var valid bool
|
|
if err := tx.QueryRowContext(ctx, `
|
|
SELECT EXISTS (
|
|
SELECT 1 FROM social_account a JOIN content_draft d ON d.account_id = a.id
|
|
WHERE a.id = $1 AND a.version = $2 AND d.id = $3 AND d.version = $4
|
|
)`, confirmation.AccountID, confirmation.AccountVersion, confirmation.DraftID, confirmation.DraftVersion).Scan(&valid); err != nil {
|
|
return errors.New("validate confirmation versions")
|
|
}
|
|
if !valid {
|
|
return ErrConflict
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
INSERT INTO confirmation (id, account_id, account_version, draft_id, draft_version, version)
|
|
VALUES ($1, $2, $3, $4, $5, $6)`, confirmation.ID, confirmation.AccountID, confirmation.AccountVersion,
|
|
confirmation.DraftID, confirmation.DraftVersion, confirmation.Version); err != nil {
|
|
return publicDatabaseError(err)
|
|
}
|
|
if err := appendAudit(ctx, tx, "draft_confirmed", confirmation.AccountID, confirmation.ID, confirmation.Version, "", "", map[string]any{
|
|
"account_version": confirmation.AccountVersion, "draft_id": confirmation.DraftID, "draft_version": confirmation.DraftVersion,
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
return commit(tx)
|
|
}
|
|
|
|
func (s *Store) Enqueue(ctx context.Context, task Task) (Task, bool, error) {
|
|
if !refPattern.MatchString(task.ID) || !refPattern.MatchString(task.IdempotencyKey) || !idPattern.MatchString(task.AccountID) ||
|
|
!refPattern.MatchString(task.DraftID) || task.AccountVersion < 1 || task.DraftVersion < 1 ||
|
|
((task.ConfirmationID == "") != (task.ConfirmationVersion == 0)) ||
|
|
(task.ConfirmationID != "" && (!refPattern.MatchString(task.ConfirmationID) || task.ConfirmationVersion < 1)) {
|
|
return Task{}, false, ErrInvalid
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return Task{}, false, errors.New("begin task transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
var insertedID string
|
|
err = tx.QueryRowContext(ctx, `
|
|
INSERT INTO operation_task (id, idempotency_key, account_id, account_version, draft_id, draft_version, confirmation_id, confirmation_version)
|
|
VALUES ($1, $2, $3, $4, $5, $6, NULLIF($7, ''), NULLIF($8, 0))
|
|
ON CONFLICT (idempotency_key) DO NOTHING RETURNING id`, task.ID, task.IdempotencyKey, task.AccountID, task.AccountVersion,
|
|
task.DraftID, task.DraftVersion, task.ConfirmationID, task.ConfirmationVersion).Scan(&insertedID)
|
|
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return Task{}, false, publicDatabaseError(err)
|
|
}
|
|
if insertedID != "" {
|
|
task.State = "queued"
|
|
if err := appendAudit(ctx, tx, "task_queued", task.AccountID, task.ConfirmationID, task.ConfirmationVersion, "", task.ID, nil); err != nil {
|
|
return Task{}, false, err
|
|
}
|
|
if err := commit(tx); err != nil {
|
|
return Task{}, false, err
|
|
}
|
|
return task, true, nil
|
|
}
|
|
|
|
var existing Task
|
|
var confirmationID sql.NullString
|
|
var confirmationVersion sql.NullInt64
|
|
err = tx.QueryRowContext(ctx, `
|
|
SELECT id, idempotency_key, account_id, account_version, draft_id, draft_version, confirmation_id, confirmation_version, state
|
|
FROM operation_task WHERE idempotency_key = $1`, task.IdempotencyKey).Scan(&existing.ID, &existing.IdempotencyKey,
|
|
&existing.AccountID, &existing.AccountVersion, &existing.DraftID, &existing.DraftVersion, &confirmationID, &confirmationVersion, &existing.State)
|
|
if err != nil {
|
|
return Task{}, false, rowError(err)
|
|
}
|
|
existing.ConfirmationID = confirmationID.String
|
|
existing.ConfirmationVersion = confirmationVersion.Int64
|
|
if existing.AccountID != task.AccountID || existing.AccountVersion != task.AccountVersion || existing.DraftID != task.DraftID ||
|
|
existing.DraftVersion != task.DraftVersion || existing.ConfirmationID != task.ConfirmationID || existing.ConfirmationVersion != task.ConfirmationVersion {
|
|
return Task{}, false, ErrConflict
|
|
}
|
|
if err := commit(tx); err != nil {
|
|
return Task{}, false, err
|
|
}
|
|
return existing, false, nil
|
|
}
|
|
|
|
func (s *Store) PauseAccount(ctx context.Context, accountID string) error {
|
|
return s.disableAccount(ctx, accountID, false)
|
|
}
|
|
|
|
func (s *Store) RevokeAccount(ctx context.Context, accountID string) error {
|
|
return s.disableAccount(ctx, accountID, true)
|
|
}
|
|
|
|
func (s *Store) disableAccount(ctx context.Context, accountID string, revoke bool) error {
|
|
if !idPattern.MatchString(accountID) {
|
|
return ErrInvalid
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return errors.New("begin account state transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
var version int64
|
|
var authorizationStatus, runtimeStatus string
|
|
if err := tx.QueryRowContext(ctx, `SELECT version, authorization_status, status FROM social_account WHERE id = $1 FOR UPDATE`, accountID).
|
|
Scan(&version, &authorizationStatus, &runtimeStatus); err != nil {
|
|
return rowError(err)
|
|
}
|
|
unchanged := (revoke && authorizationStatus == "revoked") || (!revoke && runtimeStatus == "paused")
|
|
if !unchanged {
|
|
if err := tx.QueryRowContext(ctx, `
|
|
UPDATE social_account
|
|
SET authorization_status = CASE WHEN $2 THEN 'revoked' ELSE authorization_status END,
|
|
status = 'paused', paused_at = now(), revoked_at = CASE WHEN $2 THEN now() ELSE revoked_at END,
|
|
version = version + 1, updated_at = now()
|
|
WHERE id = $1 RETURNING version`, accountID, revoke).Scan(&version); err != nil {
|
|
return errors.New("change account state")
|
|
}
|
|
}
|
|
held, err := holdQueuedTasks(ctx, tx, accountID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if unchanged {
|
|
return commit(tx)
|
|
}
|
|
eventType := "account_paused"
|
|
if revoke {
|
|
eventType = "account_revoked"
|
|
}
|
|
if err := appendAudit(ctx, tx, eventType, accountID, "", 0, "", "", map[string]any{"account_version": version, "tasks_held": held}); err != nil {
|
|
return err
|
|
}
|
|
return commit(tx)
|
|
}
|
|
|
|
func (s *Store) ResumeAccount(ctx context.Context, accountID string) error {
|
|
if !idPattern.MatchString(accountID) {
|
|
return ErrInvalid
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return errors.New("begin resume transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
var authorizationStatus, runtimeStatus string
|
|
if err := tx.QueryRowContext(ctx, `SELECT authorization_status, status FROM social_account WHERE id = $1 FOR UPDATE`, accountID).
|
|
Scan(&authorizationStatus, &runtimeStatus); err != nil {
|
|
return rowError(err)
|
|
}
|
|
if authorizationStatus == "revoked" {
|
|
return ErrConflict
|
|
}
|
|
var ready bool
|
|
if err := tx.QueryRowContext(ctx, `
|
|
SELECT EXISTS (
|
|
SELECT 1 FROM environment_binding binding
|
|
JOIN network_exit network ON network.id = binding.network_exit_id
|
|
WHERE binding.account_id = $1 AND network.health_status = 'healthy'
|
|
)`, accountID).Scan(&ready); err != nil {
|
|
return errors.New("validate account binding")
|
|
}
|
|
if !ready {
|
|
return ErrConflict
|
|
}
|
|
if runtimeStatus == "active" {
|
|
return commit(tx)
|
|
}
|
|
var version int64
|
|
if err := tx.QueryRowContext(ctx, `
|
|
UPDATE social_account SET status = 'active', paused_at = NULL, version = version + 1, updated_at = now()
|
|
WHERE id = $1 RETURNING version`, accountID).Scan(&version); err != nil {
|
|
return errors.New("resume account")
|
|
}
|
|
if err := appendAudit(ctx, tx, "account_resumed", accountID, "", 0, "", "", map[string]any{"account_version": version}); err != nil {
|
|
return err
|
|
}
|
|
return commit(tx)
|
|
}
|
|
|
|
func holdQueuedTasks(ctx context.Context, tx *sql.Tx, accountID string) (int64, error) {
|
|
result, err := tx.ExecContext(ctx, `
|
|
UPDATE operation_task SET state = 'policy_hold', updated_at = now()
|
|
WHERE account_id = $1 AND state = 'queued'`, accountID)
|
|
if err != nil {
|
|
return 0, errors.New("hold queued account tasks")
|
|
}
|
|
held, _ := result.RowsAffected()
|
|
return held, nil
|
|
}
|
|
|
|
func (s *Store) CancelTask(ctx context.Context, taskID string) error {
|
|
if !refPattern.MatchString(taskID) {
|
|
return ErrInvalid
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return errors.New("begin cancel transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
var accountID, state string
|
|
var attemptID sql.NullString
|
|
var confirmationID sql.NullString
|
|
var confirmationVersion sql.NullInt64
|
|
err = tx.QueryRowContext(ctx, `
|
|
UPDATE operation_task SET state = CASE WHEN state = 'executing' THEN 'needs_confirmation' ELSE 'cancelled' END,
|
|
lease_owner = NULL, lease_until = NULL, updated_at = now()
|
|
WHERE id = $1 AND state IN ('queued', 'executing', 'needs_confirmation', 'policy_hold')
|
|
RETURNING account_id, state, current_attempt_id, confirmation_id, confirmation_version`, taskID).Scan(
|
|
&accountID, &state, &attemptID, &confirmationID, &confirmationVersion)
|
|
if err != nil {
|
|
return rowError(err)
|
|
}
|
|
if attemptID.Valid {
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE execution_attempt SET finished_at = now(), outcome = 'uncertain'
|
|
WHERE id = $1 AND finished_at IS NULL`, attemptID.String); err != nil {
|
|
return errors.New("mark cancelled attempt uncertain")
|
|
}
|
|
}
|
|
if err := appendAudit(ctx, tx, "task_cancel_requested", accountID, confirmationID.String, confirmationVersion.Int64, attemptID.String, taskID, map[string]string{"state": state}); err != nil {
|
|
return err
|
|
}
|
|
return commit(tx)
|
|
}
|
|
|
|
func (s *Store) ExecuteMock(ctx context.Context, workerID, outcome string) (Execution, error) {
|
|
if !refPattern.MatchString(workerID) || (outcome != "succeeded" && outcome != "failed" && outcome != "uncertain" && outcome != "policy_hold") {
|
|
return Execution{}, ErrInvalid
|
|
}
|
|
execution, err := s.claim(ctx, workerID)
|
|
if err != nil || !execution.WasClaimed {
|
|
return execution, err
|
|
}
|
|
return s.complete(ctx, workerID, execution, outcome)
|
|
}
|
|
|
|
func (s *Store) claim(ctx context.Context, workerID string) (Execution, error) {
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return Execution{}, errors.New("begin claim transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
if err := quarantineExpired(ctx, tx); err != nil {
|
|
return Execution{}, err
|
|
}
|
|
if err := quarantineInvalid(ctx, tx); err != nil {
|
|
return Execution{}, err
|
|
}
|
|
var execution Execution
|
|
err = tx.QueryRowContext(ctx, `
|
|
WITH candidate AS (
|
|
SELECT t.id
|
|
FROM operation_task t
|
|
JOIN social_account a ON a.id = t.account_id
|
|
JOIN content_draft d ON d.id = t.draft_id
|
|
JOIN confirmation c ON c.id = t.confirmation_id
|
|
WHERE t.state = 'queued' AND a.status = 'active'
|
|
AND a.version = t.account_version
|
|
AND d.account_id = t.account_id AND d.version = t.draft_version
|
|
AND c.account_id = t.account_id AND c.account_version = t.account_version
|
|
AND c.draft_id = t.draft_id AND c.draft_version = t.draft_version
|
|
AND c.version = t.confirmation_version
|
|
ORDER BY t.created_at, t.id
|
|
FOR UPDATE OF t SKIP LOCKED LIMIT 1
|
|
)
|
|
UPDATE operation_task t SET state = 'executing', lease_owner = $1,
|
|
lease_until = now() + interval '1 minute', updated_at = now()
|
|
FROM candidate WHERE t.id = candidate.id
|
|
RETURNING t.id, t.account_id, t.confirmation_id, t.confirmation_version`, workerID).Scan(
|
|
&execution.TaskID, &execution.AccountID, &execution.ConfirmationID, &execution.ConfirmationVersion)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
if err := commit(tx); err != nil {
|
|
return Execution{}, err
|
|
}
|
|
return Execution{}, nil
|
|
}
|
|
if err != nil {
|
|
return Execution{}, errors.New("claim task")
|
|
}
|
|
execution.AttemptID = newID()
|
|
execution.State = "executing"
|
|
execution.WasClaimed = true
|
|
if _, err := tx.ExecContext(ctx, `INSERT INTO execution_attempt (id, task_id) VALUES ($1, $2)`, execution.AttemptID, execution.TaskID); err != nil {
|
|
return Execution{}, publicDatabaseError(err)
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `UPDATE operation_task SET current_attempt_id = $1 WHERE id = $2`, execution.AttemptID, execution.TaskID); err != nil {
|
|
return Execution{}, errors.New("attach execution attempt")
|
|
}
|
|
if err := appendAudit(ctx, tx, "task_claimed", execution.AccountID, execution.ConfirmationID, execution.ConfirmationVersion, execution.AttemptID, execution.TaskID, map[string]string{"worker_id": workerID}); err != nil {
|
|
return Execution{}, err
|
|
}
|
|
if err := commit(tx); err != nil {
|
|
return Execution{}, err
|
|
}
|
|
return execution, nil
|
|
}
|
|
|
|
func (s *Store) complete(ctx context.Context, workerID string, execution Execution, outcome string) (Execution, error) {
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return Execution{}, errors.New("begin completion transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
var leaseUntil time.Time
|
|
err = tx.QueryRowContext(ctx, `
|
|
SELECT lease_until FROM operation_task
|
|
WHERE id = $1 AND state = 'executing' AND lease_owner = $2 AND current_attempt_id = $3
|
|
FOR UPDATE`, execution.TaskID, workerID, execution.AttemptID).Scan(&leaseUntil)
|
|
if err != nil {
|
|
return Execution{}, rowError(err)
|
|
}
|
|
state := map[string]string{"succeeded": "succeeded", "failed": "failed", "uncertain": "needs_confirmation", "policy_hold": "policy_hold"}[outcome]
|
|
if leaseUntil.Before(time.Now()) {
|
|
outcome, state = "uncertain", "needs_confirmation"
|
|
}
|
|
result, _ := json.Marshal(map[string]string{"mock_outcome": outcome})
|
|
if _, err := tx.ExecContext(ctx, `UPDATE execution_attempt SET finished_at = now(), outcome = $1, result = $2 WHERE id = $3`, outcome, result, execution.AttemptID); err != nil {
|
|
return Execution{}, errors.New("finish execution attempt")
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE operation_task SET state = $1, lease_owner = NULL, lease_until = NULL, updated_at = now() WHERE id = $2`, state, execution.TaskID); err != nil {
|
|
return Execution{}, errors.New("finish task")
|
|
}
|
|
if err := appendAudit(ctx, tx, "task_finished", execution.AccountID, execution.ConfirmationID, execution.ConfirmationVersion, execution.AttemptID, execution.TaskID, map[string]string{"state": state}); err != nil {
|
|
return Execution{}, err
|
|
}
|
|
if err := commit(tx); err != nil {
|
|
return Execution{}, err
|
|
}
|
|
execution.State = state
|
|
return execution, nil
|
|
}
|
|
|
|
func quarantineExpired(ctx context.Context, tx *sql.Tx) error {
|
|
rows, err := tx.QueryContext(ctx, `
|
|
UPDATE operation_task SET state = 'needs_confirmation', lease_owner = NULL, lease_until = NULL, updated_at = now()
|
|
WHERE state = 'executing' AND lease_until < now()
|
|
RETURNING id, account_id, current_attempt_id, confirmation_id, confirmation_version`)
|
|
if err != nil {
|
|
return errors.New("quarantine expired tasks")
|
|
}
|
|
type expiredTask struct {
|
|
taskID, accountID, attemptID, confirmationID string
|
|
confirmationVersion int64
|
|
}
|
|
var expired []expiredTask
|
|
for rows.Next() {
|
|
var taskID, accountID string
|
|
var attemptID sql.NullString
|
|
var confirmationID sql.NullString
|
|
var confirmationVersion sql.NullInt64
|
|
if err := rows.Scan(&taskID, &accountID, &attemptID, &confirmationID, &confirmationVersion); err != nil {
|
|
rows.Close()
|
|
return errors.New("read expired task")
|
|
}
|
|
expired = append(expired, expiredTask{taskID, accountID, attemptID.String, confirmationID.String, confirmationVersion.Int64})
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return errors.New("close expired task rows")
|
|
}
|
|
for _, task := range expired {
|
|
if task.attemptID != "" {
|
|
if _, err := tx.ExecContext(ctx, `UPDATE execution_attempt SET finished_at = now(), outcome = 'uncertain' WHERE id = $1 AND finished_at IS NULL`, task.attemptID); err != nil {
|
|
return errors.New("mark expired attempt uncertain")
|
|
}
|
|
}
|
|
if err := appendAudit(ctx, tx, "task_lease_expired", task.accountID, task.confirmationID, task.confirmationVersion, task.attemptID, task.taskID, nil); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func quarantineInvalid(ctx context.Context, tx *sql.Tx) error {
|
|
rows, err := tx.QueryContext(ctx, `
|
|
UPDATE operation_task t SET
|
|
state = CASE WHEN EXISTS (
|
|
SELECT 1 FROM social_account paused WHERE paused.id = t.account_id AND paused.status = 'paused'
|
|
) THEN 'policy_hold' ELSE 'needs_confirmation' END,
|
|
updated_at = now()
|
|
WHERE t.state = 'queued' AND NOT EXISTS (
|
|
SELECT 1
|
|
FROM social_account a
|
|
JOIN content_draft d ON d.id = t.draft_id
|
|
JOIN confirmation c ON c.id = t.confirmation_id
|
|
WHERE a.id = t.account_id AND a.status = 'active' AND a.version = t.account_version
|
|
AND d.account_id = t.account_id AND d.version = t.draft_version
|
|
AND c.account_id = t.account_id AND c.account_version = t.account_version
|
|
AND c.draft_id = t.draft_id AND c.draft_version = t.draft_version
|
|
AND c.version = t.confirmation_version
|
|
)
|
|
RETURNING t.id, t.account_id, t.confirmation_id, t.confirmation_version, t.state`)
|
|
if err != nil {
|
|
return errors.New("quarantine invalid tasks")
|
|
}
|
|
type invalidTask struct {
|
|
taskID, accountID, confirmationID, state string
|
|
confirmationVersion int64
|
|
}
|
|
var invalid []invalidTask
|
|
for rows.Next() {
|
|
var taskID, accountID, state string
|
|
var confirmationID sql.NullString
|
|
var confirmationVersion sql.NullInt64
|
|
if err := rows.Scan(&taskID, &accountID, &confirmationID, &confirmationVersion, &state); err != nil {
|
|
rows.Close()
|
|
return errors.New("read quarantined task")
|
|
}
|
|
invalid = append(invalid, invalidTask{taskID, accountID, confirmationID.String, state, confirmationVersion.Int64})
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return errors.New("close quarantined task rows")
|
|
}
|
|
for _, task := range invalid {
|
|
if err := appendAudit(ctx, tx, "task_quarantined", task.accountID, task.confirmationID, task.confirmationVersion, "", task.taskID, map[string]string{"state": task.state}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) Audit(ctx context.Context) ([]AuditEvent, error) {
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT id, event_type, account_id, confirmation_id, confirmation_version, attempt_id, task_id,
|
|
browser_env_alias, network_exit_id, runtime_instance_id, binding_version, actor, reason_code,
|
|
operation_id, action, outcome, old_image_version, new_image_version,
|
|
details, created_at
|
|
FROM audit_event ORDER BY id`)
|
|
if err != nil {
|
|
return nil, errors.New("read audit events")
|
|
}
|
|
defer rows.Close()
|
|
var events []AuditEvent
|
|
for rows.Next() {
|
|
var event AuditEvent
|
|
var accountID, confirmationID, attemptID, taskID, browserEnvAlias, networkExitID sql.NullString
|
|
var runtimeInstanceID, actor, reasonCode, operationID, action, outcome, oldImage, newImage sql.NullString
|
|
var confirmationVersion, bindingVersion sql.NullInt64
|
|
if err := rows.Scan(&event.ID, &event.EventType, &accountID, &confirmationID, &confirmationVersion, &attemptID, &taskID,
|
|
&browserEnvAlias, &networkExitID, &runtimeInstanceID, &bindingVersion, &actor, &reasonCode,
|
|
&operationID, &action, &outcome, &oldImage, &newImage,
|
|
&event.Details, &event.CreatedAt); err != nil {
|
|
return nil, errors.New("decode audit event")
|
|
}
|
|
event.AccountID, event.ConfirmationID, event.ConfirmationVersion = accountID.String, confirmationID.String, confirmationVersion.Int64
|
|
event.AttemptID, event.TaskID = attemptID.String, taskID.String
|
|
event.BrowserEnvAlias, event.NetworkExitID = browserEnvAlias.String, networkExitID.String
|
|
event.RuntimeInstanceID, event.BindingVersion = runtimeInstanceID.String, bindingVersion.Int64
|
|
event.Actor, event.ReasonCode = actor.String, reasonCode.String
|
|
event.OperationID, event.Action, event.Outcome = operationID.String, action.String, outcome.String
|
|
event.OldImageVersion, event.NewImageVersion = oldImage.String, newImage.String
|
|
events = append(events, event)
|
|
}
|
|
return events, rows.Err()
|
|
}
|
|
|
|
func appendAudit(ctx context.Context, tx *sql.Tx, eventType, accountID, confirmationID string, confirmationVersion int64, attemptID, taskID string, details any) error {
|
|
if details == nil {
|
|
details = map[string]any{}
|
|
}
|
|
encoded, err := json.Marshal(details)
|
|
if err != nil {
|
|
return errors.New("encode audit details")
|
|
}
|
|
_, err = tx.ExecContext(ctx, `
|
|
INSERT INTO audit_event
|
|
(event_type, account_id, confirmation_id, confirmation_version, attempt_id, task_id, actor, reason_code, details)
|
|
VALUES ($1, NULLIF($2, ''), NULLIF($3, ''), NULLIF($4, 0), NULLIF($5, ''), NULLIF($6, ''), 'local-user', $1, $7)`,
|
|
eventType, accountID, confirmationID, confirmationVersion, attemptID, taskID, encoded)
|
|
if err != nil {
|
|
return errors.New("append audit event")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func newID() string {
|
|
var value [16]byte
|
|
_, _ = rand.Read(value[:])
|
|
return hex.EncodeToString(value[:])
|
|
}
|
|
|
|
func commit(tx *sql.Tx) error {
|
|
if err := tx.Commit(); err != nil {
|
|
return errors.New("commit transaction")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func rowError(err error) error {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return ErrNotFound
|
|
}
|
|
return publicDatabaseError(err)
|
|
}
|
|
|
|
func publicDatabaseError(err error) error {
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
var postgresError *pgconn.PgError
|
|
if errors.As(err, &postgresError) && (postgresError.Code == "23505" || postgresError.Code == "23503" || postgresError.Code == "23514") {
|
|
return ErrConflict
|
|
}
|
|
return errors.New("phase A persistence operation failed")
|
|
}
|