mirror of
https://github.com/prosolis/gogobee.git
synced 2026-07-15 00:32:40 +00:00
euro: money that moves over a retrying wire needs a name
Every caller of Credit/Debit today is a Matrix message, and a Matrix message arrives once. The Pete games escrow will claim buy-ins over a poll loop, where a claim that succeeds but whose ack is lost gets retried — and the player pays twice. euro_transactions had no external id and no unique constraint to stop it. Adds external_id + a partial unique index, and CreditIdem/DebitIdem: balance mutation and transaction log in one tx, keyed by the escrow GUID. A replay is a no-op that still reports ok; a rejection leaves no trace, so the same GUID can be retried once the player is good for it.
This commit is contained in:
@@ -452,6 +452,11 @@ func runMigrations(d *sql.DB) error {
|
|||||||
// the buyer when the target survives (the unseal), so piling on carries the
|
// the buyer when the target survives (the unseal), so piling on carries the
|
||||||
// same exposure the original purchase does.
|
// same exposure the original purchase does.
|
||||||
`ALTER TABLE mischief_contracts ADD COLUMN escalated_by TEXT`,
|
`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 {
|
for _, stmt := range columnMigrations {
|
||||||
if _, err := d.Exec(stmt); err != nil {
|
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)
|
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
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,8 +1,11 @@
|
|||||||
package plugin
|
package plugin
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"database/sql"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
|
"math"
|
||||||
"os"
|
"os"
|
||||||
"regexp"
|
"regexp"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -413,6 +416,131 @@ func (p *EuroPlugin) GetBalance(userID id.UserID) float64 {
|
|||||||
return balance
|
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) {
|
func (p *EuroPlugin) logTransaction(userID id.UserID, amount float64, reason string) {
|
||||||
db.Exec("euro: log transaction",
|
db.Exec("euro: log transaction",
|
||||||
"INSERT INTO euro_transactions (user_id, amount, reason) VALUES (?, ?, ?)",
|
"INSERT INTO euro_transactions (user_id, amount, reason) VALUES (?, ?, ?)",
|
||||||
|
|||||||
174
internal/plugin/euro_idem_test.go
Normal file
174
internal/plugin/euro_idem_test.go
Normal file
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user