From b53b9f1adccceacc2825eab22ba14a78d0ce5cf4 Mon Sep 17 00:00:00 2001 From: youfak Date: Wed, 29 Jul 2026 18:27:03 +0800 Subject: [PATCH] feat: add transactional admin state reference store --- .../adminstate/contract_external_test.go | 15 + .../adminstate/contracttest/contract.go | 295 +++++++++++++ internal/domain/adminstate/memory.go | 412 ++++++++++++++++++ 3 files changed, 722 insertions(+) create mode 100644 internal/domain/adminstate/contract_external_test.go create mode 100644 internal/domain/adminstate/contracttest/contract.go create mode 100644 internal/domain/adminstate/memory.go diff --git a/internal/domain/adminstate/contract_external_test.go b/internal/domain/adminstate/contract_external_test.go new file mode 100644 index 0000000..dda194b --- /dev/null +++ b/internal/domain/adminstate/contract_external_test.go @@ -0,0 +1,15 @@ +package adminstate_test + +import ( + "testing" + + "proxy-pool/internal/domain/adminstate" + "proxy-pool/internal/domain/adminstate/contracttest" +) + +func TestMemoryStoreContract(t *testing.T) { + contracttest.Run(t, func(t *testing.T) adminstate.Store { + t.Helper() + return adminstate.NewMemoryStore() + }) +} diff --git a/internal/domain/adminstate/contracttest/contract.go b/internal/domain/adminstate/contracttest/contract.go new file mode 100644 index 0000000..c00759a --- /dev/null +++ b/internal/domain/adminstate/contracttest/contract.go @@ -0,0 +1,295 @@ +package contracttest + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "proxy-pool/internal/domain/adminstate" +) + +type Factory func(*testing.T) adminstate.Store + +func Run(t *testing.T, factory Factory) { + t.Helper() + t.Run("config transaction and immutability", func(t *testing.T) { + runConfigContract(t, factory(t)) + }) + t.Run("upstream idempotency", func(t *testing.T) { + runUpstreamContract(t, factory(t)) + }) + t.Run("routing compare and swap", func(t *testing.T) { + runRoutingContract(t, factory(t)) + }) + t.Run("outbox lease and acknowledgement", func(t *testing.T) { + runOutboxContract(t, factory(t)) + }) + t.Run("context cancellation", func(t *testing.T) { + runContextContract(t, factory(t)) + }) +} + +func runConfigContract(t *testing.T, store adminstate.Store) { + t.Helper() + command := configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes)) + result, err := store.CommitConfig(context.Background(), command) + if err != nil || !result.Changed || result.Revision != 1 || result.RequestID != command.RequestID { + t.Fatalf("CommitConfig(first) = %+v, %v", result, err) + } + + snapshot, err := store.Snapshot(context.Background()) + if err != nil || snapshot.Revision != 1 || snapshot.Config == nil || + snapshot.Config.ConfigVersion != "cfg-1" || len(snapshot.Upstreams) != 2 || len(snapshot.Routings) != 1 { + t.Fatalf("Snapshot() = %+v, %v", snapshot, err) + } + snapshot.Config.ConfigVersion = "mutated" + snapshot.Upstreams[0].Name = "mutated" + snapshot.Routings[0].Upstreams[0] = "mutated" + again, err := store.Snapshot(context.Background()) + if err != nil || again.Config.ConfigVersion != "cfg-1" || + again.Upstreams[0].Name == "mutated" || again.Routings[0].Upstreams[0] == "mutated" { + t.Fatalf("Snapshot() leaked mutable state: %+v, %v", again, err) + } + + command.RequestID = "req-config-replay" + command.OccurredAt = command.OccurredAt.Add(time.Second) + replayed, err := store.CommitConfig(context.Background(), command) + if err != nil || replayed.Changed || replayed.Revision != 1 { + t.Fatalf("CommitConfig(replay) = %+v, %v", replayed, err) + } + audits, err := store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 10}) + if err != nil || len(audits) != 2 || !audits[0].Changed || audits[1].Changed || + audits[0].Revision != 1 || audits[1].Revision != 1 { + t.Fatalf("ReadAudit() = %+v, %v", audits, err) + } + + conflict := command + conflict.RequestID = "req-config-conflict" + conflict.Checksum = strings.Repeat("b", adminstate.SHA256HexBytes) + if _, err := store.CommitConfig(context.Background(), conflict); !errors.Is(err, adminstate.ErrConflict) { + t.Fatalf("CommitConfig(conflict) error = %v", err) + } + audits, err = store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 10}) + if err != nil || len(audits) != 2 { + t.Fatalf("ReadAudit(after conflict) = %+v, %v", audits, err) + } + + invalid := command + invalid.RequestID = "req-config-invalid" + invalid.Routings[0].CurrentUpstream = "missing" + if _, err := store.CommitConfig(context.Background(), invalid); !errors.Is(err, adminstate.ErrInvalidCommand) { + t.Fatalf("CommitConfig(invalid) error = %v", err) + } + final, err := store.Snapshot(context.Background()) + if err != nil || final.Revision != 1 || final.Config.ConfigVersion != "cfg-1" { + t.Fatalf("Snapshot(after invalid) = %+v, %v", final, err) + } +} + +func runUpstreamContract(t *testing.T, store adminstate.Store) { + t.Helper() + now := contractNow() + commit(t, store, configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes))) + + result, err := store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{ + RequestID: "req-disable", Actor: contractActor(), OccurredAt: now.Add(time.Second), + Name: "provider-a", Enabled: false, + }) + if err != nil || !result.Changed || result.Revision != 2 { + t.Fatalf("SetUpstreamEnabled(disable) = %+v, %v", result, err) + } + replayed, err := store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{ + RequestID: "req-disable-replay", Actor: contractActor(), OccurredAt: now.Add(2 * time.Second), + Name: "provider-a", Enabled: false, + }) + if err != nil || replayed.Changed || replayed.Revision != 2 { + t.Fatalf("SetUpstreamEnabled(replay) = %+v, %v", replayed, err) + } + if _, err := store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{ + RequestID: "req-missing", Actor: contractActor(), OccurredAt: now.Add(3 * time.Second), + Name: "missing", Enabled: true, + }); !errors.Is(err, adminstate.ErrNotFound) { + t.Fatalf("SetUpstreamEnabled(missing) error = %v", err) + } + + snapshot, err := store.Snapshot(context.Background()) + if err != nil || snapshot.Revision != 2 || upstreamEnabled(snapshot, "provider-a") { + t.Fatalf("Snapshot() = %+v, %v", snapshot, err) + } + audits, err := store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 10}) + if err != nil || len(audits) != 3 || !audits[1].Changed || audits[2].Changed { + t.Fatalf("ReadAudit() = %+v, %v", audits, err) + } + events, err := store.Claim(context.Background(), adminstate.ClaimCommand{ + ConsumerID: "publisher-a", Now: now.Add(4 * time.Second), Limit: 10, Lease: time.Minute, + }) + if err != nil || len(events) != 2 || events[0].Revision != 1 || events[1].Revision != 2 { + t.Fatalf("Claim() = %+v, %v", events, err) + } +} + +func runRoutingContract(t *testing.T, store adminstate.Store) { + t.Helper() + now := contractNow() + commit(t, store, configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes))) + + if _, err := store.SwitchRouting(context.Background(), adminstate.SwitchRoutingCommand{ + RequestID: "req-bad-target", Actor: contractActor(), OccurredAt: now.Add(time.Second), + Name: "checkout", ExpectedCurrent: "provider-a", Target: "provider-c", + }); !errors.Is(err, adminstate.ErrInvalidCommand) { + t.Fatalf("SwitchRouting(bad target) error = %v", err) + } + + start := make(chan struct{}) + var changed atomic.Int64 + var conflicts atomic.Int64 + var unexpectedMu sync.Mutex + var unexpected []error + var workers sync.WaitGroup + for index := range 100 { + workers.Add(1) + go func(index int) { + defer workers.Done() + <-start + result, err := store.SwitchRouting(context.Background(), adminstate.SwitchRoutingCommand{ + RequestID: fmt.Sprintf("req-switch-%03d", index), Actor: contractActor(), + OccurredAt: now.Add(2 * time.Second), Name: "checkout", + ExpectedCurrent: "provider-a", Target: "provider-b", Reason: "capacity", + }) + switch { + case err == nil && result.Changed: + changed.Add(1) + case errors.Is(err, adminstate.ErrConflict): + conflicts.Add(1) + default: + unexpectedMu.Lock() + unexpected = append(unexpected, err) + unexpectedMu.Unlock() + } + }(index) + } + close(start) + workers.Wait() + if changed.Load() != 1 || conflicts.Load() != 99 || len(unexpected) != 0 { + t.Fatalf("concurrent switch changed=%d conflicts=%d unexpected=%v", changed.Load(), conflicts.Load(), unexpected) + } + snapshot, err := store.Snapshot(context.Background()) + if err != nil || snapshot.Revision != 2 || snapshot.Routings[0].CurrentUpstream != "provider-b" { + t.Fatalf("Snapshot() = %+v, %v", snapshot, err) + } +} + +func runOutboxContract(t *testing.T, store adminstate.Store) { + t.Helper() + now := contractNow() + commit(t, store, configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes))) + result, err := store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{ + RequestID: "req-disable", Actor: contractActor(), OccurredAt: now.Add(time.Second), + Name: "provider-a", Enabled: false, + }) + if err != nil || !result.Changed { + t.Fatalf("SetUpstreamEnabled() = %+v, %v", result, err) + } + + first, err := store.Claim(context.Background(), adminstate.ClaimCommand{ + ConsumerID: "publisher-a", Now: now.Add(2 * time.Second), Limit: 1, Lease: time.Minute, + }) + if err != nil || len(first) != 1 || first[0].ID != 1 { + t.Fatalf("Claim(first) = %+v, %v", first, err) + } + first[0].Payload[0] = 'X' + second, err := store.Claim(context.Background(), adminstate.ClaimCommand{ + ConsumerID: "publisher-b", Now: now.Add(2 * time.Second), Limit: 10, Lease: time.Minute, + }) + if err != nil || len(second) != 1 || second[0].ID != 2 { + t.Fatalf("Claim(second) = %+v, %v", second, err) + } + if err := store.Acknowledge(context.Background(), adminstate.AcknowledgeCommand{ + ConsumerID: "publisher-b", Now: now.Add(3 * time.Second), EventIDs: []uint64{1}, + }); !errors.Is(err, adminstate.ErrConflict) { + t.Fatalf("Acknowledge(wrong owner) error = %v", err) + } + + reclaimed, err := store.Claim(context.Background(), adminstate.ClaimCommand{ + ConsumerID: "publisher-b", Now: now.Add(2*time.Minute + time.Second), Limit: 10, Lease: time.Minute, + }) + if err != nil || len(reclaimed) != 2 || reclaimed[0].ID != 1 || reclaimed[1].ID != 2 || reclaimed[0].Payload[0] == 'X' { + t.Fatalf("Claim(reclaimed) = %+v, %v", reclaimed, err) + } + if err := store.Acknowledge(context.Background(), adminstate.AcknowledgeCommand{ + ConsumerID: "publisher-b", Now: now.Add(2*time.Minute + 2*time.Second), EventIDs: []uint64{1, 2}, + }); err != nil { + t.Fatalf("Acknowledge(valid): %v", err) + } + empty, err := store.Claim(context.Background(), adminstate.ClaimCommand{ + ConsumerID: "publisher-c", Now: now.Add(4 * time.Minute), Limit: 10, Lease: time.Minute, + }) + if err != nil || len(empty) != 0 { + t.Fatalf("Claim(after ACK) = %+v, %v", empty, err) + } +} + +func runContextContract(t *testing.T, store adminstate.Store) { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + cancel() + command := configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes)) + if _, err := store.CommitConfig(ctx, command); !errors.Is(err, context.Canceled) { + t.Fatalf("CommitConfig(canceled) error = %v", err) + } + if _, err := store.Snapshot(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("Snapshot(canceled) error = %v", err) + } + if _, err := store.ReadAudit(ctx, adminstate.AuditQuery{Limit: 1}); !errors.Is(err, context.Canceled) { + t.Fatalf("ReadAudit(canceled) error = %v", err) + } + if _, err := store.Claim(ctx, adminstate.ClaimCommand{ + ConsumerID: "publisher-a", Now: contractNow(), Limit: 1, Lease: time.Second, + }); !errors.Is(err, context.Canceled) { + t.Fatalf("Claim(canceled) error = %v", err) + } +} + +func configCommand(requestID, version, checksum string) adminstate.CommitConfigCommand { + return adminstate.CommitConfigCommand{ + RequestID: requestID, Actor: contractActor(), OccurredAt: contractNow(), + ConfigVersion: version, Checksum: checksum, Source: "configs/proxy-pool.yaml", + Upstreams: []adminstate.UpstreamDefinition{ + {Name: "provider-a", Enabled: true}, + {Name: "provider-b", Enabled: true}, + }, + Routings: []adminstate.RoutingDefinition{{ + Name: "checkout", Enabled: true, Upstreams: []string{"provider-a", "provider-b"}, + CurrentUpstream: "provider-a", + }}, + } +} + +func contractActor() adminstate.Actor { + return adminstate.Actor{ID: "admin-a", SourceIP: "192.0.2.10"} +} + +func contractNow() time.Time { + return time.Date(2026, 7, 29, 10, 0, 0, 0, time.UTC) +} + +func commit(t *testing.T, store adminstate.Store, command adminstate.CommitConfigCommand) { + t.Helper() + if result, err := store.CommitConfig(context.Background(), command); err != nil || !result.Changed { + t.Fatalf("CommitConfig() = %+v, %v", result, err) + } +} + +func upstreamEnabled(snapshot adminstate.Snapshot, name string) bool { + for _, upstream := range snapshot.Upstreams { + if upstream.Name == name { + return upstream.Enabled + } + } + return false +} diff --git a/internal/domain/adminstate/memory.go b/internal/domain/adminstate/memory.go new file mode 100644 index 0000000..5f563e7 --- /dev/null +++ b/internal/domain/adminstate/memory.go @@ -0,0 +1,412 @@ +package adminstate + +import ( + "context" + "encoding/json" + "math" + "net/netip" + "sort" + "sync" + "time" +) + +var _ Store = (*MemoryStore)(nil) + +type MemoryStore struct { + mu sync.Mutex + revision uint64 + config *ConfigRevision + configChecksums map[string]string + upstreams map[string]UpstreamState + routings map[string]RoutingState + audits []AuditRecord + events []Event + nextAuditID uint64 + nextEventID uint64 +} + +func NewMemoryStore() *MemoryStore { + return &MemoryStore{ + configChecksums: make(map[string]string), + upstreams: make(map[string]UpstreamState), + routings: make(map[string]RoutingState), + } +} + +func (store *MemoryStore) CommitConfig(ctx context.Context, command CommitConfigCommand) (MutationResult, error) { + result := MutationResult{RequestID: command.RequestID} + if err := contextError(ctx); err != nil { + return result, err + } + if store == nil || validateCommitConfigCommand(command) != nil { + return result, ErrInvalidCommand + } + command = cloneCommitConfigCommand(command) + + store.mu.Lock() + defer store.mu.Unlock() + if err := ctx.Err(); err != nil { + return result, err + } + if checksum, exists := store.configChecksums[command.ConfigVersion]; exists { + if checksum != command.Checksum || store.config == nil || store.config.ConfigVersion != command.ConfigVersion { + return result, ErrConflict + } + result.Revision = store.revision + store.appendAuditLocked(command.RequestID, command.Actor, ActionCommitConfig, "config", + command.ConfigVersion, false, store.revision, "", command.OccurredAt) + return result, nil + } + if store.revision == math.MaxUint64 { + return result, ErrUnavailable + } + + nextRevision := store.revision + 1 + nextConfig := &ConfigRevision{ + Revision: nextRevision, ConfigVersion: command.ConfigVersion, Checksum: command.Checksum, + Source: command.Source, CreatedAt: command.OccurredAt.UTC(), + } + nextUpstreams := make(map[string]UpstreamState, len(command.Upstreams)) + for _, definition := range command.Upstreams { + nextUpstreams[definition.Name] = UpstreamState{ + Name: definition.Name, Enabled: definition.Enabled, Revision: nextRevision, + UpdatedAt: command.OccurredAt.UTC(), + } + } + nextRoutings := make(map[string]RoutingState, len(command.Routings)) + for _, definition := range command.Routings { + nextRoutings[definition.Name] = RoutingState{ + Name: definition.Name, Enabled: definition.Enabled, + Upstreams: append([]string(nil), definition.Upstreams...), CurrentUpstream: definition.CurrentUpstream, + Revision: nextRevision, UpdatedAt: command.OccurredAt.UTC(), + } + } + payload, err := encodeEventPayload(map[string]any{ + "configVersion": command.ConfigVersion, + "checksum": command.Checksum, + "revision": nextRevision, + }) + if err != nil { + return result, ErrUnavailable + } + + store.revision = nextRevision + store.config = nextConfig + store.configChecksums[command.ConfigVersion] = command.Checksum + store.upstreams = nextUpstreams + store.routings = nextRoutings + store.appendAuditLocked(command.RequestID, command.Actor, ActionCommitConfig, "config", + command.ConfigVersion, true, nextRevision, "", command.OccurredAt) + store.appendEventLocked(nextRevision, "config.committed", "config", command.ConfigVersion, + payload, command.OccurredAt) + return MutationResult{RequestID: command.RequestID, Changed: true, Revision: nextRevision}, nil +} + +func (store *MemoryStore) SetUpstreamEnabled(ctx context.Context, command SetUpstreamCommand) (MutationResult, error) { + result := MutationResult{RequestID: command.RequestID} + if err := contextError(ctx); err != nil { + return result, err + } + if store == nil || validateSetUpstreamCommand(command) != nil { + return result, ErrInvalidCommand + } + + store.mu.Lock() + defer store.mu.Unlock() + if err := ctx.Err(); err != nil { + return result, err + } + state, exists := store.upstreams[command.Name] + if !exists { + return result, ErrNotFound + } + if state.Enabled == command.Enabled { + result.Revision = store.revision + store.appendAuditLocked(command.RequestID, command.Actor, ActionSetUpstream, "upstream", + command.Name, false, store.revision, "", command.OccurredAt) + return result, nil + } + if store.revision == math.MaxUint64 { + return result, ErrUnavailable + } + + nextRevision := store.revision + 1 + payload, err := encodeEventPayload(map[string]any{ + "enabled": command.Enabled, + "name": command.Name, + "revision": nextRevision, + }) + if err != nil { + return result, ErrUnavailable + } + state.Enabled = command.Enabled + state.Revision = nextRevision + state.UpdatedAt = command.OccurredAt.UTC() + store.revision = nextRevision + store.upstreams[command.Name] = state + store.appendAuditLocked(command.RequestID, command.Actor, ActionSetUpstream, "upstream", + command.Name, true, nextRevision, "", command.OccurredAt) + store.appendEventLocked(nextRevision, "upstream.enabled_changed", "upstream", command.Name, + payload, command.OccurredAt) + return MutationResult{RequestID: command.RequestID, Changed: true, Revision: nextRevision}, nil +} + +func (store *MemoryStore) SwitchRouting(ctx context.Context, command SwitchRoutingCommand) (MutationResult, error) { + result := MutationResult{RequestID: command.RequestID} + if err := contextError(ctx); err != nil { + return result, err + } + if store == nil || validateSwitchRoutingCommand(command) != nil { + return result, ErrInvalidCommand + } + + store.mu.Lock() + defer store.mu.Unlock() + if err := ctx.Err(); err != nil { + return result, err + } + state, exists := store.routings[command.Name] + if !exists { + return result, ErrNotFound + } + if !state.Enabled || state.CurrentUpstream != command.ExpectedCurrent { + return result, ErrConflict + } + if !contains(state.Upstreams, command.Target) { + return result, ErrInvalidCommand + } + if state.CurrentUpstream == command.Target { + result.Revision = store.revision + store.appendAuditLocked(command.RequestID, command.Actor, ActionSwitchRoute, "routing", + command.Name, false, store.revision, command.Reason, command.OccurredAt) + return result, nil + } + if store.revision == math.MaxUint64 { + return result, ErrUnavailable + } + + nextRevision := store.revision + 1 + payload, err := encodeEventPayload(map[string]any{ + "current": command.Target, + "name": command.Name, + "previous": command.ExpectedCurrent, + "reason": command.Reason, + "revision": nextRevision, + }) + if err != nil { + return result, ErrUnavailable + } + state.CurrentUpstream = command.Target + state.Revision = nextRevision + state.UpdatedAt = command.OccurredAt.UTC() + store.revision = nextRevision + store.routings[command.Name] = state + store.appendAuditLocked(command.RequestID, command.Actor, ActionSwitchRoute, "routing", + command.Name, true, nextRevision, command.Reason, command.OccurredAt) + store.appendEventLocked(nextRevision, "routing.switched", "routing", command.Name, + payload, command.OccurredAt) + return MutationResult{RequestID: command.RequestID, Changed: true, Revision: nextRevision}, nil +} + +func (store *MemoryStore) Snapshot(ctx context.Context) (Snapshot, error) { + if err := contextError(ctx); err != nil { + return Snapshot{}, err + } + if store == nil { + return Snapshot{}, ErrInvalidCommand + } + store.mu.Lock() + defer store.mu.Unlock() + if err := ctx.Err(); err != nil { + return Snapshot{}, err + } + + snapshot := Snapshot{Revision: store.revision} + if store.config != nil { + config := *store.config + snapshot.Config = &config + } + for _, upstream := range store.upstreams { + snapshot.Upstreams = append(snapshot.Upstreams, upstream) + } + for _, routing := range store.routings { + routing.Upstreams = append([]string(nil), routing.Upstreams...) + snapshot.Routings = append(snapshot.Routings, routing) + } + sort.Slice(snapshot.Upstreams, func(left, right int) bool { + return snapshot.Upstreams[left].Name < snapshot.Upstreams[right].Name + }) + sort.Slice(snapshot.Routings, func(left, right int) bool { + return snapshot.Routings[left].Name < snapshot.Routings[right].Name + }) + return cloneSnapshot(snapshot), nil +} + +func (store *MemoryStore) ReadAudit(ctx context.Context, query AuditQuery) ([]AuditRecord, error) { + if err := contextError(ctx); err != nil { + return nil, err + } + if store == nil || validateAuditQuery(query) != nil { + return nil, ErrInvalidCommand + } + store.mu.Lock() + defer store.mu.Unlock() + if err := ctx.Err(); err != nil { + return nil, err + } + result := make([]AuditRecord, 0, query.Limit) + for _, record := range store.audits { + if record.ID <= query.AfterID { + continue + } + result = append(result, record) + if len(result) == query.Limit { + break + } + } + return cloneAuditRecords(result), nil +} + +func (store *MemoryStore) Claim(ctx context.Context, command ClaimCommand) ([]Event, error) { + if err := contextError(ctx); err != nil { + return nil, err + } + if store == nil || validateClaimCommand(command) != nil { + return nil, ErrInvalidCommand + } + store.mu.Lock() + defer store.mu.Unlock() + if err := ctx.Err(); err != nil { + return nil, err + } + + claimUntil := command.Now.UTC().Add(command.Lease) + result := make([]Event, 0, command.Limit) + for index := range store.events { + event := &store.events[index] + if event.PublishedAt != nil || (event.ClaimUntil != nil && command.Now.Before(*event.ClaimUntil)) { + continue + } + event.ClaimedBy = command.ConsumerID + event.ClaimUntil = &claimUntil + result = append(result, cloneEvent(*event)) + if len(result) == command.Limit { + break + } + } + return result, nil +} + +func (store *MemoryStore) Acknowledge(ctx context.Context, command AcknowledgeCommand) error { + if err := contextError(ctx); err != nil { + return err + } + if store == nil || validateAcknowledgeCommand(command) != nil { + return ErrInvalidCommand + } + store.mu.Lock() + defer store.mu.Unlock() + if err := ctx.Err(); err != nil { + return err + } + + indexes := make([]int, 0, len(command.EventIDs)) + for _, eventID := range command.EventIDs { + index := store.eventIndexLocked(eventID) + if index < 0 { + return ErrNotFound + } + event := store.events[index] + if event.PublishedAt != nil || event.ClaimedBy != command.ConsumerID || event.ClaimUntil == nil || + !command.Now.Before(*event.ClaimUntil) { + return ErrConflict + } + indexes = append(indexes, index) + } + publishedAt := command.Now.UTC() + for _, index := range indexes { + store.events[index].PublishedAt = &publishedAt + } + return nil +} + +func (store *MemoryStore) appendAuditLocked( + requestID string, + actor Actor, + action Action, + resourceType string, + resourceName string, + changed bool, + revision uint64, + reason string, + occurredAt time.Time, +) { + store.nextAuditID++ + actor = canonicalActor(actor) + store.audits = append(store.audits, AuditRecord{ + ID: store.nextAuditID, RequestID: requestID, Actor: actor, Action: action, + ResourceType: resourceType, ResourceName: resourceName, Changed: changed, + Revision: revision, Reason: reason, OccurredAt: occurredAt.UTC(), + }) +} + +func (store *MemoryStore) appendEventLocked( + revision uint64, + eventType string, + aggregateType string, + aggregateID string, + payload json.RawMessage, + occurredAt time.Time, +) { + store.nextEventID++ + store.events = append(store.events, Event{ + ID: store.nextEventID, Revision: revision, Type: eventType, + AggregateType: aggregateType, AggregateID: aggregateID, + Payload: append(json.RawMessage(nil), payload...), OccurredAt: occurredAt.UTC(), + }) +} + +func (store *MemoryStore) eventIndexLocked(eventID uint64) int { + index := sort.Search(len(store.events), func(index int) bool { + return store.events[index].ID >= eventID + }) + if index >= len(store.events) || store.events[index].ID != eventID { + return -1 + } + return index +} + +func contextError(ctx context.Context) error { + if ctx == nil { + return ErrInvalidCommand + } + return ctx.Err() +} + +func contains(values []string, target string) bool { + for _, value := range values { + if value == target { + return true + } + } + return false +} + +func encodeEventPayload(value any) (json.RawMessage, error) { + payload, err := json.Marshal(value) + if err != nil { + return nil, err + } + return payload, nil +} + +func canonicalActor(actor Actor) Actor { + if actor.SourceIP == "" { + return actor + } + address, err := netip.ParseAddr(actor.SourceIP) + if err == nil { + actor.SourceIP = address.Unmap().String() + } + return actor +}