amadan

carlos / platform Public

console: sequenced self-serve events and their at-most-once delivery

InsertSelfServeEventTx allocates each event's seq from the row's
event_seq inside the caller's transaction; DeliverSelfServeEvent marks an
event emitted and inserts its audit row (auditActivityTx) in one
transaction, so two workers deliver it once and a failure delivers
nothing.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude bc27b0ee9f065606ea18fd78ce05a7da28241f7f parent 29cee09
1 file changed, +80 −0
  • internal/console/selfserve_store.go +80 −0
diff --git a/internal/console/selfserve_store.go b/internal/console/selfserve_store.go
index d7b955abf..02630c569 100644
--- a/internal/console/selfserve_store.go
+++ b/internal/console/selfserve_store.go
@@ -15,6 +15,7 @@ package console
// Timestamps are Unix nanoseconds; zero is "not yet".
import (
+ "context"
"database/sql"
"encoding/json"
"fmt"
@@ -418,3 +419,82 @@ func ssWon(res sql.Result, err error) (bool, error) {
}
return n == 1, nil
}
+
+// ---- events -----------------------------------------------------------------
+
+// InsertSelfServeEventTx inserts e inside tx, allocating its seq from the
+// row's event_seq in the same transaction (event_seq = event_seq + 1). The
+// account, app and host are the row's. e.At zero is now.
+func InsertSelfServeEventTx(tx *sql.Tx, e SelfServeEvent) error {
+ res, err := tx.Exec(`UPDATE self_serve_moves SET event_seq = event_seq + 1 WHERE id = ?`, e.ID)
+ if err := ssOne(res, err, "event", e.ID); err != nil {
+ return err
+ }
+ if e.At.IsZero() {
+ e.At = time.Now().UTC()
+ }
+ if err := tx.QueryRow(`SELECT event_seq, account, app, host FROM self_serve_moves WHERE id = ?`, e.ID).
+ Scan(&e.Seq, &e.Account, &e.App, &e.Host); err != nil {
+ return err
+ }
+ _, err = tx.Exec(`INSERT INTO self_serve_events (id, seq, phase, account, app, host, at, reason, detail, emitted)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 0)`, e.ID, e.Seq, e.Phase, e.Account, e.App, e.Host, ssNanos(e.At), e.Reason, e.Detail)
+ return err
+}
+
+// UnemittedSelfServeEvents are up to limit undelivered events, oldest first.
+func (s *Store) UnemittedSelfServeEvents(limit int) ([]SelfServeEvent, error) {
+ rows, err := s.db.Query(`SELECT id, seq, phase, account, app, host, at, reason, detail
+FROM self_serve_events WHERE emitted = 0 ORDER BY at, id, seq LIMIT ?`, limit)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []SelfServeEvent
+ for rows.Next() {
+ var e SelfServeEvent
+ var at int64
+ if err := rows.Scan(&e.ID, &e.Seq, &e.Phase, &e.Account, &e.App, &e.Host, &at, &e.Reason, &e.Detail); err != nil {
+ return nil, err
+ }
+ e.At = ssTime(at)
+ out = append(out, e)
+ }
+ return out, rows.Err()
+}
+
+// DeliverSelfServeEvent delivers e at most once, in one transaction: it
+// first marks the event emitted (WHERE id = ? AND seq = ? AND emitted = 0);
+// matching nothing commits nothing and returns (another worker delivered
+// it); otherwise deliver inserts the audit row inside tx. A failure rolls
+// both back, leaving the event undelivered.
+func (s *Store) DeliverSelfServeEvent(ctx context.Context, e SelfServeEvent, deliver func(tx *sql.Tx) error) error {
+ tx, err := s.db.BeginTx(ctx, nil)
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+ res, err := tx.ExecContext(ctx, `UPDATE self_serve_events SET emitted = 1 WHERE id = ? AND seq = ? AND emitted = 0`, e.ID, e.Seq)
+ won, err := ssWon(res, err)
+ if err != nil || !won {
+ return err
+ }
+ if err := deliver(tx); err != nil {
+ return err
+ }
+ return tx.Commit()
+}
+
+// auditActivityTx inserts one audit row inside tx, for the account named by
+// its sqid: the transactional twin of RecordActivity, so an event's audit
+// line and its emitted mark commit together.
+func auditActivityTx(tx *sql.Tx, acct, app, address string, a Activity) error {
+ var accountID int64
+ if err := tx.QueryRow(`SELECT id FROM accounts WHERE sqid = ?`, acct).Scan(&accountID); err != nil {
+ return fmt.Errorf("console: audit for account %q: %w", acct, err)
+ }
+ _, err := tx.Exec(`INSERT INTO audit
+ (account_id, app, address, action, target_type, target, message, detail, created_at)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, accountID, app, address, a.Action, a.TargetType, a.Target, a.Message, a.Detail, time.Now().Unix())
+ return err
+}