From c5311aa9c0e2e55f7b38c7bc048de18fd78c75d0 Mon Sep 17 00:00:00 2001 From: youfak Date: Wed, 29 Jul 2026 21:47:50 +0800 Subject: [PATCH] refactor: share admin state validation --- docs/adr/006-postgresql-admin-state.md | 3 + internal/domain/adminstate/adminstate.go | 18 ++++-- internal/domain/adminstate/memory.go | 12 ++-- internal/domain/adminstate/validation_test.go | 56 +++++++++++-------- progress.md | 4 ++ 5 files changed, 59 insertions(+), 34 deletions(-) diff --git a/docs/adr/006-postgresql-admin-state.md b/docs/adr/006-postgresql-admin-state.md index 6520953..ee93193 100644 --- a/docs/adr/006-postgresql-admin-state.md +++ b/docs/adr/006-postgresql-admin-state.md @@ -51,6 +51,9 @@ Repository。 行为契约。Controller Admin 应用层只负责 DTO 映射、配置解析/校验和运行态发布， 不自行拼装数据库事务。 +所有命令和查询在领域类型上提供公用 `Validate()`；Memory 与 PostgreSQL +Adapter 必须在访问存储前调用同一方法，不复制名称、引用、分页或租约校验。 + ### 全局修订 每次产生管理状态变化时分配单调递增的全局 `revision`： diff --git a/internal/domain/adminstate/adminstate.go b/internal/domain/adminstate/adminstate.go index 1887baf..45f8afe 100644 --- a/internal/domain/adminstate/adminstate.go +++ b/internal/domain/adminstate/adminstate.go @@ -194,7 +194,8 @@ type AcknowledgeCommand struct { EventIDs []uint64 } -func validateSetUpstreamCommand(command SetUpstreamCommand) error { +// Validate checks whether the command is safe for every Store adapter to execute. +func (command SetUpstreamCommand) Validate() error { if err := validateMutationBase(command.RequestID, command.Actor, command.OccurredAt); err != nil || !validIdentifier(command.Name) { return ErrInvalidCommand @@ -202,7 +203,8 @@ func validateSetUpstreamCommand(command SetUpstreamCommand) error { return nil } -func validateSwitchRoutingCommand(command SwitchRoutingCommand) error { +// Validate checks whether the command is safe for every Store adapter to execute. +func (command SwitchRoutingCommand) Validate() error { if err := validateMutationBase(command.RequestID, command.Actor, command.OccurredAt); err != nil || !validIdentifier(command.Name) || !validIdentifier(command.ExpectedCurrent) || !validIdentifier(command.Target) || !validOptionalText(command.Reason, MaxReasonBytes) { @@ -211,7 +213,8 @@ func validateSwitchRoutingCommand(command SwitchRoutingCommand) error { return nil } -func validateCommitConfigCommand(command CommitConfigCommand) error { +// Validate checks the complete management snapshot and all of its references. +func (command CommitConfigCommand) Validate() error { if err := validateMutationBase(command.RequestID, command.Actor, command.OccurredAt); err != nil || !validRequiredText(command.ConfigVersion, MaxIdentifierBytes) || !validSHA256(command.Checksum) || !validRequiredText(command.Source, MaxSourceBytes) || @@ -260,7 +263,8 @@ func validateCommitConfigCommand(command CommitConfigCommand) error { return nil } -func validateClaimCommand(command ClaimCommand) error { +// Validate checks the bounded outbox claim request. +func (command ClaimCommand) Validate() error { if !validIdentifier(command.ConsumerID) || command.Now.IsZero() || command.Limit <= 0 || command.Limit > MaxPageSize || command.Lease <= 0 { return ErrInvalidCommand @@ -268,7 +272,8 @@ func validateClaimCommand(command ClaimCommand) error { return nil } -func validateAcknowledgeCommand(command AcknowledgeCommand) error { +// Validate checks the bounded, duplicate-free outbox acknowledgement request. +func (command AcknowledgeCommand) Validate() error { if !validIdentifier(command.ConsumerID) || command.Now.IsZero() || len(command.EventIDs) == 0 || len(command.EventIDs) > MaxPageSize { return ErrInvalidCommand @@ -286,7 +291,8 @@ func validateAcknowledgeCommand(command AcknowledgeCommand) error { return nil } -func validateAuditQuery(query AuditQuery) error { +// Validate checks the bounded audit pagination request. +func (query AuditQuery) Validate() error { if query.Limit <= 0 || query.Limit > MaxPageSize { return ErrInvalidCommand } diff --git a/internal/domain/adminstate/memory.go b/internal/domain/adminstate/memory.go index 4cb581e..348b06c 100644 --- a/internal/domain/adminstate/memory.go +++ b/internal/domain/adminstate/memory.go @@ -39,7 +39,7 @@ func (store *MemoryStore) CommitConfig(ctx context.Context, command CommitConfig if err := contextError(ctx); err != nil { return result, err } - if store == nil || validateCommitConfigCommand(command) != nil { + if store == nil || command.Validate() != nil { return result, ErrInvalidCommand } command = cloneCommitConfigCommand(command) @@ -109,7 +109,7 @@ func (store *MemoryStore) SetUpstreamEnabled(ctx context.Context, command SetUps if err := contextError(ctx); err != nil { return result, err } - if store == nil || validateSetUpstreamCommand(command) != nil { + if store == nil || command.Validate() != nil { return result, ErrInvalidCommand } @@ -158,7 +158,7 @@ func (store *MemoryStore) SwitchRouting(ctx context.Context, command SwitchRouti if err := contextError(ctx); err != nil { return result, err } - if store == nil || validateSwitchRoutingCommand(command) != nil { + if store == nil || command.Validate() != nil { return result, ErrInvalidCommand } @@ -248,7 +248,7 @@ func (store *MemoryStore) ReadAudit(ctx context.Context, query AuditQuery) ([]Au if err := contextError(ctx); err != nil { return nil, err } - if store == nil || validateAuditQuery(query) != nil { + if store == nil || query.Validate() != nil { return nil, ErrInvalidCommand } store.mu.Lock() @@ -273,7 +273,7 @@ func (store *MemoryStore) Claim(ctx context.Context, command ClaimCommand) ([]Ev if err := contextError(ctx); err != nil { return nil, err } - if store == nil || validateClaimCommand(command) != nil { + if store == nil || command.Validate() != nil { return nil, ErrInvalidCommand } store.mu.Lock() @@ -303,7 +303,7 @@ func (store *MemoryStore) Acknowledge(ctx context.Context, command AcknowledgeCo if err := contextError(ctx); err != nil { return err } - if store == nil || validateAcknowledgeCommand(command) != nil { + if store == nil || command.Validate() != nil { return ErrInvalidCommand } store.mu.Lock() diff --git a/internal/domain/adminstate/validation_test.go b/internal/domain/adminstate/validation_test.go index ee0cb75..694e80d 100644 --- a/internal/domain/adminstate/validation_test.go +++ b/internal/domain/adminstate/validation_test.go @@ -15,8 +15,8 @@ func TestValidateSetUpstreamCommand(t *testing.T) { RequestID: "req-upstream", Actor: Actor{ID: "admin-a", SourceIP: "192.0.2.10"}, OccurredAt: now, Name: "provider-a", Enabled: true, } - if err := validateSetUpstreamCommand(valid); err != nil { - t.Fatalf("validateSetUpstreamCommand(valid): %v", err) + if err := valid.Validate(); err != nil { + t.Fatalf("SetUpstreamCommand.Validate(valid): %v", err) } tests := []struct { @@ -35,8 +35,8 @@ func TestValidateSetUpstreamCommand(t *testing.T) { t.Parallel() command := valid test.mutate(&command) - if err := validateSetUpstreamCommand(command); !errors.Is(err, ErrInvalidCommand) { - t.Fatalf("validateSetUpstreamCommand() error = %v, want ErrInvalidCommand", err) + if err := command.Validate(); !errors.Is(err, ErrInvalidCommand) { + t.Fatalf("SetUpstreamCommand.Validate() error = %v, want ErrInvalidCommand", err) } }) } @@ -49,8 +49,8 @@ func TestValidateSwitchRoutingCommand(t *testing.T) { RequestID: "req-switch", Actor: Actor{ID: "admin-a"}, OccurredAt: now, Name: "checkout", ExpectedCurrent: "provider-a", Target: "provider-b", Reason: "capacity", } - if err := validateSwitchRoutingCommand(valid); err != nil { - t.Fatalf("validateSwitchRoutingCommand(valid): %v", err) + if err := valid.Validate(); err != nil { + t.Fatalf("SwitchRoutingCommand.Validate(valid): %v", err) } for _, test := range []struct { @@ -67,8 +67,8 @@ func TestValidateSwitchRoutingCommand(t *testing.T) { t.Parallel() command := valid test.mutate(&command) - if err := validateSwitchRoutingCommand(command); !errors.Is(err, ErrInvalidCommand) { - t.Fatalf("validateSwitchRoutingCommand() error = %v, want ErrInvalidCommand", err) + if err := command.Validate(); !errors.Is(err, ErrInvalidCommand) { + t.Fatalf("SwitchRoutingCommand.Validate() error = %v, want ErrInvalidCommand", err) } }) } @@ -77,8 +77,8 @@ func TestValidateSwitchRoutingCommand(t *testing.T) { func TestValidateCommitConfigCommandAndReferences(t *testing.T) { t.Parallel() valid := validCommitConfigCommand() - if err := validateCommitConfigCommand(valid); err != nil { - t.Fatalf("validateCommitConfigCommand(valid): %v", err) + if err := valid.Validate(); err != nil { + t.Fatalf("CommitConfigCommand.Validate(valid): %v", err) } for _, test := range []struct { @@ -102,8 +102,8 @@ func TestValidateCommitConfigCommandAndReferences(t *testing.T) { t.Parallel() command := cloneCommitConfigCommand(valid) test.mutate(&command) - if err := validateCommitConfigCommand(command); !errors.Is(err, ErrInvalidCommand) { - t.Fatalf("validateCommitConfigCommand() error = %v, want ErrInvalidCommand", err) + if err := command.Validate(); !errors.Is(err, ErrInvalidCommand) { + t.Fatalf("CommitConfigCommand.Validate() error = %v, want ErrInvalidCommand", err) } }) } @@ -112,15 +112,15 @@ func TestValidateCommitConfigCommandAndReferences(t *testing.T) { func TestValidateOutboxCommands(t *testing.T) { t.Parallel() now := time.Date(2026, 7, 29, 10, 0, 0, 0, time.UTC) - if err := validateClaimCommand(ClaimCommand{ + if err := (ClaimCommand{ ConsumerID: "publisher-a", Now: now, Limit: 100, Lease: time.Minute, - }); err != nil { - t.Fatalf("validateClaimCommand(valid): %v", err) + }).Validate(); err != nil { + t.Fatalf("ClaimCommand.Validate(valid): %v", err) } - if err := validateAcknowledgeCommand(AcknowledgeCommand{ + if err := (AcknowledgeCommand{ ConsumerID: "publisher-a", Now: now, EventIDs: []uint64{1, 2}, - }); err != nil { - t.Fatalf("validateAcknowledgeCommand(valid): %v", err) + }).Validate(); err != nil { + t.Fatalf("AcknowledgeCommand.Validate(valid): %v", err) } invalidClaims := []ClaimCommand{ @@ -131,8 +131,8 @@ func TestValidateOutboxCommands(t *testing.T) { {ConsumerID: "publisher-a", Now: now, Limit: 1}, } for _, command := range invalidClaims { - if err := validateClaimCommand(command); !errors.Is(err, ErrInvalidCommand) { - t.Fatalf("validateClaimCommand(%+v) error = %v", command, err) + if err := command.Validate(); !errors.Is(err, ErrInvalidCommand) { + t.Fatalf("ClaimCommand.Validate(%+v) error = %v", command, err) } } @@ -144,8 +144,20 @@ func TestValidateOutboxCommands(t *testing.T) { {ConsumerID: "publisher-a", Now: now, EventIDs: []uint64{1, 1}}, } for _, command := range invalidAcks { - if err := validateAcknowledgeCommand(command); !errors.Is(err, ErrInvalidCommand) { - t.Fatalf("validateAcknowledgeCommand(%+v) error = %v", command, err) + if err := command.Validate(); !errors.Is(err, ErrInvalidCommand) { + t.Fatalf("AcknowledgeCommand.Validate(%+v) error = %v", command, err) + } + } +} + +func TestAuditQueryValidate(t *testing.T) { + t.Parallel() + if err := (AuditQuery{Limit: 10}).Validate(); err != nil { + t.Fatalf("AuditQuery.Validate(valid): %v", err) + } + for _, query := range []AuditQuery{{}, {Limit: -1}, {Limit: MaxPageSize + 1}} { + if err := query.Validate(); !errors.Is(err, ErrInvalidCommand) { + t.Fatalf("AuditQuery.Validate(%+v) error = %v, want ErrInvalidCommand", query, err) } } } diff --git a/progress.md b/progress.md index 859c45a..4f0221b 100644 --- a/progress.md +++ b/progress.md @@ -2,6 +2,10 @@ ## 2026-07-29 +- `adminstate` 六类命令/查询已封装公用 `Validate()`，MemoryStore 改为统一复用； + 后续 pgx Adapter 不再重复实现名称、引用、分页和 Outbox 租约输入校验。 +- 一次命令装配审计误用了不存在的 `controller/distribution/service.go`；实际领域 + 服务位于 `controller/extraction`，Distribution 包当前只负责 HTTP Handler。 - 根据对话最终定稿统一 Sequential：至少两个 Upstream，省略 `endBehavior` 时 默认 `stop`；新增领域和严格配置回归测试。disabled candidate、跨实例游标和 `onUnavailable` 请求链仍未提前标记完成。