415 lines
12 KiB
Go
415 lines
12 KiB
Go
package adminstate
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"math"
|
|
"net/netip"
|
|
"sort"
|
|
"strings"
|
|
"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)
|
|
command.Checksum = strings.ToLower(command.Checksum)
|
|
|
|
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
|
|
}
|