rogee
1e92e16ed3
- AccountList.createAccount 成功后关闭创建弹窗（此前只提示不关闭）
- Cookies 前端改为可选：留空代表扫码登录场景，提交时不携带该字段
- 后端 validAccount 允许空凭据（此前 http.ParseCookie("") 直接拒绝）
- CreateAccount 空凭据时跳过 keyring 写入，不产生空值记录；回滚补偿仅在实际写入过凭据时执行
验证：go test ./... / go vet ./... / 双端构建通过；web vitest 66 用例通过（新增无 cookies 创建+弹窗关闭用例）；npm run build 通过
1939 lines
80 KiB
Go
1939 lines
80 KiB
Go
package phasea
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
_ "embed"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"regexp"
|
|
"strings"
|
|
"time"
|
|
"unicode/utf8"
|
|
|
|
"git.ipao.vip/rogee/creator-hub/internal/taskstate"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
_ "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")
|
|
ErrAccountCreationUnknown = errors.New("account creation result is unknown")
|
|
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}$`)
|
|
eventPattern = regexp.MustCompile(`^[a-z0-9_]{1,64}$`)
|
|
)
|
|
|
|
type Store struct {
|
|
db *sql.DB
|
|
notify taskstate.Notifier
|
|
accountCommit func(*sql.Tx) error
|
|
}
|
|
|
|
type Account struct {
|
|
ID string `json:"id"`
|
|
Name string `json:"name"`
|
|
Platform string `json:"platform"`
|
|
PlatformAccountKey string `json:"platform_account_key"`
|
|
Tags []string `json:"tags"`
|
|
Cookies string `json:"-"`
|
|
CredentialReference CredentialReference `json:"-"`
|
|
CredentialKey string `json:"-"`
|
|
AuthorizationStatus string `json:"authorization_status"`
|
|
RuntimeStatus string `json:"runtime_status"`
|
|
Version int64 `json:"version"`
|
|
}
|
|
|
|
type CredentialReference struct {
|
|
ID string
|
|
Provider string
|
|
}
|
|
|
|
type CredentialBridge interface {
|
|
// Store may fail after a partial write; Delete must be idempotent for compensation.
|
|
Store(context.Context, CredentialReference, string, string) error
|
|
Delete(context.Context, CredentialReference, string) error
|
|
}
|
|
|
|
type Draft struct {
|
|
ID string `json:"id"`
|
|
AccountID string `json:"account_id"`
|
|
Version int64 `json:"version"`
|
|
Content string `json:"content"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
}
|
|
|
|
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"`
|
|
ConfirmedAt time.Time `json:"confirmed_at"`
|
|
}
|
|
|
|
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"`
|
|
HoldReason string `json:"hold_reason,omitempty"`
|
|
VerificationResult string `json:"verification_result,omitempty"`
|
|
VerifiedAt *time.Time `json:"verified_at,omitempty"`
|
|
VerifiedBy string `json:"verified_by,omitempty"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
}
|
|
|
|
type TaskAttempt struct {
|
|
ID string `json:"id"`
|
|
TaskID string `json:"task_id,omitempty"`
|
|
StartedAt time.Time `json:"started_at"`
|
|
FinishedAt *time.Time `json:"finished_at,omitempty"`
|
|
Outcome string `json:"outcome,omitempty"`
|
|
Evidence map[string]string `json:"evidence"`
|
|
}
|
|
|
|
type TaskAttemptDetail struct {
|
|
TaskAttempt
|
|
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"`
|
|
}
|
|
|
|
type TaskDetail struct {
|
|
Task
|
|
Confirmation ConfirmationSnapshot `json:"confirmation"`
|
|
Attempts []TaskAttempt `json:"attempts"`
|
|
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"`
|
|
AllowedAction string `json:"allowed_action,omitempty"`
|
|
ReadinessReason string `json:"readiness_reason,omitempty"`
|
|
}
|
|
|
|
type ConfirmationSnapshot struct {
|
|
Confirmation
|
|
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"`
|
|
}
|
|
|
|
type DraftDetail struct {
|
|
Draft
|
|
Account Account `json:"account"`
|
|
Versions []Draft `json:"versions"`
|
|
Confirmations []ConfirmationSnapshot `json:"confirmations"`
|
|
Tasks []Task `json:"tasks"`
|
|
}
|
|
|
|
type ReadinessError struct {
|
|
Reason string
|
|
Unavailable bool
|
|
}
|
|
|
|
func (e *ReadinessError) Error() string { return e.Reason }
|
|
|
|
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"`
|
|
}
|
|
|
|
type AuditFilter struct {
|
|
AccountID, TaskID, AttemptID, BrowserEnvAlias, NetworkExitID, EventType string
|
|
From, To *time.Time
|
|
Page, PageSize int
|
|
}
|
|
|
|
type AuditPage struct {
|
|
Data []AuditEvent `json:"data"`
|
|
Total int `json:"total"`
|
|
Page int `json:"page"`
|
|
PageSize int `json:"page_size"`
|
|
}
|
|
|
|
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) SetTaskNotifier(notify taskstate.Notifier) { s.notify = notify }
|
|
|
|
func (s *Store) notifyTransitions(transitions []taskstate.Transition) {
|
|
if s.notify == nil {
|
|
return
|
|
}
|
|
for _, transition := range transitions {
|
|
s.notify(transition)
|
|
}
|
|
}
|
|
|
|
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, credentials CredentialBridge) (err error) {
|
|
if account.Tags == nil {
|
|
account.Tags = []string{}
|
|
}
|
|
if !validAccount(account) || credentials == nil {
|
|
return ErrInvalid
|
|
}
|
|
// 空凭据（扫码登录场景）不写 keyring：凭据留待后续登录/同步链路补齐
|
|
stored := false
|
|
if account.Cookies != "" {
|
|
if err := credentials.Store(ctx, account.CredentialReference, account.CredentialKey, account.Cookies); err != nil {
|
|
storeErr := errors.New("store account credential")
|
|
if cleanupErr := credentials.Delete(context.WithoutCancel(ctx), account.CredentialReference, account.CredentialKey); cleanupErr != nil {
|
|
storeErr = errors.Join(storeErr, errors.New("delete incomplete account credential"))
|
|
}
|
|
return storeErr
|
|
}
|
|
stored = true
|
|
}
|
|
defer func() {
|
|
if stored && err != nil && !errors.Is(err, ErrAccountCreationUnknown) {
|
|
if cleanupErr := credentials.Delete(context.WithoutCancel(ctx), account.CredentialReference, account.CredentialKey); cleanupErr != nil {
|
|
err = errors.Join(err, errors.New("delete orphaned account credential"))
|
|
}
|
|
}
|
|
}()
|
|
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, name, platform, platform_account_key, tags, authorization_kind, authorization_status, status)
|
|
VALUES ($1, $2, $3, $4, $5, $6, 'owned', 'authorized', 'paused')`, account.ID, account.CredentialReference.ID,
|
|
account.Name, account.Platform, account.PlatformAccountKey, account.Tags); err != nil {
|
|
return publicDatabaseError(err)
|
|
}
|
|
if err := appendAudit(ctx, tx, "account_created", "account_created", account.ID, "", 0, "", "", map[string]string{"platform": account.Platform}); err != nil {
|
|
return err
|
|
}
|
|
var commitErr error
|
|
if s.accountCommit != nil {
|
|
commitErr = s.accountCommit(tx)
|
|
} else {
|
|
commitErr = tx.Commit()
|
|
}
|
|
if commitErr == nil {
|
|
return nil
|
|
}
|
|
if !commitKnownRolledBack(commitErr) {
|
|
return ErrAccountCreationUnknown
|
|
}
|
|
return errors.New("commit account transaction")
|
|
}
|
|
|
|
func commitKnownRolledBack(err error) bool {
|
|
if errors.Is(err, pgx.ErrTxCommitRollback) {
|
|
return true
|
|
}
|
|
var postgresError *pgconn.PgError
|
|
return errors.As(err, &postgresError)
|
|
}
|
|
|
|
func (s *Store) ListAccounts(ctx context.Context) ([]Account, error) {
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT account.id, account.name, account.platform, account.platform_account_key, account.tags,
|
|
account.authorization_status, account.status, account.version
|
|
FROM social_account account
|
|
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.name, account.platform, account.platform_account_key, account.tags,
|
|
account.authorization_status, account.status, account.version
|
|
FROM social_account account
|
|
WHERE account.id = $1`, id))
|
|
}
|
|
|
|
type accountScanner interface{ Scan(...any) error }
|
|
|
|
func scanAccount(row accountScanner) (Account, error) {
|
|
var account Account
|
|
var tags pgtype.FlatArray[string]
|
|
if err := row.Scan(&account.ID, &account.Name, &account.Platform, &account.PlatformAccountKey, pgtype.NewMap().SQLScanner(&tags),
|
|
&account.AuthorizationStatus, &account.RuntimeStatus, &account.Version); err != nil {
|
|
return Account{}, rowError(err)
|
|
}
|
|
account.Tags = []string(tags)
|
|
return account, nil
|
|
}
|
|
|
|
func validAccount(account Account) bool {
|
|
if !idPattern.MatchString(account.ID) || strings.TrimSpace(account.Name) != account.Name || account.Name == "" ||
|
|
!utf8.ValidString(account.Name) || utf8.RuneCountInString(account.Name) > 128 ||
|
|
!platformKeyPattern.MatchString(account.PlatformAccountKey) || len(account.Tags) > 20 ||
|
|
len(account.Cookies) > 8192 || !refPattern.MatchString(account.CredentialReference.ID) ||
|
|
!credentialKeyPattern.MatchString(account.CredentialKey) ||
|
|
(account.CredentialReference.Provider != "os_keyring" && account.CredentialReference.Provider != "secret_manager") {
|
|
return false
|
|
}
|
|
switch account.Platform {
|
|
case "douyin", "xiaohongshu", "wechat-official", "kuaishou":
|
|
default:
|
|
return false
|
|
}
|
|
for _, tag := range account.Tags {
|
|
if tag == "" || strings.TrimSpace(tag) != tag || !utf8.ValidString(tag) || utf8.RuneCountInString(tag) > 32 {
|
|
return false
|
|
}
|
|
}
|
|
// 空凭据合法（扫码登录场景）；非空时才校验 Cookie Header 格式
|
|
if account.Cookies != "" {
|
|
if _, err := http.ParseCookie(account.Cookies); err != nil {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
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) CreateDraftVersion(ctx context.Context, accountID, content string) (Draft, error) {
|
|
if !idPattern.MatchString(accountID) || strings.TrimSpace(content) == "" {
|
|
return Draft{}, ErrInvalid
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return Draft{}, errors.New("begin draft transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
if err := tx.QueryRowContext(ctx, `SELECT id FROM social_account WHERE id = $1 FOR UPDATE`, accountID).Scan(&accountID); err != nil {
|
|
return Draft{}, rowError(err)
|
|
}
|
|
draft := Draft{ID: "draft-" + newID()[:24], AccountID: accountID, Content: content}
|
|
if err := tx.QueryRowContext(ctx, `
|
|
INSERT INTO content_draft (id, account_id, version, content)
|
|
SELECT $1, $2, COALESCE(max(version), 0) + 1, $3 FROM content_draft WHERE account_id = $2
|
|
RETURNING version, created_at`, draft.ID, accountID, content).Scan(&draft.Version, &draft.CreatedAt); err != nil {
|
|
return Draft{}, publicDatabaseError(err)
|
|
}
|
|
if err := appendAudit(ctx, tx, "draft_created", "draft_created", accountID, "", 0, "", "", map[string]any{
|
|
"draft_id": draft.ID, "draft_version": draft.Version,
|
|
}); err != nil {
|
|
return Draft{}, err
|
|
}
|
|
if err := commit(tx); err != nil {
|
|
return Draft{}, err
|
|
}
|
|
return draft, nil
|
|
}
|
|
|
|
func (s *Store) ListDrafts(ctx context.Context, accountID string) ([]Draft, error) {
|
|
if accountID != "" && !idPattern.MatchString(accountID) {
|
|
return nil, ErrInvalid
|
|
}
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT id, account_id, version, content, created_at FROM content_draft
|
|
WHERE $1 = '' OR account_id = $1 ORDER BY account_id, version DESC, created_at DESC`, accountID)
|
|
if err != nil {
|
|
return nil, errors.New("read drafts")
|
|
}
|
|
defer rows.Close()
|
|
drafts := []Draft{}
|
|
for rows.Next() {
|
|
draft, err := scanDraft(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
drafts = append(drafts, draft)
|
|
}
|
|
return drafts, rows.Err()
|
|
}
|
|
|
|
func (s *Store) GetDraft(ctx context.Context, id string) (Draft, error) {
|
|
if !refPattern.MatchString(id) {
|
|
return Draft{}, ErrInvalid
|
|
}
|
|
return scanDraft(s.db.QueryRowContext(ctx, `
|
|
SELECT id, account_id, version, content, created_at FROM content_draft WHERE id = $1`, id))
|
|
}
|
|
|
|
type draftScanner interface{ Scan(...any) error }
|
|
|
|
func scanDraft(row draftScanner) (Draft, error) {
|
|
var draft Draft
|
|
if err := row.Scan(&draft.ID, &draft.AccountID, &draft.Version, &draft.Content, &draft.CreatedAt); err != nil {
|
|
return Draft{}, rowError(err)
|
|
}
|
|
return draft, nil
|
|
}
|
|
|
|
func (s *Store) GetDraftDetail(ctx context.Context, id string) (DraftDetail, error) {
|
|
draft, err := s.GetDraft(ctx, id)
|
|
if err != nil {
|
|
return DraftDetail{}, err
|
|
}
|
|
account, err := s.GetAccount(ctx, draft.AccountID)
|
|
if err != nil {
|
|
return DraftDetail{}, err
|
|
}
|
|
versions, err := s.ListDrafts(ctx, draft.AccountID)
|
|
if err != nil {
|
|
return DraftDetail{}, err
|
|
}
|
|
confirmations, err := s.ListConfirmations(ctx, draft.ID)
|
|
if err != nil {
|
|
return DraftDetail{}, err
|
|
}
|
|
tasks, err := s.ListTasks(ctx, draft.AccountID, draft.ID)
|
|
if err != nil {
|
|
return DraftDetail{}, err
|
|
}
|
|
return DraftDetail{Draft: draft, Account: account, Versions: versions, Confirmations: confirmations, Tasks: tasks}, nil
|
|
}
|
|
|
|
func (s *Store) ConfirmDraft(ctx context.Context, draftID string, accountVersion, draftVersion int64) (Confirmation, bool, error) {
|
|
if !refPattern.MatchString(draftID) || accountVersion < 1 || draftVersion < 1 {
|
|
return Confirmation{}, false, ErrInvalid
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return Confirmation{}, false, errors.New("begin confirmation transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
var accountID string
|
|
var currentAccountVersion, currentDraftVersion, latestDraftVersion int64
|
|
if err := tx.QueryRowContext(ctx, `
|
|
SELECT draft.account_id, account.version, draft.version,
|
|
(SELECT max(version) FROM content_draft WHERE account_id = draft.account_id)
|
|
FROM content_draft draft JOIN social_account account ON account.id = draft.account_id
|
|
WHERE draft.id = $1 FOR UPDATE OF draft, account`, draftID).
|
|
Scan(&accountID, ¤tAccountVersion, ¤tDraftVersion, &latestDraftVersion); err != nil {
|
|
return Confirmation{}, false, rowError(err)
|
|
}
|
|
if accountVersion != currentAccountVersion {
|
|
return Confirmation{}, false, &ReadinessError{Reason: "account_version_changed"}
|
|
}
|
|
if draftVersion != currentDraftVersion || draftVersion != latestDraftVersion {
|
|
return Confirmation{}, false, &ReadinessError{Reason: "draft_version_changed"}
|
|
}
|
|
var existing Confirmation
|
|
err = tx.QueryRowContext(ctx, `
|
|
SELECT id, account_id, account_version, draft_id, draft_version, version, confirmed_at
|
|
FROM confirmation WHERE account_id = $1 AND account_version = $2 AND draft_id = $3 AND draft_version = $4
|
|
ORDER BY version DESC LIMIT 1`, accountID, accountVersion, draftID, draftVersion).
|
|
Scan(&existing.ID, &existing.AccountID, &existing.AccountVersion, &existing.DraftID, &existing.DraftVersion, &existing.Version, &existing.ConfirmedAt)
|
|
if err == nil {
|
|
if err := commit(tx); err != nil {
|
|
return Confirmation{}, false, err
|
|
}
|
|
return existing, false, nil
|
|
}
|
|
if !errors.Is(err, sql.ErrNoRows) {
|
|
return Confirmation{}, false, publicDatabaseError(err)
|
|
}
|
|
confirmation := Confirmation{
|
|
ID: "confirmation-" + newID()[:20], AccountID: accountID, AccountVersion: accountVersion,
|
|
DraftID: draftID, DraftVersion: draftVersion,
|
|
}
|
|
if err := tx.QueryRowContext(ctx, `
|
|
INSERT INTO confirmation (id, account_id, account_version, draft_id, draft_version, version)
|
|
SELECT $1, $2, $3, $4, $5, COALESCE(max(version), 0) + 1 FROM confirmation WHERE draft_id = $4
|
|
RETURNING version, confirmed_at`, confirmation.ID, accountID, accountVersion, draftID, draftVersion).
|
|
Scan(&confirmation.Version, &confirmation.ConfirmedAt); err != nil {
|
|
return Confirmation{}, false, publicDatabaseError(err)
|
|
}
|
|
if err := appendAudit(ctx, tx, "draft_confirmed", "draft_confirmed", accountID, confirmation.ID, confirmation.Version, "", "", map[string]any{
|
|
"account_version": accountVersion, "draft_id": draftID, "draft_version": draftVersion,
|
|
}); err != nil {
|
|
return Confirmation{}, false, err
|
|
}
|
|
if err := commit(tx); err != nil {
|
|
return Confirmation{}, false, err
|
|
}
|
|
return confirmation, true, nil
|
|
}
|
|
|
|
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", "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) ListConfirmations(ctx context.Context, draftID string) ([]ConfirmationSnapshot, error) {
|
|
if draftID != "" && !refPattern.MatchString(draftID) {
|
|
return nil, ErrInvalid
|
|
}
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT confirmation.id, confirmation.account_id, confirmation.account_version, confirmation.draft_id,
|
|
confirmation.draft_version, confirmation.version, confirmation.confirmed_at,
|
|
audit.browser_env_alias, audit.network_exit_id, audit.runtime_instance_id, audit.binding_version
|
|
FROM confirmation
|
|
LEFT JOIN LATERAL (
|
|
SELECT browser_env_alias, network_exit_id, runtime_instance_id, binding_version
|
|
FROM audit_event WHERE confirmation_id = confirmation.id AND event_type = 'draft_confirmed'
|
|
ORDER BY id DESC LIMIT 1
|
|
) audit ON true
|
|
WHERE $1 = '' OR confirmation.draft_id = $1
|
|
ORDER BY confirmation.confirmed_at DESC, confirmation.version DESC`, draftID)
|
|
if err != nil {
|
|
return nil, errors.New("read confirmations")
|
|
}
|
|
defer rows.Close()
|
|
confirmations := []ConfirmationSnapshot{}
|
|
for rows.Next() {
|
|
confirmation, err := scanConfirmation(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
confirmations = append(confirmations, confirmation)
|
|
}
|
|
return confirmations, rows.Err()
|
|
}
|
|
|
|
func (s *Store) GetConfirmation(ctx context.Context, id string) (ConfirmationSnapshot, error) {
|
|
if !refPattern.MatchString(id) {
|
|
return ConfirmationSnapshot{}, ErrInvalid
|
|
}
|
|
return scanConfirmation(s.db.QueryRowContext(ctx, `
|
|
SELECT confirmation.id, confirmation.account_id, confirmation.account_version, confirmation.draft_id,
|
|
confirmation.draft_version, confirmation.version, confirmation.confirmed_at,
|
|
audit.browser_env_alias, audit.network_exit_id, audit.runtime_instance_id, audit.binding_version
|
|
FROM confirmation
|
|
LEFT JOIN LATERAL (
|
|
SELECT browser_env_alias, network_exit_id, runtime_instance_id, binding_version
|
|
FROM audit_event WHERE confirmation_id = confirmation.id AND event_type = 'draft_confirmed'
|
|
ORDER BY id DESC LIMIT 1
|
|
) audit ON true WHERE confirmation.id = $1`, id))
|
|
}
|
|
|
|
type confirmationScanner interface{ Scan(...any) error }
|
|
|
|
func scanConfirmation(row confirmationScanner) (ConfirmationSnapshot, error) {
|
|
var confirmation ConfirmationSnapshot
|
|
var browser, network, runtime sql.NullString
|
|
var bindingVersion sql.NullInt64
|
|
if err := row.Scan(&confirmation.ID, &confirmation.AccountID, &confirmation.AccountVersion, &confirmation.DraftID,
|
|
&confirmation.DraftVersion, &confirmation.Version, &confirmation.ConfirmedAt,
|
|
&browser, &network, &runtime, &bindingVersion); err != nil {
|
|
return ConfirmationSnapshot{}, rowError(err)
|
|
}
|
|
confirmation.BrowserEnvAlias, confirmation.NetworkExitID = browser.String, network.String
|
|
confirmation.RuntimeInstanceID, confirmation.BindingVersion = runtime.String, bindingVersion.Int64
|
|
return confirmation, nil
|
|
}
|
|
|
|
func (s *Store) ListTasks(ctx context.Context, accountID, draftID string) ([]Task, error) {
|
|
return s.ListTasksFiltered(ctx, accountID, draftID, "")
|
|
}
|
|
|
|
func (s *Store) ListTasksFiltered(ctx context.Context, accountID, draftID, state string) ([]Task, error) {
|
|
if (accountID != "" && !idPattern.MatchString(accountID)) || (draftID != "" && !refPattern.MatchString(draftID)) ||
|
|
(state != "" && state != "queued" && state != "executing" && state != "succeeded" && state != "failed" &&
|
|
state != "needs_confirmation" && state != "policy_hold" && state != "cancelled") {
|
|
return nil, ErrInvalid
|
|
}
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT id, idempotency_key, account_id, account_version, draft_id, draft_version,
|
|
confirmation_id, confirmation_version, state, hold_reason, verification_result, verified_at, verified_by,
|
|
created_at, updated_at
|
|
FROM operation_task WHERE ($1 = '' OR account_id = $1) AND ($2 = '' OR draft_id = $2)
|
|
AND ($3 = '' OR state = $3)
|
|
ORDER BY created_at DESC, id`, accountID, draftID, state)
|
|
if err != nil {
|
|
return nil, errors.New("read tasks")
|
|
}
|
|
defer rows.Close()
|
|
tasks := []Task{}
|
|
for rows.Next() {
|
|
task, err := scanTask(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
tasks = append(tasks, task)
|
|
}
|
|
return tasks, rows.Err()
|
|
}
|
|
|
|
func (s *Store) GetTask(ctx context.Context, id string) (Task, error) {
|
|
if !refPattern.MatchString(id) {
|
|
return Task{}, ErrInvalid
|
|
}
|
|
return scanTask(s.db.QueryRowContext(ctx, `
|
|
SELECT id, idempotency_key, account_id, account_version, draft_id, draft_version,
|
|
confirmation_id, confirmation_version, state, hold_reason, verification_result, verified_at, verified_by,
|
|
created_at, updated_at
|
|
FROM operation_task WHERE id = $1`, id))
|
|
}
|
|
|
|
type taskScanner interface{ Scan(...any) error }
|
|
|
|
func scanTask(row taskScanner) (Task, error) {
|
|
var task Task
|
|
var confirmationID, holdReason, verificationResult, verifiedBy sql.NullString
|
|
var confirmationVersion sql.NullInt64
|
|
var verifiedAt sql.NullTime
|
|
if err := row.Scan(&task.ID, &task.IdempotencyKey, &task.AccountID, &task.AccountVersion, &task.DraftID,
|
|
&task.DraftVersion, &confirmationID, &confirmationVersion, &task.State, &holdReason, &verificationResult,
|
|
&verifiedAt, &verifiedBy, &task.CreatedAt, &task.UpdatedAt); err != nil {
|
|
return Task{}, rowError(err)
|
|
}
|
|
task.ConfirmationID, task.ConfirmationVersion = confirmationID.String, confirmationVersion.Int64
|
|
task.HoldReason, task.VerificationResult, task.VerifiedBy = holdReason.String, verificationResult.String, verifiedBy.String
|
|
if verifiedAt.Valid {
|
|
task.VerifiedAt = &verifiedAt.Time
|
|
}
|
|
return task, nil
|
|
}
|
|
|
|
func (s *Store) GetTaskDetail(ctx context.Context, id string) (TaskDetail, error) {
|
|
task, err := s.GetTask(ctx, id)
|
|
if err != nil {
|
|
return TaskDetail{}, err
|
|
}
|
|
var confirmation ConfirmationSnapshot
|
|
if task.ConfirmationID != "" {
|
|
confirmation, err = s.GetConfirmation(ctx, task.ConfirmationID)
|
|
if err != nil {
|
|
return TaskDetail{}, err
|
|
}
|
|
}
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT id, started_at, finished_at, outcome, result
|
|
FROM execution_attempt WHERE task_id = $1 ORDER BY started_at, id`, id)
|
|
if err != nil {
|
|
return TaskDetail{}, errors.New("read task attempts")
|
|
}
|
|
defer rows.Close()
|
|
attempts := []TaskAttempt{}
|
|
for rows.Next() {
|
|
var attempt TaskAttempt
|
|
var finishedAt sql.NullTime
|
|
var outcome sql.NullString
|
|
var result json.RawMessage
|
|
if err := rows.Scan(&attempt.ID, &attempt.StartedAt, &finishedAt, &outcome, &result); err != nil {
|
|
return TaskDetail{}, errors.New("decode task attempt")
|
|
}
|
|
if finishedAt.Valid {
|
|
attempt.FinishedAt = &finishedAt.Time
|
|
}
|
|
attempt.Outcome, attempt.Evidence = outcome.String, safeEvidence(result)
|
|
attempts = append(attempts, attempt)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return TaskDetail{}, errors.New("read task attempts")
|
|
}
|
|
detail := TaskDetail{Task: task, Confirmation: confirmation, Attempts: attempts, AllowedAction: taskAllowedAction(task)}
|
|
if task.VerificationResult == "not_executed" {
|
|
reason, _, err := taskReadinessReason(ctx, s.db, task.ID)
|
|
if err != nil {
|
|
return TaskDetail{}, err
|
|
}
|
|
detail.ReadinessReason = reason
|
|
if reason != "" {
|
|
detail.AllowedAction = ""
|
|
switch reason {
|
|
case "account_version_changed", "draft_version_changed", "confirmation_version_changed", "binding_version_changed":
|
|
detail.AllowedAction = "reconfirm"
|
|
case "confirmation_missing":
|
|
if task.ConfirmationID != "" {
|
|
detail.AllowedAction = "reconfirm"
|
|
}
|
|
}
|
|
}
|
|
}
|
|
var browser, network, runtime sql.NullString
|
|
var bindingVersion sql.NullInt64
|
|
err = s.db.QueryRowContext(ctx, `
|
|
SELECT browser_env_alias, network_exit_id, runtime_instance_id, binding_version
|
|
FROM audit_event WHERE task_id = $1 AND event_type = 'task_claimed' ORDER BY id DESC LIMIT 1`, id).
|
|
Scan(&browser, &network, &runtime, &bindingVersion)
|
|
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return TaskDetail{}, errors.New("read task correlation")
|
|
}
|
|
detail.BrowserEnvAlias, detail.NetworkExitID = browser.String, network.String
|
|
detail.RuntimeInstanceID, detail.BindingVersion = runtime.String, bindingVersion.Int64
|
|
return detail, nil
|
|
}
|
|
|
|
func (s *Store) GetTaskAttemptDetail(ctx context.Context, id string) (TaskAttemptDetail, error) {
|
|
if !refPattern.MatchString(id) {
|
|
return TaskAttemptDetail{}, ErrInvalid
|
|
}
|
|
var detail TaskAttemptDetail
|
|
var finishedAt sql.NullTime
|
|
var outcome, browser, network, runtime sql.NullString
|
|
var bindingVersion sql.NullInt64
|
|
var result json.RawMessage
|
|
err := s.db.QueryRowContext(ctx, `
|
|
SELECT attempt.id, attempt.task_id, attempt.started_at, attempt.finished_at, attempt.outcome, attempt.result,
|
|
audit.browser_env_alias, audit.network_exit_id, audit.runtime_instance_id, audit.binding_version
|
|
FROM execution_attempt attempt
|
|
LEFT JOIN LATERAL (
|
|
SELECT browser_env_alias, network_exit_id, runtime_instance_id, binding_version
|
|
FROM audit_event WHERE attempt_id = attempt.id AND event_type = 'task_claimed' ORDER BY id DESC LIMIT 1
|
|
) audit ON true
|
|
WHERE attempt.id = $1`, id).Scan(&detail.ID, &detail.TaskID, &detail.StartedAt, &finishedAt, &outcome, &result,
|
|
&browser, &network, &runtime, &bindingVersion)
|
|
if err != nil {
|
|
return TaskAttemptDetail{}, rowError(err)
|
|
}
|
|
if finishedAt.Valid {
|
|
detail.FinishedAt = &finishedAt.Time
|
|
}
|
|
detail.Outcome, detail.Evidence = outcome.String, safeEvidence(result)
|
|
detail.BrowserEnvAlias, detail.NetworkExitID = browser.String, network.String
|
|
detail.RuntimeInstanceID, detail.BindingVersion = runtime.String, bindingVersion.Int64
|
|
return detail, nil
|
|
}
|
|
|
|
func safeEvidence(raw json.RawMessage) map[string]string {
|
|
var values map[string]any
|
|
if json.Unmarshal(raw, &values) != nil {
|
|
return map[string]string{}
|
|
}
|
|
evidence := map[string]string{}
|
|
if outcome, ok := values["mock_outcome"].(string); ok {
|
|
evidence["mock_outcome"] = outcome
|
|
}
|
|
return evidence
|
|
}
|
|
|
|
func taskAllowedAction(task Task) string {
|
|
if task.State != "policy_hold" && task.State != "needs_confirmation" {
|
|
return ""
|
|
}
|
|
if task.VerificationResult == "not_executed" {
|
|
return "resume"
|
|
}
|
|
if task.VerificationResult == "succeeded" || task.VerificationResult == "failed" {
|
|
return "finish"
|
|
}
|
|
switch task.HoldReason {
|
|
case "confirmation_missing":
|
|
if task.ConfirmationID == "" {
|
|
return ""
|
|
}
|
|
return "reconfirm"
|
|
case "account_version_changed", "draft_version_changed", "confirmation_version_changed", "binding_version_changed":
|
|
return "reconfirm"
|
|
case "account_paused", "account_revoked", "binding_missing", "environment_missing", "exit_missing", "exit_unhealthy",
|
|
"runtime_stop_pending", "runtime_missing", "runtime_lease_expired", "execution_lease_expired", "task_result_uncertain", "task_policy_hold":
|
|
return "verify"
|
|
default:
|
|
return ""
|
|
}
|
|
}
|
|
|
|
func (s *Store) EnqueueConfirmation(ctx context.Context, confirmationID string) (Task, bool, error) {
|
|
if !refPattern.MatchString(confirmationID) {
|
|
return Task{}, false, ErrInvalid
|
|
}
|
|
sum := sha256.Sum256([]byte(confirmationID))
|
|
idempotencyKey := "enqueue-" + hex.EncodeToString(sum[:])
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return Task{}, false, errors.New("begin confirmed enqueue transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
if existing, err := scanTask(tx.QueryRowContext(ctx, `
|
|
SELECT id, idempotency_key, account_id, account_version, draft_id, draft_version,
|
|
confirmation_id, confirmation_version, state, hold_reason, verification_result, verified_at, verified_by,
|
|
created_at, updated_at
|
|
FROM operation_task WHERE idempotency_key = $1`, idempotencyKey)); err == nil {
|
|
if err := commit(tx); err != nil {
|
|
return Task{}, false, err
|
|
}
|
|
return existing, false, nil
|
|
} else if !errors.Is(err, ErrNotFound) {
|
|
return Task{}, false, err
|
|
}
|
|
|
|
var task Task
|
|
var currentAccountVersion, currentDraftVersion, latestDraftVersion int64
|
|
var authorizationStatus, accountStatus string
|
|
if err := tx.QueryRowContext(ctx, `
|
|
SELECT confirmation.account_id, confirmation.account_version, confirmation.draft_id,
|
|
confirmation.draft_version, confirmation.version, account.version, account.authorization_status,
|
|
account.status, draft.version, (SELECT max(version) FROM content_draft WHERE account_id = confirmation.account_id)
|
|
FROM confirmation
|
|
JOIN social_account account ON account.id = confirmation.account_id
|
|
JOIN content_draft draft ON draft.id = confirmation.draft_id
|
|
WHERE confirmation.id = $1 FOR UPDATE OF confirmation, account, draft`, confirmationID).
|
|
Scan(&task.AccountID, &task.AccountVersion, &task.DraftID, &task.DraftVersion, &task.ConfirmationVersion,
|
|
¤tAccountVersion, &authorizationStatus, &accountStatus, ¤tDraftVersion, &latestDraftVersion); err != nil {
|
|
return Task{}, false, rowError(err)
|
|
}
|
|
task.ConfirmationID, task.IdempotencyKey, task.ID = confirmationID, idempotencyKey, "task-"+newID()[:24]
|
|
if task.AccountVersion != currentAccountVersion {
|
|
return Task{}, false, &ReadinessError{Reason: "account_version_changed"}
|
|
}
|
|
if task.DraftVersion != currentDraftVersion || task.DraftVersion != latestDraftVersion {
|
|
return Task{}, false, &ReadinessError{Reason: "draft_version_changed"}
|
|
}
|
|
if authorizationStatus != "authorized" {
|
|
return Task{}, false, &ReadinessError{Reason: "account_revoked"}
|
|
}
|
|
if accountStatus != "active" {
|
|
return Task{}, false, &ReadinessError{Reason: "account_paused"}
|
|
}
|
|
var bindingID string
|
|
var networkExitID sql.NullString
|
|
var bindingVersion int64
|
|
var cleanupPending bool
|
|
if err := tx.QueryRowContext(ctx, `
|
|
SELECT id, network_exit_id, version, runtime_cleanup_pending
|
|
FROM environment_binding WHERE account_id = $1 FOR SHARE`, task.AccountID).
|
|
Scan(&bindingID, &networkExitID, &bindingVersion, &cleanupPending); err != nil {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return Task{}, false, &ReadinessError{Reason: "binding_missing", Unavailable: true}
|
|
}
|
|
return Task{}, false, publicDatabaseError(err)
|
|
}
|
|
if networkExitID.Valid {
|
|
var exitStatus string
|
|
if err := tx.QueryRowContext(ctx, `SELECT health_status FROM network_exit WHERE id = $1 FOR SHARE`, networkExitID.String).Scan(&exitStatus); err != nil {
|
|
return Task{}, false, publicDatabaseError(err)
|
|
}
|
|
if exitStatus != "healthy" {
|
|
return Task{}, false, &ReadinessError{Reason: "network_exit_unhealthy", Unavailable: true}
|
|
}
|
|
}
|
|
if cleanupPending {
|
|
return Task{}, false, &ReadinessError{Reason: "runtime_stop_pending", Unavailable: true}
|
|
}
|
|
var runtimeID string
|
|
var runtimeBindingVersion int64
|
|
var leaseActive bool
|
|
if err := tx.QueryRowContext(ctx, `
|
|
SELECT id, binding_version, lease_until > now() FROM runtime_instance
|
|
WHERE binding_id = $1 AND released_at IS NULL ORDER BY acquired_at DESC LIMIT 1 FOR SHARE`, bindingID).
|
|
Scan(&runtimeID, &runtimeBindingVersion, &leaseActive); err != nil {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return Task{}, false, &ReadinessError{Reason: "runtime_missing", Unavailable: true}
|
|
}
|
|
return Task{}, false, publicDatabaseError(err)
|
|
}
|
|
if !leaseActive {
|
|
return Task{}, false, &ReadinessError{Reason: "runtime_lease_expired", Unavailable: true}
|
|
}
|
|
if runtimeBindingVersion != bindingVersion {
|
|
return Task{}, false, &ReadinessError{Reason: "binding_version_changed"}
|
|
}
|
|
created, inserted, err := enqueueTask(ctx, tx, task)
|
|
if err != nil {
|
|
return Task{}, false, err
|
|
}
|
|
if err := commit(tx); err != nil {
|
|
return Task{}, false, err
|
|
}
|
|
return created, inserted, nil
|
|
}
|
|
|
|
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()
|
|
created, inserted, err := enqueueTask(ctx, tx, task)
|
|
if err != nil {
|
|
return Task{}, false, err
|
|
}
|
|
if err := commit(tx); err != nil {
|
|
return Task{}, false, err
|
|
}
|
|
return created, inserted, nil
|
|
}
|
|
|
|
func enqueueTask(ctx context.Context, tx *sql.Tx, task Task) (Task, bool, error) {
|
|
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, created_at`, task.ID, task.IdempotencyKey, task.AccountID, task.AccountVersion,
|
|
task.DraftID, task.DraftVersion, task.ConfirmationID, task.ConfirmationVersion).Scan(&insertedID, &task.CreatedAt)
|
|
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return Task{}, false, publicDatabaseError(err)
|
|
}
|
|
if insertedID != "" {
|
|
task.State = "queued"
|
|
task.UpdatedAt = task.CreatedAt
|
|
if err := appendAudit(ctx, tx, "task_queued", "task_queued", task.AccountID, task.ConfirmationID, task.ConfirmationVersion, "", task.ID, nil); err != nil {
|
|
return Task{}, false, err
|
|
}
|
|
return task, true, nil
|
|
}
|
|
existing, err := scanTask(tx.QueryRowContext(ctx, `
|
|
SELECT id, idempotency_key, account_id, account_version, draft_id, draft_version,
|
|
confirmation_id, confirmation_version, state, hold_reason, verification_result, verified_at, verified_by,
|
|
created_at, updated_at
|
|
FROM operation_task WHERE idempotency_key = $1`, task.IdempotencyKey))
|
|
if err != nil {
|
|
return Task{}, false, err
|
|
}
|
|
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
|
|
}
|
|
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")
|
|
}
|
|
}
|
|
reason := "account_paused"
|
|
if revoke {
|
|
reason = "account_revoked"
|
|
}
|
|
held, err := holdQueuedTasks(ctx, tx, accountID, reason)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
interrupted, err := interruptExecutingTasks(ctx, tx, accountID, reason)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !unchanged {
|
|
eventType := "account_paused"
|
|
if revoke {
|
|
eventType = "account_revoked"
|
|
}
|
|
if err := appendAudit(ctx, tx, eventType, reason, accountID, "", 0, "", "", map[string]any{
|
|
"account_version": version, "tasks_held": len(held), "attempts_interrupted": len(interrupted),
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return s.commitTransitions(tx, append(held, interrupted...))
|
|
}
|
|
|
|
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
|
|
LEFT JOIN network_exit network ON network.id = binding.network_exit_id
|
|
WHERE binding.account_id = $1 AND (binding.network_exit_id IS NULL OR network.health_status = 'healthy')
|
|
AND NOT binding.runtime_cleanup_pending
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM runtime_instance runtime
|
|
WHERE runtime.binding_id = binding.id AND runtime.released_at IS NULL
|
|
)
|
|
)`, 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", "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, reason string) ([]taskstate.Transition, error) {
|
|
rows, err := tx.QueryContext(ctx, `
|
|
UPDATE operation_task SET state = 'policy_hold', hold_reason = $2,
|
|
verification_result = NULL, verified_at = NULL, verified_by = NULL, updated_at = now()
|
|
WHERE account_id = $1 AND state = 'queued'
|
|
RETURNING state, hold_reason, account_id, id`, accountID, reason)
|
|
if err != nil {
|
|
return nil, errors.New("hold queued account tasks")
|
|
}
|
|
return scanTransitions(rows, "read held account tasks")
|
|
}
|
|
|
|
func interruptExecutingTasks(ctx context.Context, tx *sql.Tx, accountID, reason string) ([]taskstate.Transition, error) {
|
|
rows, err := tx.QueryContext(ctx, `
|
|
UPDATE operation_task SET state = 'needs_confirmation', hold_reason = $2,
|
|
verification_result = NULL, verified_at = NULL, verified_by = NULL,
|
|
lease_owner = NULL, lease_until = NULL, updated_at = now()
|
|
WHERE account_id = $1 AND state = 'executing'
|
|
RETURNING id, current_attempt_id, confirmation_id, confirmation_version`, accountID, reason)
|
|
if err != nil {
|
|
return nil, errors.New("interrupt executing account tasks")
|
|
}
|
|
type interruptedTask struct {
|
|
taskID, attemptID, confirmationID string
|
|
confirmationVersion int64
|
|
}
|
|
var tasks []interruptedTask
|
|
for rows.Next() {
|
|
var task interruptedTask
|
|
var attemptID, confirmationID sql.NullString
|
|
var confirmationVersion sql.NullInt64
|
|
if err := rows.Scan(&task.taskID, &attemptID, &confirmationID, &confirmationVersion); err != nil {
|
|
rows.Close()
|
|
return nil, errors.New("read interrupted account task")
|
|
}
|
|
task.attemptID, task.confirmationID, task.confirmationVersion = attemptID.String, confirmationID.String, confirmationVersion.Int64
|
|
tasks = append(tasks, task)
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, errors.New("close interrupted account task rows")
|
|
}
|
|
transitions := make([]taskstate.Transition, 0, len(tasks))
|
|
for _, task := range tasks {
|
|
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 nil, errors.New("mark interrupted attempt uncertain")
|
|
}
|
|
}
|
|
if err := appendAudit(ctx, tx, "task_interrupted", reason, accountID, task.confirmationID,
|
|
task.confirmationVersion, task.attemptID, task.taskID, nil); err != nil {
|
|
return nil, err
|
|
}
|
|
transitions = append(transitions, taskstate.Transition{
|
|
State: "needs_confirmation", ReasonCode: reason, AccountID: accountID, TaskID: task.taskID,
|
|
})
|
|
}
|
|
return transitions, nil
|
|
}
|
|
|
|
func (s *Store) VerifyTask(ctx context.Context, taskID, result string) error {
|
|
if !refPattern.MatchString(taskID) || (result != "not_executed" && result != "succeeded" && result != "failed") {
|
|
return ErrInvalid
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return errors.New("begin task verification transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
task, err := scanTask(tx.QueryRowContext(ctx, `
|
|
SELECT id, idempotency_key, account_id, account_version, draft_id, draft_version,
|
|
confirmation_id, confirmation_version, state, hold_reason, verification_result, verified_at, verified_by,
|
|
created_at, updated_at
|
|
FROM operation_task WHERE id = $1 FOR UPDATE`, taskID))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if task.State != "policy_hold" && task.State != "needs_confirmation" {
|
|
return ErrConflict
|
|
}
|
|
if task.VerificationResult != "" {
|
|
return ErrConflict
|
|
}
|
|
switch task.HoldReason {
|
|
case "account_paused", "account_revoked", "binding_missing", "environment_missing", "exit_missing", "exit_unhealthy",
|
|
"runtime_stop_pending", "runtime_missing", "runtime_lease_expired", "execution_lease_expired", "task_result_uncertain", "task_policy_hold":
|
|
default:
|
|
return ErrConflict
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE operation_task SET verification_result = $2, verified_at = now(), verified_by = 'local-user', updated_at = now()
|
|
WHERE id = $1`, taskID, result); err != nil {
|
|
return errors.New("record task verification")
|
|
}
|
|
if err := appendAudit(ctx, tx, "task_verified", "manual_verification_recorded", task.AccountID, task.ConfirmationID,
|
|
task.ConfirmationVersion, "", task.ID, map[string]string{"verification_result": result}); err != nil {
|
|
return err
|
|
}
|
|
return commit(tx)
|
|
}
|
|
|
|
func (s *Store) ResumeTask(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 task resume transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
var accountID, confirmationID string
|
|
var confirmationVersion int64
|
|
err = tx.QueryRowContext(ctx, `
|
|
SELECT task.account_id, task.confirmation_id, task.confirmation_version
|
|
FROM operation_task task
|
|
JOIN social_account account ON account.id = task.account_id
|
|
JOIN content_draft draft ON draft.id = task.draft_id
|
|
JOIN confirmation confirmation ON confirmation.id = task.confirmation_id
|
|
JOIN environment_binding binding ON binding.account_id = task.account_id
|
|
JOIN browser_env environment ON environment.alias = binding.browser_env_alias
|
|
LEFT JOIN network_exit network ON network.id = binding.network_exit_id
|
|
JOIN runtime_instance runtime ON runtime.binding_id = binding.id AND runtime.released_at IS NULL
|
|
WHERE task.id = $1 AND task.state IN ('policy_hold', 'needs_confirmation')
|
|
AND task.verification_result = 'not_executed'
|
|
AND account.status = 'active' AND account.authorization_status = 'authorized'
|
|
AND account.version = task.account_version
|
|
AND draft.account_id = task.account_id AND draft.version = task.draft_version
|
|
AND confirmation.account_id = task.account_id AND confirmation.account_version = task.account_version
|
|
AND confirmation.draft_id = task.draft_id AND confirmation.draft_version = task.draft_version
|
|
AND confirmation.version = task.confirmation_version
|
|
AND (binding.network_exit_id IS NULL OR network.health_status = 'healthy') AND NOT binding.runtime_cleanup_pending
|
|
AND runtime.binding_version = binding.version AND runtime.lease_until > now()
|
|
FOR UPDATE OF task, account, draft, confirmation, binding, environment, runtime`, taskID).
|
|
Scan(&accountID, &confirmationID, &confirmationVersion)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
task, taskErr := scanTask(tx.QueryRowContext(ctx, `
|
|
SELECT id, idempotency_key, account_id, account_version, draft_id, draft_version,
|
|
confirmation_id, confirmation_version, state, hold_reason, verification_result, verified_at, verified_by,
|
|
created_at, updated_at
|
|
FROM operation_task WHERE id = $1`, taskID))
|
|
if taskErr != nil {
|
|
return taskErr
|
|
}
|
|
if task.VerificationResult != "not_executed" || (task.State != "policy_hold" && task.State != "needs_confirmation") {
|
|
return ErrConflict
|
|
}
|
|
reason, unavailable, reasonErr := taskReadinessReason(ctx, tx, taskID)
|
|
if reasonErr != nil {
|
|
return reasonErr
|
|
}
|
|
if reason == "" {
|
|
return ErrConflict
|
|
}
|
|
return &ReadinessError{Reason: reason, Unavailable: unavailable}
|
|
}
|
|
if err != nil {
|
|
return errors.New("validate task resume readiness")
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE operation_task SET state = 'queued', hold_reason = NULL,
|
|
verification_result = NULL, verified_at = NULL, verified_by = NULL,
|
|
lease_owner = NULL, lease_until = NULL, updated_at = now()
|
|
WHERE id = $1`, taskID); err != nil {
|
|
return errors.New("resume task")
|
|
}
|
|
if err := appendAudit(ctx, tx, "task_resumed", "manual_verification_not_executed", accountID, confirmationID,
|
|
confirmationVersion, "", taskID, nil); err != nil {
|
|
return err
|
|
}
|
|
return commit(tx)
|
|
}
|
|
|
|
type rowQuerier interface {
|
|
QueryRowContext(context.Context, string, ...any) *sql.Row
|
|
}
|
|
|
|
func taskReadinessReason(ctx context.Context, queryer rowQuerier, taskID string) (string, bool, error) {
|
|
var reason string
|
|
err := queryer.QueryRowContext(ctx, `
|
|
SELECT CASE
|
|
WHEN account.id IS NULL THEN 'account_missing'
|
|
WHEN account.authorization_status = 'revoked' THEN 'account_revoked'
|
|
WHEN account.status <> 'active' THEN 'account_paused'
|
|
WHEN account.version <> task.account_version THEN 'account_version_changed'
|
|
WHEN draft.id IS NULL OR draft.account_id <> task.account_id OR draft.version <> task.draft_version THEN 'draft_version_changed'
|
|
WHEN confirmation.id IS NULL THEN 'confirmation_missing'
|
|
WHEN confirmation.account_id <> task.account_id OR confirmation.account_version <> task.account_version
|
|
OR confirmation.draft_id <> task.draft_id OR confirmation.draft_version <> task.draft_version
|
|
OR confirmation.version <> task.confirmation_version THEN 'confirmation_version_changed'
|
|
WHEN binding.id IS NULL THEN 'binding_missing'
|
|
WHEN environment.alias IS NULL THEN 'environment_missing'
|
|
WHEN binding.network_exit_id IS NOT NULL AND network.id IS NULL THEN 'exit_missing'
|
|
WHEN binding.network_exit_id IS NOT NULL AND network.health_status <> 'healthy' THEN 'exit_unhealthy'
|
|
WHEN binding.runtime_cleanup_pending THEN 'runtime_stop_pending'
|
|
WHEN runtime.id IS NULL THEN 'runtime_missing'
|
|
WHEN runtime.binding_version IS DISTINCT FROM binding.version THEN 'binding_version_changed'
|
|
WHEN runtime.lease_until <= now() THEN 'runtime_lease_expired'
|
|
ELSE ''
|
|
END
|
|
FROM operation_task task
|
|
LEFT JOIN social_account account ON account.id = task.account_id
|
|
LEFT JOIN content_draft draft ON draft.id = task.draft_id
|
|
LEFT JOIN confirmation confirmation ON confirmation.id = task.confirmation_id
|
|
LEFT JOIN environment_binding binding ON binding.account_id = task.account_id
|
|
LEFT JOIN browser_env environment ON environment.alias = binding.browser_env_alias
|
|
LEFT JOIN network_exit network ON network.id = binding.network_exit_id
|
|
LEFT JOIN runtime_instance runtime ON runtime.binding_id = binding.id AND runtime.released_at IS NULL
|
|
WHERE task.id = $1`, taskID).Scan(&reason)
|
|
if err != nil {
|
|
return "", false, rowError(err)
|
|
}
|
|
unavailable := reason == "binding_missing" || reason == "environment_missing" || reason == "exit_missing" ||
|
|
reason == "exit_unhealthy" || reason == "runtime_stop_pending" || reason == "runtime_missing" || reason == "runtime_lease_expired"
|
|
return reason, unavailable, nil
|
|
}
|
|
|
|
func (s *Store) FinishTask(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 task finish transaction")
|
|
}
|
|
defer tx.Rollback()
|
|
var accountID, result string
|
|
var confirmationID sql.NullString
|
|
var confirmationVersion sql.NullInt64
|
|
err = tx.QueryRowContext(ctx, `
|
|
UPDATE operation_task SET state = verification_result, hold_reason = NULL,
|
|
lease_owner = NULL, lease_until = NULL, updated_at = now()
|
|
WHERE id = $1 AND state IN ('policy_hold', 'needs_confirmation')
|
|
AND verification_result IN ('succeeded', 'failed')
|
|
RETURNING account_id, confirmation_id, confirmation_version, verification_result`, taskID).
|
|
Scan(&accountID, &confirmationID, &confirmationVersion, &result)
|
|
if err != nil {
|
|
return rowError(err)
|
|
}
|
|
if err := appendAudit(ctx, tx, "task_manually_finished", "manual_verification_"+result, accountID, confirmationID.String,
|
|
confirmationVersion.Int64, "", taskID, map[string]string{"state": result}); err != nil {
|
|
return err
|
|
}
|
|
return commit(tx)
|
|
}
|
|
|
|
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, holdReason 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,
|
|
hold_reason = CASE WHEN state = 'executing' THEN 'task_result_uncertain' END,
|
|
verification_result = NULL, verified_at = NULL, verified_by = NULL,
|
|
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, COALESCE(hold_reason, ''), current_attempt_id, confirmation_id, confirmation_version`, taskID).Scan(
|
|
&accountID, &state, &holdReason, &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", "task_cancel_requested", accountID, confirmationID.String, confirmationVersion.Int64, attemptID.String, taskID, map[string]string{"state": state}); err != nil {
|
|
return err
|
|
}
|
|
return s.commitTransitions(tx, []taskstate.Transition{{State: state, ReasonCode: holdReason, AccountID: accountID, TaskID: taskID}})
|
|
}
|
|
|
|
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()
|
|
expired, err := quarantineExpired(ctx, tx)
|
|
if err != nil {
|
|
return Execution{}, err
|
|
}
|
|
invalid, err := quarantineInvalid(ctx, tx)
|
|
if err != nil {
|
|
return Execution{}, err
|
|
}
|
|
transitions := append(expired, invalid...)
|
|
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
|
|
JOIN environment_binding binding ON binding.account_id = a.id
|
|
JOIN browser_env environment ON environment.alias = binding.browser_env_alias
|
|
LEFT JOIN network_exit network ON network.id = binding.network_exit_id
|
|
JOIN runtime_instance runtime ON runtime.binding_id = binding.id
|
|
AND runtime.released_at IS NULL AND runtime.lease_until > now() AND runtime.binding_version = binding.version
|
|
WHERE t.state = 'queued' AND a.status = 'active' AND a.authorization_status = 'authorized'
|
|
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
|
|
AND (binding.network_exit_id IS NULL OR network.health_status = 'healthy') AND NOT binding.runtime_cleanup_pending
|
|
ORDER BY t.created_at, t.id
|
|
FOR UPDATE OF t, a, binding, runtime SKIP LOCKED LIMIT 1
|
|
)
|
|
UPDATE operation_task t SET state = 'executing', hold_reason = NULL,
|
|
verification_result = NULL, verified_at = NULL, verified_by = NULL,
|
|
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 := s.commitTransitions(tx, transitions); 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", "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 := s.commitTransitions(tx, transitions); 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
|
|
var holdReason string
|
|
err = tx.QueryRowContext(ctx, `
|
|
SELECT task.lease_until, CASE
|
|
WHEN account.id IS NULL THEN 'account_missing'
|
|
WHEN account.authorization_status <> 'authorized' THEN 'account_revoked'
|
|
WHEN account.status <> 'active' THEN 'account_paused'
|
|
WHEN account.version <> task.account_version THEN 'account_version_changed'
|
|
WHEN draft.id IS NULL OR draft.account_id <> task.account_id OR draft.version <> task.draft_version THEN 'draft_version_changed'
|
|
WHEN confirmation.id IS NULL OR confirmation.account_id <> task.account_id
|
|
OR confirmation.account_version <> task.account_version OR confirmation.draft_id <> task.draft_id
|
|
OR confirmation.draft_version <> task.draft_version OR confirmation.version <> task.confirmation_version THEN 'confirmation_version_changed'
|
|
WHEN binding.id IS NULL THEN 'binding_missing'
|
|
WHEN environment.alias IS NULL THEN 'environment_missing'
|
|
WHEN binding.network_exit_id IS NOT NULL AND network.id IS NULL THEN 'exit_missing'
|
|
WHEN binding.network_exit_id IS NOT NULL AND network.health_status <> 'healthy' THEN 'exit_unhealthy'
|
|
WHEN binding.runtime_cleanup_pending THEN 'runtime_stop_pending'
|
|
WHEN runtime.id IS NULL OR runtime.lease_until <= now() THEN 'runtime_missing'
|
|
WHEN claim.binding_version IS DISTINCT FROM binding.version
|
|
OR claim.browser_env_alias IS DISTINCT FROM environment.alias
|
|
OR claim.network_exit_id IS DISTINCT FROM network.id
|
|
OR claim.runtime_instance_id IS DISTINCT FROM runtime.id
|
|
OR runtime.binding_version IS DISTINCT FROM binding.version THEN 'task_result_uncertain'
|
|
ELSE ''
|
|
END
|
|
FROM operation_task task
|
|
LEFT JOIN LATERAL (
|
|
SELECT * FROM social_account WHERE id = task.account_id FOR UPDATE
|
|
) account ON true
|
|
LEFT JOIN LATERAL (
|
|
SELECT * FROM content_draft WHERE id = task.draft_id FOR UPDATE
|
|
) draft ON true
|
|
LEFT JOIN LATERAL (
|
|
SELECT * FROM confirmation WHERE id = task.confirmation_id FOR UPDATE
|
|
) confirmation ON true
|
|
LEFT JOIN LATERAL (
|
|
SELECT * FROM environment_binding WHERE account_id = task.account_id FOR UPDATE
|
|
) binding ON true
|
|
LEFT JOIN LATERAL (
|
|
SELECT * FROM browser_env WHERE alias = binding.browser_env_alias FOR UPDATE
|
|
) environment ON true
|
|
LEFT JOIN LATERAL (
|
|
SELECT * FROM network_exit WHERE id = binding.network_exit_id FOR UPDATE
|
|
) network ON true
|
|
LEFT JOIN LATERAL (
|
|
SELECT * FROM runtime_instance WHERE binding_id = binding.id AND released_at IS NULL FOR UPDATE
|
|
) runtime ON true
|
|
LEFT JOIN LATERAL (
|
|
SELECT browser_env_alias, network_exit_id, runtime_instance_id, binding_version
|
|
FROM audit_event WHERE attempt_id = $3 AND event_type = 'task_claimed' ORDER BY id DESC LIMIT 1
|
|
) claim ON true
|
|
WHERE task.id = $1 AND task.state = 'executing' AND task.lease_owner = $2 AND task.current_attempt_id = $3
|
|
FOR UPDATE OF task`, execution.TaskID, workerID, execution.AttemptID).Scan(&leaseUntil, &holdReason)
|
|
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()) || holdReason != "" {
|
|
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 state == "needs_confirmation" && holdReason == "" {
|
|
holdReason = "task_result_uncertain"
|
|
} else if state == "policy_hold" {
|
|
holdReason = "task_policy_hold"
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE operation_task SET state = $1, hold_reason = NULLIF($2, ''),
|
|
verification_result = NULL, verified_at = NULL, verified_by = NULL,
|
|
lease_owner = NULL, lease_until = NULL, updated_at = now() WHERE id = $3`, state, holdReason, execution.TaskID); err != nil {
|
|
return Execution{}, errors.New("finish task")
|
|
}
|
|
reason := map[string]string{
|
|
"succeeded": "task_succeeded", "failed": "task_failed",
|
|
"needs_confirmation": "task_result_uncertain", "policy_hold": "task_policy_hold",
|
|
}[state]
|
|
if holdReason != "" && state == "needs_confirmation" {
|
|
reason = holdReason
|
|
}
|
|
if err := appendAudit(ctx, tx, "task_finished", reason, execution.AccountID, execution.ConfirmationID, execution.ConfirmationVersion, execution.AttemptID, execution.TaskID, map[string]string{"state": state}); err != nil {
|
|
return Execution{}, err
|
|
}
|
|
if err := s.commitTransitions(tx, []taskstate.Transition{{
|
|
State: state, ReasonCode: holdReason, AccountID: execution.AccountID, TaskID: execution.TaskID,
|
|
}}); err != nil {
|
|
return Execution{}, err
|
|
}
|
|
execution.State = state
|
|
return execution, nil
|
|
}
|
|
|
|
func quarantineExpired(ctx context.Context, tx *sql.Tx) ([]taskstate.Transition, error) {
|
|
rows, err := tx.QueryContext(ctx, `
|
|
UPDATE operation_task SET state = 'needs_confirmation', hold_reason = 'execution_lease_expired',
|
|
verification_result = NULL, verified_at = NULL, verified_by = NULL,
|
|
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 nil, 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 nil, errors.New("read expired task")
|
|
}
|
|
expired = append(expired, expiredTask{taskID, accountID, attemptID.String, confirmationID.String, confirmationVersion.Int64})
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, errors.New("close expired task rows")
|
|
}
|
|
transitions := make([]taskstate.Transition, 0, len(expired))
|
|
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 nil, errors.New("mark expired attempt uncertain")
|
|
}
|
|
}
|
|
if err := appendAudit(ctx, tx, "task_lease_expired", "execution_lease_expired", task.accountID, task.confirmationID, task.confirmationVersion, task.attemptID, task.taskID, nil); err != nil {
|
|
return nil, err
|
|
}
|
|
transitions = append(transitions, taskstate.Transition{
|
|
State: "needs_confirmation", ReasonCode: "execution_lease_expired", AccountID: task.accountID, TaskID: task.taskID,
|
|
})
|
|
}
|
|
return transitions, nil
|
|
}
|
|
|
|
func quarantineInvalid(ctx context.Context, tx *sql.Tx) ([]taskstate.Transition, error) {
|
|
rows, err := tx.QueryContext(ctx, `
|
|
WITH invalid AS (
|
|
SELECT t.id,
|
|
CASE
|
|
WHEN account.id IS NULL OR account.status <> 'active' OR account.authorization_status <> 'authorized' THEN 'policy_hold'
|
|
WHEN account.version <> t.account_version THEN 'needs_confirmation'
|
|
WHEN draft.id IS NULL OR draft.account_id <> t.account_id OR draft.version <> t.draft_version THEN 'needs_confirmation'
|
|
WHEN confirmation.id IS NULL OR confirmation.account_id <> t.account_id
|
|
OR confirmation.account_version <> t.account_version OR confirmation.draft_id <> t.draft_id
|
|
OR confirmation.draft_version <> t.draft_version OR confirmation.version <> t.confirmation_version THEN 'needs_confirmation'
|
|
WHEN runtime.id IS NOT NULL AND runtime.binding_version IS DISTINCT FROM binding.version THEN 'needs_confirmation'
|
|
ELSE 'policy_hold'
|
|
END AS state,
|
|
CASE
|
|
WHEN account.id IS NULL THEN 'account_missing'
|
|
WHEN account.authorization_status = 'revoked' THEN 'account_revoked'
|
|
WHEN account.status <> 'active' THEN 'account_paused'
|
|
WHEN account.version <> t.account_version THEN 'account_version_changed'
|
|
WHEN draft.id IS NULL OR draft.account_id <> t.account_id OR draft.version <> t.draft_version THEN 'draft_version_changed'
|
|
WHEN confirmation.id IS NULL THEN 'confirmation_missing'
|
|
WHEN confirmation.account_id <> t.account_id OR confirmation.account_version <> t.account_version
|
|
OR confirmation.draft_id <> t.draft_id OR confirmation.draft_version <> t.draft_version
|
|
OR confirmation.version <> t.confirmation_version THEN 'confirmation_version_changed'
|
|
WHEN binding.id IS NULL THEN 'binding_missing'
|
|
WHEN environment.alias IS NULL THEN 'environment_missing'
|
|
WHEN binding.network_exit_id IS NOT NULL AND network.id IS NULL THEN 'exit_missing'
|
|
WHEN binding.network_exit_id IS NOT NULL AND network.health_status <> 'healthy' THEN 'exit_unhealthy'
|
|
WHEN binding.runtime_cleanup_pending THEN 'runtime_stop_pending'
|
|
WHEN runtime.id IS NULL THEN 'runtime_missing'
|
|
WHEN runtime.binding_version IS DISTINCT FROM binding.version THEN 'binding_version_changed'
|
|
WHEN runtime.lease_until <= now() THEN 'runtime_lease_expired'
|
|
END AS reason_code
|
|
FROM operation_task t
|
|
LEFT JOIN social_account account ON account.id = t.account_id
|
|
LEFT JOIN content_draft draft ON draft.id = t.draft_id
|
|
LEFT JOIN confirmation confirmation ON confirmation.id = t.confirmation_id
|
|
LEFT JOIN environment_binding binding ON binding.account_id = t.account_id
|
|
LEFT JOIN browser_env environment ON environment.alias = binding.browser_env_alias
|
|
LEFT JOIN network_exit network ON network.id = binding.network_exit_id
|
|
LEFT JOIN runtime_instance runtime ON runtime.binding_id = binding.id AND runtime.released_at IS NULL
|
|
WHERE t.state = 'queued' AND (
|
|
account.id IS NULL OR account.status <> 'active' OR account.authorization_status <> 'authorized'
|
|
OR account.version <> t.account_version
|
|
OR draft.id IS NULL OR draft.account_id <> t.account_id OR draft.version <> t.draft_version
|
|
OR confirmation.id IS NULL OR confirmation.account_id <> t.account_id
|
|
OR confirmation.account_version <> t.account_version OR confirmation.draft_id <> t.draft_id
|
|
OR confirmation.draft_version <> t.draft_version OR confirmation.version <> t.confirmation_version
|
|
OR binding.id IS NULL OR environment.alias IS NULL
|
|
OR (binding.network_exit_id IS NOT NULL AND (network.id IS NULL OR network.health_status <> 'healthy'))
|
|
OR binding.runtime_cleanup_pending OR runtime.id IS NULL
|
|
OR runtime.binding_version IS DISTINCT FROM binding.version OR runtime.lease_until <= now()
|
|
)
|
|
FOR UPDATE OF t SKIP LOCKED
|
|
)
|
|
UPDATE operation_task task SET state = invalid.state, hold_reason = invalid.reason_code,
|
|
verification_result = NULL, verified_at = NULL, verified_by = NULL, updated_at = now()
|
|
FROM invalid WHERE task.id = invalid.id
|
|
RETURNING task.id, task.account_id, task.confirmation_id, task.confirmation_version, task.state, invalid.reason_code`)
|
|
if err != nil {
|
|
return nil, errors.New("quarantine invalid tasks")
|
|
}
|
|
type invalidTask struct {
|
|
taskID, accountID, confirmationID, state, reasonCode string
|
|
confirmationVersion int64
|
|
}
|
|
var invalid []invalidTask
|
|
for rows.Next() {
|
|
var taskID, accountID, state, reasonCode string
|
|
var confirmationID sql.NullString
|
|
var confirmationVersion sql.NullInt64
|
|
if err := rows.Scan(&taskID, &accountID, &confirmationID, &confirmationVersion, &state, &reasonCode); err != nil {
|
|
rows.Close()
|
|
return nil, errors.New("read quarantined task")
|
|
}
|
|
invalid = append(invalid, invalidTask{taskID, accountID, confirmationID.String, state, reasonCode, confirmationVersion.Int64})
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, errors.New("close quarantined task rows")
|
|
}
|
|
transitions := make([]taskstate.Transition, 0, len(invalid))
|
|
for _, task := range invalid {
|
|
if err := appendAudit(ctx, tx, "task_quarantined", task.reasonCode, task.accountID, task.confirmationID, task.confirmationVersion, "", task.taskID, map[string]string{"state": task.state}); err != nil {
|
|
return nil, err
|
|
}
|
|
transitions = append(transitions, taskstate.Transition{
|
|
State: task.state, ReasonCode: task.reasonCode, AccountID: task.accountID, TaskID: task.taskID,
|
|
})
|
|
}
|
|
return transitions, nil
|
|
}
|
|
|
|
func scanTransitions(rows *sql.Rows, failure string) ([]taskstate.Transition, error) {
|
|
defer rows.Close()
|
|
var transitions []taskstate.Transition
|
|
for rows.Next() {
|
|
var transition taskstate.Transition
|
|
if err := rows.Scan(&transition.State, &transition.ReasonCode, &transition.AccountID, &transition.TaskID); err != nil {
|
|
return nil, errors.New(failure)
|
|
}
|
|
transitions = append(transitions, transition)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, errors.New(failure)
|
|
}
|
|
return transitions, nil
|
|
}
|
|
|
|
func (s *Store) Audit(ctx context.Context) ([]AuditEvent, error) {
|
|
page, err := s.ListAudit(ctx, AuditFilter{Page: 1, PageSize: 1000})
|
|
return page.Data, err
|
|
}
|
|
|
|
func (s *Store) ListAudit(ctx context.Context, filter AuditFilter) (AuditPage, error) {
|
|
if (filter.AccountID != "" && !idPattern.MatchString(filter.AccountID)) ||
|
|
(filter.TaskID != "" && !refPattern.MatchString(filter.TaskID)) ||
|
|
(filter.AttemptID != "" && !refPattern.MatchString(filter.AttemptID)) ||
|
|
(filter.BrowserEnvAlias != "" && !refPattern.MatchString(filter.BrowserEnvAlias)) ||
|
|
(filter.NetworkExitID != "" && !refPattern.MatchString(filter.NetworkExitID)) ||
|
|
(filter.EventType != "" && !eventPattern.MatchString(filter.EventType)) || filter.Page < 1 ||
|
|
filter.PageSize < 1 || filter.PageSize > 1000 || (filter.From != nil && filter.To != nil && filter.From.After(*filter.To)) {
|
|
return AuditPage{}, ErrInvalid
|
|
}
|
|
var from, to any
|
|
if filter.From != nil {
|
|
from = *filter.From
|
|
}
|
|
if filter.To != nil {
|
|
to = *filter.To
|
|
}
|
|
var total int
|
|
err := s.db.QueryRowContext(ctx, `
|
|
SELECT count(*) FROM audit_event
|
|
WHERE ($1 = '' OR account_id = $1) AND ($2 = '' OR task_id = $2) AND ($3 = '' OR attempt_id = $3)
|
|
AND ($4 = '' OR browser_env_alias = $4) AND ($5 = '' OR network_exit_id = $5) AND ($6 = '' OR event_type = $6)
|
|
AND ($7::timestamptz IS NULL OR created_at >= $7) AND ($8::timestamptz IS NULL OR created_at <= $8)`,
|
|
filter.AccountID, filter.TaskID, filter.AttemptID, filter.BrowserEnvAlias, filter.NetworkExitID, filter.EventType, from, to).Scan(&total)
|
|
if err != nil {
|
|
return AuditPage{}, errors.New("count audit events")
|
|
}
|
|
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
|
|
WHERE ($1 = '' OR account_id = $1) AND ($2 = '' OR task_id = $2) AND ($3 = '' OR attempt_id = $3)
|
|
AND ($4 = '' OR browser_env_alias = $4) AND ($5 = '' OR network_exit_id = $5) AND ($6 = '' OR event_type = $6)
|
|
AND ($7::timestamptz IS NULL OR created_at >= $7) AND ($8::timestamptz IS NULL OR created_at <= $8)
|
|
ORDER BY created_at DESC, id DESC LIMIT $9 OFFSET $10`, filter.AccountID, filter.TaskID, filter.AttemptID,
|
|
filter.BrowserEnvAlias, filter.NetworkExitID, filter.EventType, from, to, filter.PageSize, (filter.Page-1)*filter.PageSize)
|
|
if err != nil {
|
|
return AuditPage{}, errors.New("read audit events")
|
|
}
|
|
defer rows.Close()
|
|
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 AuditPage{}, 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
|
|
event.Details = safeDetails(event.Details)
|
|
events = append(events, event)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return AuditPage{}, errors.New("read audit events")
|
|
}
|
|
return AuditPage{Data: events, Total: total, Page: filter.Page, PageSize: filter.PageSize}, nil
|
|
}
|
|
|
|
func safeDetails(raw json.RawMessage) json.RawMessage {
|
|
var value any
|
|
if json.Unmarshal(raw, &value) != nil {
|
|
return json.RawMessage(`{}`)
|
|
}
|
|
value = allowDetails(value)
|
|
encoded, err := json.Marshal(value)
|
|
if err != nil {
|
|
return json.RawMessage(`{}`)
|
|
}
|
|
return encoded
|
|
}
|
|
|
|
var allowedDetailKeys = map[string]bool{
|
|
"platform": true, "draft_id": true, "draft_version": true, "account_version": true,
|
|
"tasks_held": true, "attempts_interrupted": true, "verification_result": true,
|
|
"state": true, "worker_id": true, "mock_outcome": true,
|
|
}
|
|
|
|
func allowDetails(value any) any {
|
|
switch value := value.(type) {
|
|
case map[string]any:
|
|
for key, child := range value {
|
|
if !allowedDetailKeys[strings.ToLower(key)] {
|
|
delete(value, key)
|
|
continue
|
|
}
|
|
value[key] = allowDetails(child)
|
|
}
|
|
case []any:
|
|
for index, child := range value {
|
|
value[index] = allowDetails(child)
|
|
}
|
|
}
|
|
return value
|
|
}
|
|
|
|
func appendAudit(ctx context.Context, tx *sql.Tx, eventType, reasonCode, 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,
|
|
browser_env_alias, network_exit_id, runtime_instance_id, binding_version, actor, reason_code, details)
|
|
SELECT $1, NULLIF($3, ''), NULLIF($4, ''), NULLIF($5, 0), NULLIF($6, ''), NULLIF($7, ''),
|
|
binding.browser_env_alias, binding.network_exit_id, runtime.id, binding.version, 'local-user', $2, $8
|
|
FROM (VALUES (1)) AS singleton(value)
|
|
LEFT JOIN environment_binding binding ON binding.account_id = NULLIF($3, '')
|
|
LEFT JOIN runtime_instance runtime ON runtime.binding_id = binding.id AND runtime.released_at IS NULL`,
|
|
eventType, reasonCode, 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 NewAccountID() string { return "account-" + newID()[:24] }
|
|
|
|
func commit(tx *sql.Tx) error {
|
|
if err := tx.Commit(); err != nil {
|
|
return errors.New("commit transaction")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) commitTransitions(tx *sql.Tx, transitions []taskstate.Transition) error {
|
|
if err := commit(tx); err != nil {
|
|
return err
|
|
}
|
|
s.notifyTransitions(transitions)
|
|
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")
|
|
}
|