diff --git a/server/domain/state.go b/server/domain/state.go new file mode 100644 index 00000000..5032a3d9 --- /dev/null +++ b/server/domain/state.go @@ -0,0 +1,149 @@ +// Package domain contains database-independent matchmaking invariants. +// Adapters may persist these records in PostgreSQL, but must not redefine +// transition, revision, or idempotency behavior. +package domain + +import ( + "bytes" + "crypto/sha256" + "errors" + "fmt" +) + +type ResourceKind string + +const ( + QueueTicket ResourceKind = "queue_ticket" + Proposal ResourceKind = "proposal" + Match ResourceKind = "match" +) + +type State string + +const ( + Queued State = "QUEUED" + Proposed State = "PROPOSED" + Accepted State = "ACCEPTED" + Allocating State = "ALLOCATING" + ProcessReady State = "PROCESS_READY" + AssignmentReady State = "ASSIGNMENT_READY" + Assigned State = "ASSIGNED" + Connecting State = "CONNECTING" + Live State = "LIVE" + ResultPending State = "RESULT_PENDING" + Completed State = "COMPLETED" + Cancelled State = "CANCELLED" + Expired State = "EXPIRED" + Failed State = "FAILED" + Open State = "OPEN" + Declined State = "DECLINED" +) + +var ( + ErrConflict = errors.New("mutation conflict") + ErrStaleRevision = errors.New("stale revision") + ErrIllegalTransition = errors.New("illegal state transition") +) + +type Record struct { + Kind ResourceKind + ID string + State State + Revision uint64 + idempotent map[string]appliedMutation +} + +type appliedMutation struct { + payloadDigest [32]byte + result Result +} + +type Result struct { + Kind ResourceKind + ID string + State State + Revision uint64 +} + +func NewRecord(kind ResourceKind, id string, state State) *Record { + return &Record{Kind: kind, ID: id, State: state, idempotent: make(map[string]appliedMutation)} +} + +// Apply performs all validation before changing the record. Replaying an +// identical idempotency key returns the original result without advancing the +// revision. Reusing a key with a different payload, or presenting a stale +// revision, is inert and returns an error. +func (r *Record) Apply(idempotencyKey string, payload []byte, expectedRevision uint64, target State) (Result, error) { + if idempotencyKey == "" { + return Result{}, fmt.Errorf("%w: empty idempotency key", ErrConflict) + } + digest := sha256.Sum256(payload) + if prior, ok := r.idempotent[idempotencyKey]; ok { + if !bytes.Equal(prior.payloadDigest[:], digest[:]) { + return Result{}, fmt.Errorf("%w: idempotency key reused with different payload", ErrConflict) + } + return prior.result, nil + } + if expectedRevision != r.Revision { + return Result{}, fmt.Errorf("%w: expected %d, current %d", ErrStaleRevision, expectedRevision, r.Revision) + } + if !legalTransition(r.Kind, r.State, target) { + return Result{}, fmt.Errorf("%w: %s %s -> %s", ErrIllegalTransition, r.Kind, r.State, target) + } + + r.State = target + r.Revision++ + result := Result{Kind: r.Kind, ID: r.ID, State: r.State, Revision: r.Revision} + r.idempotent[idempotencyKey] = appliedMutation{payloadDigest: digest, result: result} + return result, nil +} + +func legalTransition(kind ResourceKind, from, to State) bool { + var targets []State + switch kind { + case QueueTicket: + targets = queueTransitions[from] + case Proposal: + targets = proposalTransitions[from] + case Match: + targets = matchTransitions[from] + default: + return false + } + for _, target := range targets { + if target == to { + return true + } + } + return false +} + +var queueTransitions = map[State][]State{ + Queued: {Proposed, Cancelled, Expired}, + Proposed: {Queued, Accepted, Cancelled, Expired}, + Accepted: {Queued, Allocating, Cancelled, Failed}, + Allocating: {ProcessReady, Failed, Cancelled}, + ProcessReady: {AssignmentReady, Failed, Cancelled}, + AssignmentReady: {Assigned, Failed, Cancelled}, + Assigned: {Connecting, Failed, Cancelled}, + Connecting: {Live, Failed, Expired}, + Live: {ResultPending, Failed}, + ResultPending: {Completed, Failed}, + Completed: {}, Cancelled: {}, Expired: {}, Failed: {}, +} + +var proposalTransitions = map[State][]State{ + Open: {Accepted, Declined, Expired, Cancelled}, + Accepted: {}, Declined: {}, Expired: {}, Cancelled: {}, +} + +var matchTransitions = map[State][]State{ + Allocating: {ProcessReady, Failed, Cancelled}, + ProcessReady: {AssignmentReady, Failed, Cancelled}, + AssignmentReady: {Assigned, Failed, Cancelled}, + Assigned: {Connecting, Failed, Cancelled}, + Connecting: {Live, Failed, Cancelled}, + Live: {ResultPending, Failed}, + ResultPending: {Completed, Failed}, + Completed: {}, Cancelled: {}, Failed: {}, +} diff --git a/server/domain/state_test.go b/server/domain/state_test.go new file mode 100644 index 00000000..cc5d880a --- /dev/null +++ b/server/domain/state_test.go @@ -0,0 +1,60 @@ +package domain + +import ( + "errors" + "testing" +) + +func TestApplyIsAtomicOnIllegalTransitionAndStaleRevision(t *testing.T) { + r := NewRecord(QueueTicket, "ticket_1234567890123456", Queued) + if _, err := r.Apply("k1", []byte(`{"state":"LIVE"}`), 0, Live); !errors.Is(err, ErrIllegalTransition) { + t.Fatalf("illegal transition error = %v", err) + } + if r.State != Queued || r.Revision != 0 { + t.Fatalf("illegal transition mutated record: %+v", r) + } + if _, err := r.Apply("k2", []byte(`{}`), 99, Proposed); !errors.Is(err, ErrStaleRevision) { + t.Fatalf("stale revision error = %v", err) + } + if r.State != Queued || r.Revision != 0 { + t.Fatalf("stale revision mutated record: %+v", r) + } +} + +func TestApplyReplaysIdenticalIdempotencyWithoutNewRevision(t *testing.T) { + r := NewRecord(QueueTicket, "ticket_1234567890123456", Queued) + payload := []byte(`{"state":"PROPOSED"}`) + first, err := r.Apply("same-key-123456", payload, 0, Proposed) + if err != nil { + t.Fatal(err) + } + second, err := r.Apply("same-key-123456", payload, 0, Proposed) + if err != nil { + t.Fatal(err) + } + if first != second || r.Revision != 1 { + t.Fatalf("replay advanced or changed result: first=%+v second=%+v record=%+v", first, second, r) + } +} + +func TestApplyRejectsIdempotencyKeyPayloadConfusion(t *testing.T) { + r := NewRecord(QueueTicket, "ticket_1234567890123456", Queued) + if _, err := r.Apply("same-key-123456", []byte("a"), 0, Proposed); err != nil { + t.Fatal(err) + } + if _, err := r.Apply("same-key-123456", []byte("b"), 1, Accepted); !errors.Is(err, ErrConflict) { + t.Fatalf("conflicting replay error = %v", err) + } + if r.State != Proposed || r.Revision != 1 { + t.Fatalf("conflicting replay mutated record: %+v", r) + } +} + +func TestTerminalStatesCannotAdvance(t *testing.T) { + for _, state := range []State{Completed, Cancelled, Expired, Failed} { + r := NewRecord(QueueTicket, "ticket_1234567890123456", state) + if _, err := r.Apply("terminal-key-123", []byte("x"), 0, Live); !errors.Is(err, ErrIllegalTransition) { + t.Fatalf("%s transition error = %v", state, err) + } + } +} diff --git a/server/go.mod b/server/go.mod new file mode 100644 index 00000000..406160f9 --- /dev/null +++ b/server/go.mod @@ -0,0 +1,3 @@ +module github.com/cosmic-clash/cosmic-clash/server + +go 1.23