refactor(projectors): single-source decodePayload[T] + DRY completeness guard (#438)

WS16b DRY-discipline. The ~90 *FromEvent decoders in internal/projectors/ each
hand-rolled the identical decode shape: a (stream_type, event_type) guard
returning ErrIgnoredEvent, an empty-payload guard, json.Unmarshal(e.Data, &p),
and the canonical "projector: invalid <event> payload: %w" wrap. Collapse the
boilerplate into one generic decodePayload[T] helper; per-event field
validation stays in the caller. 40 decoders converted (-305 LOC net).

TestDecodePayloadHelperUsedByAllProjectors AST-walks the package and fails the
build if any *FromEvent hand-rolls json.Unmarshal without routing through the
helper, unless it is a recorded exception (empty-payload-valid, multi-stream /
runtime-eventType, or non-canonical DB-params shape). The allowlist is guarded
against stale entries and a vacuous walk. Decoders that route through a shared
per-cluster sub-decoder carry no direct json.Unmarshal and are neither flagged
nor allowlisted.

ADR 0021 records the single-source-helper + self-discovering-test DRY policy and
the decision to defer the noisy `dupl` CI backstop in favor of specific
completeness tests.

Closes #437

Author: PaulDotterer

docs/adr/0021-single-source-helpers-dry.md

@@ -0,0 +1,69 @@
+# 0021 — Single-source helpers + DRY discipline
+
+- Status: accepted
+- Date: 2026-06-15
+- Related: WS16b of the SECURITY_HARDENING_WORKPLAN
+  (manchtools/power-manage-server#437; the cert/CSR fixture row lands in the
+  sdk + agent repos); ADR 0020 (the WS16 fail-closed sweep, whose silent-decode
+  fixes motivated centralizing the projector decode).
+
+## Context
+
+DRY in this codebase is enforced *as code is written* — "extract before you
+duplicate" is the Definition of Done, and each feature stream collapses the
+duplication clusters in the files it touches, in the same PR. WS16b owns only
+the residual clusters no single feature stream spans:
+
+1. **Projector payload decode.** ~90 `*FromEvent` decoders in
+   `internal/projectors/` each hand-rolled the identical shape: a
+   `(stream_type, event_type)` guard returning `ErrIgnoredEvent`, an
+   empty-payload guard, a `json.Unmarshal(e.Data, &p)`, and the canonical
+   `"projector: invalid <event> payload: %w"` error wrap. No feature stream
+   touches all projectors, so the duplication never collapsed inline.
+2. **Cert/CSR test fixtures.** ECDSA P-256 CA/cert generation was re-implemented
+   in sdk and agent test files (device/user fixtures were already centralized in
+   `server/internal/testutil`).
+
+## Decision
+
+- **The strongest DRY mechanism is a single-source helper plus a
+  self-discovering completeness test, not a generic similarity tool.** Each
+  cluster collapses to one helper that everything consumes, and an AST guard
+  makes the *specific* duplication structurally impossible to reintroduce.
+
+- **`decodePayload[T]` (server).** One generic helper centralizes the
+  projector decode boilerplate; per-event field validation stays in the caller.
+  `TestDecodePayloadHelperUsedByAllProjectors` AST-walks the package and fails
+  the build if any `*FromEvent` hand-rolls `json.Unmarshal` without routing
+  through the helper, unless it is a recorded exception. The exception
+  allowlist is guarded against stale entries and against a vacuous (zero-match)
+  walk. The legitimate exceptions are decoders whose payload is empty-valid
+  (the event carries no body), whose guard spans two stream types or a runtime
+  event-type parameter, or that build a DB-params struct keyed off the
+  envelope — shapes the fixed-argument helper cannot express. Decoders that
+  route through a shared per-cluster sub-decoder carry no direct
+  `json.Unmarshal` and are neither flagged nor allowlisted.
+
+- **Shared cert/CSR test fixtures (sdk-first).** The CA/cert builders move to a
+  shared exported helper in the sdk so agent and sdk tests consume one
+  implementation.
+
+- **The `dupl` CI backstop is deferred, not adopted.** The work plan specified a
+  `dupl` duplication-detection CI job as a *secondary, allowlisted* backstop,
+  explicitly blunt on legitimately-similar code. The primary mechanism — the
+  single-source helper + self-discovering completeness test above — is stronger
+  and noise-free, so we ship that and do not add the `dupl` job (it would
+  generate false positives on the many structurally-similar projector/handler
+  functions). If a future cluster cannot be guarded by a specific completeness
+  test, the `dupl` job is the documented fallback.
+
+## Consequences
+
+- A new projector decoder cannot hand-roll the decode — it must use
+  `decodePayload[T]` or justify an allowlist entry under review. The projector
+  decode boilerplate shrank by ~300 lines.
+- Centralizing the decode also centralizes the error-message form, so a future
+  change to the projector decode contract happens in one place.
+- The exception allowlist is the honest record of the shapes the helper does
+  not cover; the stale-entry guard keeps it shrinking as those shapes are
+  refactored.

internal/projectors/action.go

@@ -53,15 +53,9 @@ type ActionCreatedPayload struct {
 // for any other (stream, event_type) so the listener wrapper can
 // silently no-op.
 func ActionCreatedFromEvent(e store.PersistedEvent) (ActionCreatedPayload, error) {
-	if e.StreamType != "action" || e.EventType != string(eventtypes.ActionCreated) {
-		return ActionCreatedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return ActionCreatedPayload{}, fmt.Errorf("projector: empty ActionCreated payload")
-	}
-	var raw payloads.ActionCreated
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return ActionCreatedPayload{}, fmt.Errorf("projector: invalid ActionCreated payload: %w", err)
+	raw, err := decodePayload[payloads.ActionCreated](e, "action", eventtypes.ActionCreated)
+	if err != nil {
+		return ActionCreatedPayload{}, err
 	}
 	if raw.Name == "" {
 		return ActionCreatedPayload{}, fmt.Errorf("projector: ActionCreated requires name")
@@ -108,15 +102,9 @@ type ActionRenamedPayload struct {
 
 // ActionRenamedFromEvent decodes ActionRenamed.
 func ActionRenamedFromEvent(e store.PersistedEvent) (ActionRenamedPayload, error) {
-	if e.StreamType != "action" || e.EventType != string(eventtypes.ActionRenamed) {
-		return ActionRenamedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return ActionRenamedPayload{}, fmt.Errorf("projector: empty ActionRenamed payload")
-	}
-	var raw payloads.ActionRenamed
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return ActionRenamedPayload{}, fmt.Errorf("projector: invalid ActionRenamed payload: %w", err)
+	raw, err := decodePayload[payloads.ActionRenamed](e, "action", eventtypes.ActionRenamed)
+	if err != nil {
+		return ActionRenamedPayload{}, err
 	}
 	if raw.Name == "" {
 		return ActionRenamedPayload{}, fmt.Errorf("projector: ActionRenamed requires name")

internal/projectors/action_set.go

@@ -41,15 +41,9 @@ type actionSetCreatedRaw struct {
 // ErrIgnoredEvent for any other (stream, event_type) so the listener
 // wrapper can silently no-op.
 func ActionSetCreatedFromEvent(e store.PersistedEvent) (ActionSetCreatedPayload, error) {
-	if e.StreamType != "action_set" || e.EventType != string(eventtypes.ActionSetCreated) {
-		return ActionSetCreatedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return ActionSetCreatedPayload{}, fmt.Errorf("projector: empty ActionSetCreated payload")
-	}
-	var raw actionSetCreatedRaw
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return ActionSetCreatedPayload{}, fmt.Errorf("projector: invalid ActionSetCreated payload: %w", err)
+	raw, err := decodePayload[actionSetCreatedRaw](e, "action_set", eventtypes.ActionSetCreated)
+	if err != nil {
+		return ActionSetCreatedPayload{}, err
 	}
 	if raw.Name == "" {
 		return ActionSetCreatedPayload{}, fmt.Errorf("projector: ActionSetCreated requires name")
@@ -89,15 +83,9 @@ type actionSetRenamedRaw struct {
 
 // ActionSetRenamedFromEvent decodes ActionSetRenamed.
 func ActionSetRenamedFromEvent(e store.PersistedEvent) (ActionSetRenamedPayload, error) {
-	if e.StreamType != "action_set" || e.EventType != string(eventtypes.ActionSetRenamed) {
-		return ActionSetRenamedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return ActionSetRenamedPayload{}, fmt.Errorf("projector: empty ActionSetRenamed payload")
-	}
-	var raw actionSetRenamedRaw
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return ActionSetRenamedPayload{}, fmt.Errorf("projector: invalid ActionSetRenamed payload: %w", err)
+	raw, err := decodePayload[actionSetRenamedRaw](e, "action_set", eventtypes.ActionSetRenamed)
+	if err != nil {
+		return ActionSetRenamedPayload{}, err
 	}
 	if raw.Name == "" {
 		return ActionSetRenamedPayload{}, fmt.Errorf("projector: ActionSetRenamed requires name")
@@ -190,15 +178,9 @@ type actionSetMemberRaw struct {
 
 // ActionSetMemberAddedFromEvent decodes ActionSetMemberAdded.
 func ActionSetMemberAddedFromEvent(e store.PersistedEvent) (ActionSetMemberAddedPayload, error) {
-	if e.StreamType != "action_set" || e.EventType != string(eventtypes.ActionSetMemberAdded) {
-		return ActionSetMemberAddedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return ActionSetMemberAddedPayload{}, fmt.Errorf("projector: empty ActionSetMemberAdded payload")
-	}
-	var raw actionSetMemberRaw
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return ActionSetMemberAddedPayload{}, fmt.Errorf("projector: invalid ActionSetMemberAdded payload: %w", err)
+	raw, err := decodePayload[actionSetMemberRaw](e, "action_set", eventtypes.ActionSetMemberAdded)
+	if err != nil {
+		return ActionSetMemberAddedPayload{}, err
 	}
 	if raw.ActionID == "" {
 		return ActionSetMemberAddedPayload{}, fmt.Errorf("projector: ActionSetMemberAdded requires action_id")
@@ -223,15 +205,9 @@ type actionSetMemberRemovedRaw struct {
 
 // ActionSetMemberRemovedFromEvent decodes ActionSetMemberRemoved.
 func ActionSetMemberRemovedFromEvent(e store.PersistedEvent) (ActionSetMemberRemovedPayload, error) {
-	if e.StreamType != "action_set" || e.EventType != string(eventtypes.ActionSetMemberRemoved) {
-		return ActionSetMemberRemovedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return ActionSetMemberRemovedPayload{}, fmt.Errorf("projector: empty ActionSetMemberRemoved payload")
-	}
-	var raw actionSetMemberRemovedRaw
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return ActionSetMemberRemovedPayload{}, fmt.Errorf("projector: invalid ActionSetMemberRemoved payload: %w", err)
+	raw, err := decodePayload[actionSetMemberRemovedRaw](e, "action_set", eventtypes.ActionSetMemberRemoved)
+	if err != nil {
+		return ActionSetMemberRemovedPayload{}, err
 	}
 	if raw.ActionID == "" {
 		return ActionSetMemberRemovedPayload{}, fmt.Errorf("projector: ActionSetMemberRemoved requires action_id")
@@ -247,15 +223,9 @@ type ActionSetMemberReorderedPayload = ActionSetMemberAddedPayload
 // Same field set as ActionSetMemberAdded plus the same default-zero
 // behaviour for sort_order.
 func ActionSetMemberReorderedFromEvent(e store.PersistedEvent) (ActionSetMemberReorderedPayload, error) {
-	if e.StreamType != "action_set" || e.EventType != string(eventtypes.ActionSetMemberReordered) {
-		return ActionSetMemberReorderedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return ActionSetMemberReorderedPayload{}, fmt.Errorf("projector: empty ActionSetMemberReordered payload")
-	}
-	var raw actionSetMemberRaw
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return ActionSetMemberReorderedPayload{}, fmt.Errorf("projector: invalid ActionSetMemberReordered payload: %w", err)
+	raw, err := decodePayload[actionSetMemberRaw](e, "action_set", eventtypes.ActionSetMemberReordered)
+	if err != nil {
+		return ActionSetMemberReorderedPayload{}, err
 	}
 	if raw.ActionID == "" {
 		return ActionSetMemberReorderedPayload{}, fmt.Errorf("projector: ActionSetMemberReordered requires action_id")

internal/projectors/assignment.go

@@ -40,15 +40,9 @@ type AssignmentCreatedPayload struct {
 // failure surface in the listener log instead of producing a half-
 // applied row.
 func AssignmentCreatedFromEvent(e store.PersistedEvent) (AssignmentCreatedPayload, error) {
-	if e.StreamType != "assignment" || e.EventType != string(eventtypes.AssignmentCreated) {
-		return AssignmentCreatedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return AssignmentCreatedPayload{}, fmt.Errorf("projector: empty AssignmentCreated payload")
-	}
-	var raw payloads.AssignmentCreated
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return AssignmentCreatedPayload{}, fmt.Errorf("projector: invalid AssignmentCreated payload: %w", err)
+	raw, err := decodePayload[payloads.AssignmentCreated](e, "assignment", eventtypes.AssignmentCreated)
+	if err != nil {
+		return AssignmentCreatedPayload{}, err
 	}
 	switch {
 	case raw.SourceType == "":

internal/projectors/compliance.go

@@ -47,15 +47,9 @@ type complianceResultUpdatedRaw struct {
 // Returns ErrIgnoredEvent for any other (stream, event_type) so the
 // listener wrapper can silently no-op.
 func ComplianceResultUpdatedFromEvent(e store.PersistedEvent) (ComplianceResultUpdatedPayload, error) {
-	if e.StreamType != "compliance" || e.EventType != string(eventtypes.ComplianceResultUpdated) {
-		return ComplianceResultUpdatedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return ComplianceResultUpdatedPayload{}, fmt.Errorf("projector: empty ComplianceResultUpdated payload")
-	}
-	var raw complianceResultUpdatedRaw
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return ComplianceResultUpdatedPayload{}, fmt.Errorf("projector: invalid ComplianceResultUpdated payload: %w", err)
+	raw, err := decodePayload[complianceResultUpdatedRaw](e, "compliance", eventtypes.ComplianceResultUpdated)
+	if err != nil {
+		return ComplianceResultUpdatedPayload{}, err
 	}
 	if raw.DeviceID == "" {
 		return ComplianceResultUpdatedPayload{}, fmt.Errorf("projector: ComplianceResultUpdated requires device_id")
@@ -95,15 +89,9 @@ type complianceResultRemovedRaw struct {
 
 // ComplianceResultRemovedFromEvent decodes ComplianceResultRemoved.
 func ComplianceResultRemovedFromEvent(e store.PersistedEvent) (ComplianceResultRemovedPayload, error) {
-	if e.StreamType != "compliance" || e.EventType != string(eventtypes.ComplianceResultRemoved) {
-		return ComplianceResultRemovedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return ComplianceResultRemovedPayload{}, fmt.Errorf("projector: empty ComplianceResultRemoved payload")
-	}
-	var raw complianceResultRemovedRaw
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return ComplianceResultRemovedPayload{}, fmt.Errorf("projector: invalid ComplianceResultRemoved payload: %w", err)
+	raw, err := decodePayload[complianceResultRemovedRaw](e, "compliance", eventtypes.ComplianceResultRemoved)
+	if err != nil {
+		return ComplianceResultRemovedPayload{}, err
 	}
 	if raw.DeviceID == "" {
 		return ComplianceResultRemovedPayload{}, fmt.Errorf("projector: ComplianceResultRemoved requires device_id")

internal/projectors/compliance_policy.go

@@ -34,15 +34,9 @@ type compliancePolicyCreatedRaw struct {
 // Returns ErrIgnoredEvent for any other (stream, event_type) so the
 // listener wrapper can silently no-op.
 func CompliancePolicyCreatedFromEvent(e store.PersistedEvent) (CompliancePolicyCreatedPayload, error) {
-	if e.StreamType != "compliance_policy" || e.EventType != string(eventtypes.CompliancePolicyCreated) {
-		return CompliancePolicyCreatedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return CompliancePolicyCreatedPayload{}, fmt.Errorf("projector: empty CompliancePolicyCreated payload")
-	}
-	var raw compliancePolicyCreatedRaw
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return CompliancePolicyCreatedPayload{}, fmt.Errorf("projector: invalid CompliancePolicyCreated payload: %w", err)
+	raw, err := decodePayload[compliancePolicyCreatedRaw](e, "compliance_policy", eventtypes.CompliancePolicyCreated)
+	if err != nil {
+		return CompliancePolicyCreatedPayload{}, err
 	}
 	if raw.Name == "" {
 		return CompliancePolicyCreatedPayload{}, fmt.Errorf("projector: CompliancePolicyCreated requires name")
@@ -73,15 +67,9 @@ type compliancePolicyRenamedRaw struct {
 
 // CompliancePolicyRenamedFromEvent decodes CompliancePolicyRenamed.
 func CompliancePolicyRenamedFromEvent(e store.PersistedEvent) (CompliancePolicyRenamedPayload, error) {
-	if e.StreamType != "compliance_policy" || e.EventType != string(eventtypes.CompliancePolicyRenamed) {
-		return CompliancePolicyRenamedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return CompliancePolicyRenamedPayload{}, fmt.Errorf("projector: empty CompliancePolicyRenamed payload")
-	}
-	var raw compliancePolicyRenamedRaw
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return CompliancePolicyRenamedPayload{}, fmt.Errorf("projector: invalid CompliancePolicyRenamed payload: %w", err)
+	raw, err := decodePayload[compliancePolicyRenamedRaw](e, "compliance_policy", eventtypes.CompliancePolicyRenamed)
+	if err != nil {
+		return CompliancePolicyRenamedPayload{}, err
 	}
 	if raw.Name == "" {
 		return CompliancePolicyRenamedPayload{}, fmt.Errorf("projector: CompliancePolicyRenamed requires name")
@@ -148,15 +136,9 @@ type compliancePolicyRuleAddedRaw struct {
 
 // CompliancePolicyRuleAddedFromEvent decodes CompliancePolicyRuleAdded.
 func CompliancePolicyRuleAddedFromEvent(e store.PersistedEvent) (CompliancePolicyRuleAddedPayload, error) {
-	if e.StreamType != "compliance_policy" || e.EventType != string(eventtypes.CompliancePolicyRuleAdded) {
-		return CompliancePolicyRuleAddedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return CompliancePolicyRuleAddedPayload{}, fmt.Errorf("projector: empty CompliancePolicyRuleAdded payload")
-	}
-	var raw compliancePolicyRuleAddedRaw
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return CompliancePolicyRuleAddedPayload{}, fmt.Errorf("projector: invalid CompliancePolicyRuleAdded payload: %w", err)
+	raw, err := decodePayload[compliancePolicyRuleAddedRaw](e, "compliance_policy", eventtypes.CompliancePolicyRuleAdded)
+	if err != nil {
+		return CompliancePolicyRuleAddedPayload{}, err
 	}
 	if raw.ActionID == "" {
 		return CompliancePolicyRuleAddedPayload{}, fmt.Errorf("projector: CompliancePolicyRuleAdded requires action_id")
@@ -188,15 +170,9 @@ type compliancePolicyRuleRemovedRaw struct {
 
 // CompliancePolicyRuleRemovedFromEvent decodes CompliancePolicyRuleRemoved.
 func CompliancePolicyRuleRemovedFromEvent(e store.PersistedEvent) (CompliancePolicyRuleRemovedPayload, error) {
-	if e.StreamType != "compliance_policy" || e.EventType != string(eventtypes.CompliancePolicyRuleRemoved) {
-		return CompliancePolicyRuleRemovedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return CompliancePolicyRuleRemovedPayload{}, fmt.Errorf("projector: empty CompliancePolicyRuleRemoved payload")
-	}
-	var raw compliancePolicyRuleRemovedRaw
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return CompliancePolicyRuleRemovedPayload{}, fmt.Errorf("projector: invalid CompliancePolicyRuleRemoved payload: %w", err)
+	raw, err := decodePayload[compliancePolicyRuleRemovedRaw](e, "compliance_policy", eventtypes.CompliancePolicyRuleRemoved)
+	if err != nil {
+		return CompliancePolicyRuleRemovedPayload{}, err
 	}
 	if raw.ActionID == "" {
 		return CompliancePolicyRuleRemovedPayload{}, fmt.Errorf("projector: CompliancePolicyRuleRemoved requires action_id")
@@ -226,15 +202,9 @@ type compliancePolicyRuleUpdatedRaw struct {
 
 // CompliancePolicyRuleUpdatedFromEvent decodes CompliancePolicyRuleUpdated.
 func CompliancePolicyRuleUpdatedFromEvent(e store.PersistedEvent) (CompliancePolicyRuleUpdatedPayload, error) {
-	if e.StreamType != "compliance_policy" || e.EventType != string(eventtypes.CompliancePolicyRuleUpdated) {
-		return CompliancePolicyRuleUpdatedPayload{}, ErrIgnoredEvent
-	}
-	if len(e.Data) == 0 {
-		return CompliancePolicyRuleUpdatedPayload{}, fmt.Errorf("projector: empty CompliancePolicyRuleUpdated payload")
-	}
-	var raw compliancePolicyRuleUpdatedRaw
-	if err := json.Unmarshal(e.Data, &raw); err != nil {
-		return CompliancePolicyRuleUpdatedPayload{}, fmt.Errorf("projector: invalid CompliancePolicyRuleUpdated payload: %w", err)
+	raw, err := decodePayload[compliancePolicyRuleUpdatedRaw](e, "compliance_policy", eventtypes.CompliancePolicyRuleUpdated)
+	if err != nil {
+		return CompliancePolicyRuleUpdatedPayload{}, err
 	}
 	if raw.ActionID == "" {
 		return CompliancePolicyRuleUpdatedPayload{}, fmt.Errorf("projector: CompliancePolicyRuleUpdated requires action_id")