casino/pkg/room/reconcile.go

package room

import (
	"context"
	"fmt"
	"time"

	"github.com/drjones/quantum-arcade/pkg/ledger"
	"github.com/jackc/pgx/v5/pgxpool"
)

// Reconciler refunds rounds that were abandoned mid-flight.
//
// Stakes are debited when a bet is placed, so if the instance driving a game
// dies between the bet and settlement, those stakes sit with the house and the
// round never resolves. The books stay balanced — nothing is created or lost —
// but the players are quietly short, which is the same thing as being robbed
// by accident.
//
// This finds those rounds and refunds every unsettled stake. It is safe to run
// repeatedly and from any instance: refunds are recorded against the round, and
// a round already marked settled is skipped.
type Reconciler struct {
	pool   *pgxpool.Pool
	ledger *ledger.Ledger

	// Stale is how long a round may remain unsettled before it is considered
	// abandoned. It must exceed the longest possible round plus the time it
	// takes another instance to take over, or a live round would be refunded
	// out from under the players still in it.
	Stale time.Duration
}

func NewReconciler(pool *pgxpool.Pool, l *ledger.Ledger) *Reconciler {
	return &Reconciler{
		pool:   pool,
		ledger: l,
		// A round is capped at 60s of flight plus its betting and settle
		// phases; leadership moves within LeaseTTL. Two minutes is far past
		// any legitimate round and still prompt enough to matter at a party.
		Stale: 2 * time.Minute,
	}
}

// Result describes what a reconciliation pass did.
type Result struct {
	RoundsRefunded int
	BetsRefunded   int
	MsatRefunded   int64
}

// Run refunds every abandoned round it finds.
func (rc *Reconciler) Run(ctx context.Context) (Result, error) {
	var res Result

	rows, err := rc.pool.Query(ctx, `
		SELECT DISTINCT r.id
		  FROM rounds r
		  JOIN bets b ON b.round_id = r.id
		 WHERE r.settled_at IS NULL
		   AND r.voided_at IS NULL
		   AND b.settled_at IS NULL
		   AND r.opened_at < now() - make_interval(secs => $1)
		 ORDER BY r.id`,
		rc.Stale.Seconds())
	if err != nil {
		return res, fmt.Errorf("finding abandoned rounds: %w", err)
	}

	var roundIDs []int64
	for rows.Next() {
		var id int64
		if err := rows.Scan(&id); err != nil {
			rows.Close()
			return res, err
		}
		roundIDs = append(roundIDs, id)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return res, err
	}

	for _, roundID := range roundIDs {
		refunded, msat, err := rc.refundRound(ctx, roundID)
		if err != nil {
			// One bad round must not stop the rest from being made whole.
			fmt.Printf("reconcile: round %d: %v\n", roundID, err)
			continue
		}
		if refunded > 0 {
			res.RoundsRefunded++
			res.BetsRefunded += refunded
			res.MsatRefunded += msat
		}
	}
	return res, nil
}

// refundRound returns every unsettled stake in one round.
func (rc *Reconciler) refundRound(ctx context.Context, roundID int64) (int, int64, error) {
	// Claim the round by voiding it. Doing this before moving money means a
	// second pass — or another instance running concurrently — finds nothing
	// to do, so a refund cannot be issued twice.
	//
	// Void, not settled: an abandoned round produced no outcome, so it has no
	// seed to reveal and must not masquerade as a resolved round.
	tag, err := rc.pool.Exec(ctx,
		`UPDATE rounds SET voided_at = now()
		  WHERE id = $1 AND settled_at IS NULL AND voided_at IS NULL`, roundID)
	if err != nil {
		return 0, 0, fmt.Errorf("claiming round: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return 0, 0, nil // another pass got there first
	}

	rows, err := rc.pool.Query(ctx,
		`SELECT account_id, stake_msat FROM bets
		  WHERE round_id = $1 AND settled_at IS NULL`, roundID)
	if err != nil {
		return 0, 0, err
	}
	type refund struct {
		account int64
		msat    int64
	}
	var refunds []refund
	for rows.Next() {
		var r refund
		if err := rows.Scan(&r.account, &r.msat); err != nil {
			rows.Close()
			return 0, 0, err
		}
		refunds = append(refunds, r)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return 0, 0, err
	}
	if len(refunds) == 0 {
		return 0, 0, nil
	}

	house, err := rc.ledger.AccountByName(ctx, "house_pot")
	if err != nil {
		return 0, 0, err
	}

	postings := make([]ledger.Posting, 0, len(refunds)+1)
	var total int64
	for _, r := range refunds {
		postings = append(postings, ledger.Posting{AccountID: r.account, AmountMsat: r.msat})
		total += r.msat
	}
	postings = append(postings, ledger.Posting{AccountID: house, AmountMsat: -total})

	rid := roundID
	if _, err := rc.ledger.Post(ctx, "refund_abandoned", &rid, postings); err != nil {
		return 0, 0, fmt.Errorf("posting refunds: %w", err)
	}

	if _, err := rc.pool.Exec(ctx,
		`UPDATE bets SET settled_at = now(), payout_msat = stake_msat
		  WHERE round_id = $1 AND settled_at IS NULL`, roundID); err != nil {
		return 0, 0, fmt.Errorf("marking bets refunded: %w", err)
	}

	return len(refunds), total, nil
}

// RunPeriodically sweeps for abandoned rounds until the context ends.
func (rc *Reconciler) RunPeriodically(ctx context.Context, every time.Duration) {
	t := time.NewTicker(every)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			res, err := rc.Run(ctx)
			if err != nil {
				fmt.Printf("reconcile: %v\n", err)
				continue
			}
			if res.RoundsRefunded > 0 {
				fmt.Printf("reconcile: refunded %d bets across %d abandoned rounds (%d msat)\n",
					res.BetsRefunded, res.RoundsRefunded, res.MsatRefunded)
			}
		}
	}
}