CosmicClash/server/store/proposal_recovery_sql.go

package store

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/json"
	"fmt"
	"time"

	"github.com/cosmic-clash/cosmic-clash/server/domain"
)

const ProposalExpireSQL = `UPDATE proposals
SET state = 'EXPIRED', revision = revision + 1
WHERE proposal_id = $1 AND state = 'OPEN' AND expires_at <= $2`

const ProposalParticipantExpireSQL = `UPDATE proposal_participants
SET response = 'TIMED_OUT', responded_at = $2
WHERE proposal_id = $1 AND response = 'PENDING'
  AND EXISTS (SELECT 1 FROM proposals WHERE proposals.proposal_id = proposal_participants.proposal_id AND proposals.expires_at <= $2)`

const ProposalRecoverySelectSQL = `SELECT proposal_id, playlist, state, revision, expires_at
FROM proposals
WHERE proposal_id = $1
  AND EXISTS (SELECT 1 FROM proposal_participants WHERE proposal_id = proposals.proposal_id AND player_id = $2)`

const ProposalParticipantsSelectSQL = `SELECT player_id, response
FROM proposal_participants
WHERE proposal_id = $1
ORDER BY player_id`

const ProposalResponseIdempotencyScope = "proposal.respond"

const ProposalResponseIdempotencyInsertSQL = `INSERT INTO idempotency_keys
    (scope, idempotency_key, payload_digest, result)
VALUES ($1, $2, $3, $4)
ON CONFLICT (scope, idempotency_key) DO NOTHING`

const ProposalResponseIdempotencySelectSQL = `SELECT payload_digest, result
FROM idempotency_keys
WHERE scope = $1 AND idempotency_key = $2
FOR UPDATE`

const ProposalLockSQL = `SELECT playlist, state, revision, expires_at
FROM proposals
WHERE proposal_id = $1
FOR UPDATE`

const ProposalParticipantLockSQL = `SELECT response
FROM proposal_participants
WHERE proposal_id = $1 AND player_id = $2
FOR UPDATE`

const ProposalParticipantRespondSQL = `UPDATE proposal_participants
SET response = $3, responded_at = $4
WHERE proposal_id = $1 AND player_id = $2 AND response = 'PENDING'`

const ProposalCountPendingSQL = `SELECT COUNT(*)
FROM proposal_participants
WHERE proposal_id = $1 AND response = 'PENDING'`

const ProposalAcceptSQL = `UPDATE proposals
SET state = 'ACCEPTED', revision = revision + 1
WHERE proposal_id = $1 AND state = 'OPEN'`

const ProposalDeclineSQL = `UPDATE proposals
SET state = 'DECLINED', revision = revision + 1
WHERE proposal_id = $1 AND state = 'OPEN'`

const ProposalRevisionBumpSQL = `UPDATE proposals
SET revision = revision + 1
WHERE proposal_id = $1 AND state = 'OPEN'`

var ErrProposalResponseConflict = fmt.Errorf("proposal response conflict")

// GetProposal recovers the full proposal only after proving the caller is a
// participant. Expiry is advanced in the same transaction as the read so a
// missed event cannot leave a durable proposal indefinitely OPEN.
func GetProposal(ctx context.Context, db *sql.DB, playerID, proposalID string, now time.Time) (domain.Proposal, error) {
	if db == nil || playerID == "" || proposalID == "" || now.IsZero() {
		return domain.Proposal{}, fmt.Errorf("invalid proposal recovery arguments")
	}
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return domain.Proposal{}, err
	}
	defer tx.Rollback()
	if _, err := tx.ExecContext(ctx, ProposalExpireSQL, proposalID, now); err != nil {
		return domain.Proposal{}, err
	}
	if _, err := tx.ExecContext(ctx, ProposalParticipantExpireSQL, proposalID, now); err != nil {
		return domain.Proposal{}, err
	}
	var proposal domain.Proposal
	var playlist, state string
	if err := tx.QueryRowContext(ctx, ProposalRecoverySelectSQL, proposalID, playerID).Scan(&proposal.ProposalID, &playlist, &state, &proposal.Revision, &proposal.ExpiresAt); err != nil {
		return domain.Proposal{}, err
	}
	proposal.Playlist = domain.Playlist(playlist)
	proposal.State = domain.State(state)
	rows, err := tx.QueryContext(ctx, ProposalParticipantsSelectSQL, proposalID)
	if err != nil {
		return domain.Proposal{}, err
	}
	defer rows.Close()
	for rows.Next() {
		var participant domain.ProposalParticipant
		if err := rows.Scan(&participant.PlayerID, &participant.Response); err != nil {
			return domain.Proposal{}, err
		}
		proposal.Participants = append(proposal.Participants, participant)
	}
	if err := rows.Err(); err != nil {
		return domain.Proposal{}, err
	}
	if len(proposal.Participants) == 0 {
		return domain.Proposal{}, fmt.Errorf("proposal has no participants")
	}
	if err := tx.Commit(); err != nil {
		return domain.Proposal{}, err
	}
	return proposal, nil
}

// RespondToProposal is the durable mutation counterpart to GetProposal. The
// proposal row and participant row are locked in one transaction; the result
// is stored under the idempotency key before the transaction commits.
func RespondToProposal(ctx context.Context, db *sql.DB, playerID, proposalID, idempotencyKey string, accept bool, expectedRevision uint64, now time.Time) (domain.Proposal, error) {
	if db == nil || playerID == "" || proposalID == "" || len(idempotencyKey) < 16 || len(idempotencyKey) > 128 || now.IsZero() {
		return domain.Proposal{}, fmt.Errorf("invalid proposal response arguments")
	}
	digest := sha256.Sum256([]byte(fmt.Sprintf("%s|%s|%t|%d", playerID, proposalID, accept, expectedRevision)))
	var proposal domain.Proposal
	err := RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
		result, err := tx.ExecContext(ctx, ProposalResponseIdempotencyInsertSQL, ProposalResponseIdempotencyScope, idempotencyKey, digest[:], []byte("{}"))
		if err != nil {
			return err
		}
		inserted, err := result.RowsAffected()
		if err != nil {
			return err
		}
		if inserted == 0 {
			var priorDigest, priorResult []byte
			if err := tx.QueryRowContext(ctx, ProposalResponseIdempotencySelectSQL, ProposalResponseIdempotencyScope, idempotencyKey).Scan(&priorDigest, &priorResult); err != nil {
				return err
			}
			if !bytes.Equal(priorDigest, digest[:]) {
				return ErrProposalResponseConflict
			}
			if err := json.Unmarshal(priorResult, &proposal); err != nil {
				return fmt.Errorf("invalid stored proposal response: %w", err)
			}
			return nil
		}

		var playlist, state string
		var revision uint64
		var expiresAt time.Time
		if err := tx.QueryRowContext(ctx, ProposalLockSQL, proposalID).Scan(&playlist, &state, &revision, &expiresAt); err != nil {
			return err
		}
		// A mutation is also a recovery boundary. If the response arrives after
		// the window, advance both the proposal and its pending participants in
		// this same transaction before returning the closed error. Otherwise a
		// client that missed the expiry event could observe OPEN/PENDING forever
		// when its first durable interaction is an accept/decline.
		if _, err := tx.ExecContext(ctx, ProposalExpireSQL, proposalID, now); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, ProposalParticipantExpireSQL, proposalID, now); err != nil {
			return err
		}
		if !now.Before(expiresAt) {
			return domain.ErrProposalClosed
		}
		if state != string(domain.Open) || !now.Before(expiresAt) {
			return domain.ErrProposalClosed
		}
		if revision != expectedRevision {
			return domain.ErrStaleRevision
		}
		var response string
		if err := tx.QueryRowContext(ctx, ProposalParticipantLockSQL, proposalID, playerID).Scan(&response); err != nil {
			return domain.ErrNotParticipant
		}
		if response != string(domain.Pending) {
			return domain.ErrConflict
		}
		response = string(domain.DeclinedResponse)
		if accept {
			response = string(domain.AcceptedResponse)
		}
		changed, err := tx.ExecContext(ctx, ProposalParticipantRespondSQL, proposalID, playerID, response, now)
		if err != nil {
			return err
		}
		if count, err := changed.RowsAffected(); err != nil || count != 1 {
			return ErrProposalResponseConflict
		}
		targetState := string(domain.Declined)
		if accept {
			var pending int
			if err := tx.QueryRowContext(ctx, ProposalCountPendingSQL, proposalID).Scan(&pending); err != nil {
				return err
			}
			if pending == 0 {
				targetState = string(domain.Accepted)
			} else {
				targetState = state
			}
		}
		if targetState != state {
			if targetState == string(domain.Accepted) {
				_, err = tx.ExecContext(ctx, ProposalAcceptSQL, proposalID)
			} else {
				_, err = tx.ExecContext(ctx, ProposalDeclineSQL, proposalID)
			}
			if err != nil {
				return err
			}
			revision++
		} else {
			if _, err := tx.ExecContext(ctx, ProposalRevisionBumpSQL, proposalID); err != nil {
				return err
			}
			revision++
		}
		proposal = domain.Proposal{ProposalID: proposalID, Playlist: domain.Playlist(playlist), State: domain.State(targetState), Revision: revision, ExpiresAt: expiresAt}
		rows, err := tx.QueryContext(ctx, ProposalParticipantsSelectSQL, proposalID)
		if err != nil {
			return err
		}
		for rows.Next() {
			var participant domain.ProposalParticipant
			if err := rows.Scan(&participant.PlayerID, &participant.Response); err != nil {
				rows.Close()
				return err
			}
			proposal.Participants = append(proposal.Participants, participant)
		}
		if err := rows.Err(); err != nil {
			rows.Close()
			return err
		}
		rows.Close()
		stored, err := json.Marshal(proposal)
		if err != nil {
			return err
		}
		_, err = tx.ExecContext(ctx, `UPDATE idempotency_keys SET result = $3 WHERE scope = $1 AND idempotency_key = $2`, ProposalResponseIdempotencyScope, idempotencyKey, stored)
		return err
	})
	return proposal, err
}