| 1 | package workspacestate |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | ) |
| 7 | |
| 8 | // Command receipts survive subsequent lifecycle changes. Retrying an old |
| 9 | // completed request must not execute it against the session's new state. |
| 10 | func (s *Store) BeginCommand(ctx context.Context, id, fingerprint string, request json.RawMessage, expected uint64) error { |
| 11 | return s.beginCommand(ctx, id, fingerprint, request, expected, false) |
| 12 | } |
| 13 | |
| 14 | // BeginPurgeCommand admits an observed snapshot without granting deletion |
| 15 | // authority. The purge transaction checks each target's generation under lock. |
| 16 | func (s *Store) BeginPurgeCommand(ctx context.Context, id, fingerprint string, request json.RawMessage, expected uint64) error { |
| 17 | var command struct { |
| 18 | Action string `json:"action"` |
| 19 | } |
| 20 | if err := json.Unmarshal(request, &command); err != nil { |
| 21 | return err |
| 22 | } |
| 23 | if command.Action != "purge" { |
| 24 | return ErrMutationConflict |
| 25 | } |
| 26 | return s.beginCommand(ctx, id, fingerprint, request, expected, true) |
| 27 | } |
| 28 | |
| 29 | func (s *Store) beginCommand(ctx context.Context, id, fingerprint string, request json.RawMessage, expected uint64, purge bool) error { |
| 30 | return s.mutate(ctx, func(state *State) error { |
| 31 | if old, exists := state.PendingOperations[id]; exists { |
| 32 | if old.Kind != "command" || old.RequestFingerprint != fingerprint { |
| 33 | return ErrMutationConflict |
| 34 | } |
| 35 | return nil |
| 36 | } |
| 37 | if expected > state.Generation || (!purge && expected != state.Generation) { |
| 38 | return ErrMutationConflict |
| 39 | } |
| 40 | state.PendingOperations[id] = Operation{ID: id, Kind: "command", Phase: "prepared", Lifecycle: Active, SessionIDs: []string{}, ExpectedGeneration: expected, RequestFingerprint: fingerprint, Request: request} |
| 41 | return nil |
| 42 | }) |
| 43 | } |
| 44 | |
| 45 | func (s *Store) SaveCommandResult(ctx context.Context, id string, result json.RawMessage, complete bool) error { |
| 46 | return s.mutate(ctx, func(state *State) error { |
| 47 | op, exists := state.PendingOperations[id] |
| 48 | if !exists || op.Kind != "command" { |
| 49 | return ErrMutationConflict |
| 50 | } |
| 51 | if op.Phase == "committed" { |
| 52 | return nil |
| 53 | } |
| 54 | // Stamp the receipt under the same writer lock as its commit. Other |
| 55 | // processes may have advanced generation since the caller's snapshot. |
| 56 | var fields map[string]json.RawMessage |
| 57 | if err := json.Unmarshal(result, &fields); err != nil { |
| 58 | return err |
| 59 | } |
| 60 | fields["generation"], _ = json.Marshal(state.Generation + 1) |
| 61 | encoded, err := json.Marshal(fields) |
| 62 | if err != nil { |
| 63 | return err |
| 64 | } |
| 65 | op.Result = encoded |
| 66 | if complete { |
| 67 | op.Phase = "committed" |
| 68 | } |
| 69 | op.ResultGeneration = state.Generation + 1 |
| 70 | state.PendingOperations[id] = op |
| 71 | return nil |
| 72 | }) |
| 73 | } |
| 74 |