diff --git a/internal/db/db.go b/internal/db/db.go index fd34640..018add9 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -452,6 +452,11 @@ func runMigrations(d *sql.DB) error { // the buyer when the target survives (the unseal), so piling on carries the // same exposure the original purchase does. `ALTER TABLE mischief_contracts ADD COLUMN escalated_by TEXT`, + // Pete games — a caller-supplied idempotency key for money moves that + // arrive over a retrying wire rather than a Matrix message. A Matrix + // message arrives once; a poll loop whose ack is lost on the wire will + // retry, and without this the player pays twice. See euro.DebitIdem. + `ALTER TABLE euro_transactions ADD COLUMN external_id TEXT`, } for _, stmt := range columnMigrations { if _, err := d.Exec(stmt); err != nil { @@ -468,6 +473,18 @@ func runMigrations(d *sql.DB) error { return fmt.Errorf("migration %q: %w", stmt, err) } } + + // Indexes over columns the column migrations above just added. These can't + // live in the schema block, which runs before those columns exist. + indexMigrations := []string{ + `CREATE UNIQUE INDEX IF NOT EXISTS idx_euro_tx_external + ON euro_transactions(external_id) WHERE external_id IS NOT NULL`, + } + for _, stmt := range indexMigrations { + if _, err := d.Exec(stmt); err != nil { + return fmt.Errorf("index migration %q: %w", stmt, err) + } + } return nil } diff --git a/internal/plugin/euro.go b/internal/plugin/euro.go index 455b55b..e7a4772 100644 --- a/internal/plugin/euro.go +++ b/internal/plugin/euro.go @@ -1,8 +1,11 @@ package plugin import ( + "database/sql" + "errors" "fmt" "log/slog" + "math" "os" "regexp" "strings" @@ -413,6 +416,131 @@ func (p *EuroPlugin) GetBalance(userID id.UserID) float64 { return balance } +// --------------------------------------------------------------------------- +// Idempotent money moves +// +// Every caller of Credit/Debit above is a Matrix message, which arrives exactly +// once. Money that moves over a retrying wire — the Pete games escrow poll loop +// — has no such guarantee: a claim that succeeds but whose ack is lost gets +// retried, and the player pays twice. CreditIdem/DebitIdem take an externalID +// (the escrow GUID), do the balance mutation and the transaction log in one +// tx, and lean on idx_euro_tx_external so a replay is a no-op rather than a +// second debit. +// +// Anything web-initiated goes through these. Nothing else should. +// --------------------------------------------------------------------------- + +// DebitIdem subtracts euros exactly once for a given externalID. Returns ok=false +// only when the debit would breach the debt limit; a replay of an already-applied +// externalID returns ok=true without moving money again. balanceAfter is the +// balance as of the decision, and is meaningless when ok is false. +func (p *EuroPlugin) DebitIdem(userID id.UserID, amount float64, reason, externalID string) (ok bool, balanceAfter float64, err error) { + if amount <= 0 { + return false, 0, fmt.Errorf("euro: DebitIdem non-positive amount %v", amount) + } + if externalID == "" { + return false, 0, fmt.Errorf("euro: DebitIdem requires an external id") + } + debtLimit := envFloat("BLACKJACK_DEBT_LIMIT", 1000) + return p.applyIdem(userID, -amount, reason, externalID, -debtLimit) +} + +// CreditIdem adds euros exactly once for a given externalID. A replay is a no-op +// that still reports ok. +func (p *EuroPlugin) CreditIdem(userID id.UserID, amount float64, reason, externalID string) (ok bool, balanceAfter float64, err error) { + if amount <= 0 { + return false, 0, fmt.Errorf("euro: CreditIdem non-positive amount %v", amount) + } + if externalID == "" { + return false, 0, fmt.Errorf("euro: CreditIdem requires an external id") + } + // A credit has no floor to breach — math.Inf would do, but the balance can + // never land below the current one, so any sentinel below it works. + return p.applyIdem(userID, amount, reason, externalID, math.Inf(-1)) +} + +// applyIdem is the shared body: signed delta, applied once, never letting the +// balance fall below floor. Balance mutation and transaction log commit together +// or not at all. +func (p *EuroPlugin) applyIdem(userID id.UserID, delta float64, reason, externalID string, floor float64) (bool, float64, error) { + // Outside the tx: db is capped at a single connection, so a nested Exec on + // the pool would deadlock against our own open transaction. + p.ensureBalance(userID) + + d := db.Get() + tx, err := d.Begin() + if err != nil { + return false, 0, fmt.Errorf("euro: begin: %w", err) + } + defer tx.Rollback() //nolint:errcheck // no-op after a successful Commit + + // Already applied? Report the current balance and move on. + var prior int + switch err := tx.QueryRow( + `SELECT 1 FROM euro_transactions WHERE external_id = ?`, externalID, + ).Scan(&prior); { + case err == nil: + bal, err := txBalance(tx, userID) + if err != nil { + return false, 0, err + } + slog.Info("euro: idempotent replay ignored", "user", userID, "external_id", externalID, "reason", reason) + return true, bal, nil + case errors.Is(err, sql.ErrNoRows): + // Not applied yet — carry on. + default: + return false, 0, fmt.Errorf("euro: idem lookup: %w", err) + } + + res, err := tx.Exec( + `UPDATE euro_balances SET balance = balance + ?, updated_at = CURRENT_TIMESTAMP + WHERE user_id = ? AND (balance + ?) >= ?`, + delta, string(userID), delta, floor, + ) + if err != nil { + return false, 0, fmt.Errorf("euro: idem balance update: %w", err) + } + if affected, _ := res.RowsAffected(); affected == 0 { + return false, 0, nil // would breach the floor + } + + if _, err := tx.Exec( + `INSERT INTO euro_transactions (user_id, amount, reason, external_id) VALUES (?, ?, ?, ?)`, + string(userID), delta, reason, externalID, + ); err != nil { + // A racing caller won with the same externalID. The unique index is the + // backstop the SELECT above can't be: roll back our half-applied delta + // and report theirs. + if strings.Contains(err.Error(), "UNIQUE constraint failed") { + if rbErr := tx.Rollback(); rbErr != nil { + return false, 0, fmt.Errorf("euro: rollback after idem race: %w", rbErr) + } + slog.Info("euro: idempotent race lost, delta rolled back", "user", userID, "external_id", externalID) + return true, p.GetBalance(userID), nil + } + return false, 0, fmt.Errorf("euro: idem tx log: %w", err) + } + + bal, err := txBalance(tx, userID) + if err != nil { + return false, 0, err + } + if err := tx.Commit(); err != nil { + return false, 0, fmt.Errorf("euro: commit: %w", err) + } + return true, bal, nil +} + +func txBalance(tx *sql.Tx, userID id.UserID) (float64, error) { + var bal float64 + if err := tx.QueryRow( + `SELECT balance FROM euro_balances WHERE user_id = ?`, string(userID), + ).Scan(&bal); err != nil { + return 0, fmt.Errorf("euro: read balance: %w", err) + } + return bal, nil +} + func (p *EuroPlugin) logTransaction(userID id.UserID, amount float64, reason string) { db.Exec("euro: log transaction", "INSERT INTO euro_transactions (user_id, amount, reason) VALUES (?, ?, ?)", diff --git a/internal/plugin/euro_idem_test.go b/internal/plugin/euro_idem_test.go new file mode 100644 index 0000000..b3961db --- /dev/null +++ b/internal/plugin/euro_idem_test.go @@ -0,0 +1,174 @@ +package plugin + +import ( + "sync" + "testing" + + "gogobee/internal/db" + + "maunium.net/go/mautrix/id" +) + +func newEuroTestDB(t *testing.T) *EuroPlugin { + t.Helper() + dir := t.TempDir() + db.Close() + if err := db.Init(dir); err != nil { + t.Fatal(err) + } + t.Cleanup(db.Close) + return NewEuroPlugin(nil) +} + +func seedBalance(t *testing.T, p *EuroPlugin, user id.UserID, amount float64) { + t.Helper() + p.ensureBalance(user) + if _, err := db.Get().Exec( + `UPDATE euro_balances SET balance = ? WHERE user_id = ?`, amount, string(user), + ); err != nil { + t.Fatal(err) + } +} + +func TestDebitIdem_ReplaySameGUIDDebitsOnce(t *testing.T) { + p := newEuroTestDB(t) + user := id.UserID("@replay:test.invalid") + seedBalance(t, p, user, 500) + + for i := 0; i < 3; i++ { + ok, bal, err := p.DebitIdem(user, 100, "games_buyin", "guid-abc") + if err != nil { + t.Fatalf("attempt %d: %v", i, err) + } + if !ok { + t.Fatalf("attempt %d: debit rejected", i) + } + if bal != 400 { + t.Fatalf("attempt %d: balance after = %v, want 400", i, bal) + } + } + + if got := p.GetBalance(user); got != 400 { + t.Fatalf("final balance = %v, want 400 (replay debited more than once)", got) + } + + var n int + if err := db.Get().QueryRow( + `SELECT COUNT(*) FROM euro_transactions WHERE external_id = 'guid-abc'`, + ).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 1 { + t.Fatalf("logged %d transactions for one guid, want 1", n) + } +} + +func TestCreditIdem_ReplaySameGUIDCreditsOnce(t *testing.T) { + p := newEuroTestDB(t) + user := id.UserID("@cashout:test.invalid") + seedBalance(t, p, user, 0) + + for i := 0; i < 3; i++ { + ok, bal, err := p.CreditIdem(user, 250, "games_cashout", "guid-xyz") + if err != nil { + t.Fatalf("attempt %d: %v", i, err) + } + if !ok || bal != 250 { + t.Fatalf("attempt %d: ok=%v balance=%v, want true/250", i, ok, bal) + } + } + + if got := p.GetBalance(user); got != 250 { + t.Fatalf("final balance = %v, want 250", got) + } +} + +func TestDebitIdem_DistinctGUIDsBothApply(t *testing.T) { + p := newEuroTestDB(t) + user := id.UserID("@distinct:test.invalid") + seedBalance(t, p, user, 500) + + for _, guid := range []string{"guid-1", "guid-2"} { + if ok, _, err := p.DebitIdem(user, 100, "games_buyin", guid); err != nil || !ok { + t.Fatalf("%s: ok=%v err=%v", guid, ok, err) + } + } + if got := p.GetBalance(user); got != 300 { + t.Fatalf("balance = %v, want 300", got) + } +} + +func TestDebitIdem_RespectsDebtLimitAndLogsNothing(t *testing.T) { + t.Setenv("BLACKJACK_DEBT_LIMIT", "1000") + p := newEuroTestDB(t) + user := id.UserID("@broke:test.invalid") + seedBalance(t, p, user, 0) + + ok, _, err := p.DebitIdem(user, 1500, "games_buyin", "guid-broke") + if err != nil { + t.Fatal(err) + } + if ok { + t.Fatal("debit past the debt limit was accepted") + } + if got := p.GetBalance(user); got != 0 { + t.Fatalf("balance = %v, want 0 — a rejected debit moved money", got) + } + + // A rejection must leave no trace, so the same guid can be retried later + // (e.g. once the player has earned their way back above the limit). + var n int + if err := db.Get().QueryRow( + `SELECT COUNT(*) FROM euro_transactions WHERE external_id = 'guid-broke'`, + ).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 0 { + t.Fatalf("rejected debit logged %d transactions, want 0", n) + } +} + +// The SELECT-then-UPDATE in applyIdem is not by itself a guard against two +// concurrent claims of the same guid; the unique index is. Hammer it. +func TestDebitIdem_ConcurrentSameGUIDDebitsOnce(t *testing.T) { + p := newEuroTestDB(t) + user := id.UserID("@race:test.invalid") + seedBalance(t, p, user, 1000) + + const racers = 8 + var wg sync.WaitGroup + errs := make([]error, racers) + for i := 0; i < racers; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + _, _, errs[i] = p.DebitIdem(user, 100, "games_buyin", "guid-race") + }(i) + } + wg.Wait() + + for i, err := range errs { + if err != nil { + t.Fatalf("racer %d: %v", i, err) + } + } + if got := p.GetBalance(user); got != 900 { + t.Fatalf("balance = %v, want 900 — %d concurrent claims of one guid double-debited", got, racers) + } +} + +func TestIdem_RejectsEmptyExternalID(t *testing.T) { + p := newEuroTestDB(t) + user := id.UserID("@noguid:test.invalid") + seedBalance(t, p, user, 500) + + if _, _, err := p.DebitIdem(user, 10, "games_buyin", ""); err == nil { + t.Fatal("DebitIdem accepted an empty external id") + } + if _, _, err := p.CreditIdem(user, 10, "games_cashout", ""); err == nil { + t.Fatal("CreditIdem accepted an empty external id") + } + if got := p.GetBalance(user); got != 500 { + t.Fatalf("balance = %v, want 500", got) + } +}