diff --git a/internal/api/certificate_handler.go b/internal/api/certificate_handler.go index 859ce0dd..a76be8b5 100644 --- a/internal/api/certificate_handler.go +++ b/internal/api/certificate_handler.go @@ -12,6 +12,7 @@ import ( pm "github.com/manchtools/power-manage/sdk/gen/go/pm/v1" "github.com/manchtools/power-manage/server/internal/ca" "github.com/manchtools/power-manage/server/internal/eventtypes" + "github.com/manchtools/power-manage/server/internal/eventtypes/payloads" "github.com/manchtools/power-manage/server/internal/store" db "github.com/manchtools/power-manage/server/internal/store/generated" ) @@ -82,13 +83,15 @@ func (h *CertificateHandler) RenewCertificate(ctx context.Context, req *connect. } // Emit DeviceCertRenewed event (projection handler already exists in migration 001) + fingerprint := newCert.Fingerprint + notAfterStr := newCert.NotAfter.Format(time.RFC3339) if err := h.store.AppendEvent(ctx, store.Event{ StreamType: "device", StreamID: deviceID, EventType: string(eventtypes.DeviceCertRenewed), - Data: map[string]any{ - "cert_fingerprint": newCert.Fingerprint, - "cert_not_after": newCert.NotAfter.Format(time.RFC3339), + Data: payloads.DeviceCertRenewed{ + CertFingerprint: &fingerprint, + CertNotAfter: ¬AfterStr, }, ActorType: "device", ActorID: deviceID, diff --git a/internal/api/device_handler.go b/internal/api/device_handler.go index 102c0bb6..5c6f41a4 100644 --- a/internal/api/device_handler.go +++ b/internal/api/device_handler.go @@ -22,6 +22,7 @@ import ( "github.com/manchtools/power-manage/server/internal/auth" "github.com/manchtools/power-manage/server/internal/crypto" "github.com/manchtools/power-manage/server/internal/eventtypes" + "github.com/manchtools/power-manage/server/internal/eventtypes/payloads" "github.com/manchtools/power-manage/server/internal/middleware" "github.com/manchtools/power-manage/server/internal/search" "github.com/manchtools/power-manage/server/internal/store" @@ -215,9 +216,9 @@ func (h *DeviceHandler) SetDeviceLabel(ctx context.Context, req *connect.Request StreamType: "device", StreamID: req.Msg.Id, EventType: string(eventtypes.DeviceLabelSet), - Data: map[string]any{ - "key": req.Msg.Key, - "value": req.Msg.Value, + Data: payloads.DeviceLabelSet{ + Key: &req.Msg.Key, + Value: &req.Msg.Value, }, ActorType: "user", ActorID: userCtx.ID, @@ -260,8 +261,8 @@ func (h *DeviceHandler) RemoveDeviceLabel(ctx context.Context, req *connect.Requ StreamType: "device", StreamID: req.Msg.Id, EventType: string(eventtypes.DeviceLabelRemoved), - Data: map[string]any{ - "key": req.Msg.Key, + Data: payloads.DeviceLabelRemoved{ + Key: &req.Msg.Key, }, ActorType: "user", ActorID: userCtx.ID, @@ -375,12 +376,13 @@ func (h *DeviceHandler) AssignDevice(ctx context.Context, req *connect.Request[p return nil, apiErrorCtx(ctx, ErrInternal, connect.CodeInternal, "failed to get user") } + uid := userID if err := appendEvent(ctx, h.store, h.logger, store.Event{ StreamType: "device", StreamID: req.Msg.DeviceId, EventType: string(eventtypes.DeviceAssigned), - Data: map[string]any{ - "user_id": userID, + Data: payloads.DeviceUserAssignment{ + UserID: &uid, }, ActorType: "user", ActorID: userCtx.ID, @@ -403,12 +405,13 @@ func (h *DeviceHandler) AssignDevice(ctx context.Context, req *connect.Request[p return nil, apiErrorCtx(ctx, ErrInternal, connect.CodeInternal, "failed to get user group") } + gid := groupID if err := appendEvent(ctx, h.store, h.logger, store.Event{ StreamType: "device", StreamID: req.Msg.DeviceId, EventType: string(eventtypes.DeviceGroupAssigned), - Data: map[string]any{ - "group_id": groupID, + Data: payloads.DeviceGroupAssignment{ + GroupID: &gid, }, ActorType: "user", ActorID: userCtx.ID, @@ -480,8 +483,8 @@ func (h *DeviceHandler) UnassignDevice(ctx context.Context, req *connect.Request StreamType: "device", StreamID: req.Msg.DeviceId, EventType: string(eventtypes.DeviceUnassigned), - Data: map[string]any{ - "user_id": req.Msg.UserId, + Data: payloads.DeviceUserAssignment{ + UserID: &req.Msg.UserId, }, ActorType: "user", ActorID: userCtx.ID, @@ -494,8 +497,8 @@ func (h *DeviceHandler) UnassignDevice(ctx context.Context, req *connect.Request StreamType: "device", StreamID: req.Msg.DeviceId, EventType: string(eventtypes.DeviceGroupUnassigned), - Data: map[string]any{ - "group_id": req.Msg.GroupId, + Data: payloads.DeviceGroupAssignment{ + GroupID: &req.Msg.GroupId, }, ActorType: "user", ActorID: userCtx.ID, @@ -533,12 +536,13 @@ func (h *DeviceHandler) SetDeviceSyncInterval(ctx context.Context, req *connect. } // Emit DeviceSyncIntervalSet event + syncInterval := req.Msg.SyncIntervalMinutes if err := appendEvent(ctx, h.store, h.logger, store.Event{ StreamType: "device", StreamID: req.Msg.Id, EventType: string(eventtypes.DeviceSyncIntervalSet), - Data: map[string]any{ - "sync_interval_minutes": req.Msg.SyncIntervalMinutes, + Data: payloads.DeviceSyncIntervalSet{ + SyncIntervalMinutes: &syncInterval, }, ActorType: "user", ActorID: userCtx.ID, diff --git a/internal/api/registration_handler.go b/internal/api/registration_handler.go index 74810c24..c9a7a8f6 100644 --- a/internal/api/registration_handler.go +++ b/internal/api/registration_handler.go @@ -17,6 +17,7 @@ import ( pm "github.com/manchtools/power-manage/sdk/gen/go/pm/v1" "github.com/manchtools/power-manage/server/internal/ca" "github.com/manchtools/power-manage/server/internal/eventtypes" + "github.com/manchtools/power-manage/server/internal/eventtypes/payloads" "github.com/manchtools/power-manage/server/internal/store" ) @@ -193,21 +194,36 @@ func (h *RegistrationHandler) Register(ctx context.Context, req *connect.Request return nil, apiErrorCtx(ctx, ErrInternal, connect.CodeInternal, "failed to issue certificate") } - // Build event data for device registration - eventData := map[string]any{ - "hostname": req.Msg.Hostname, - "agent_version": req.Msg.AgentVersion, - "cert_fingerprint": cert.Fingerprint, - "cert_not_after": cert.NotAfter.Format(time.RFC3339), - "registration_token_id": token.ID, - "cert_pem": string(cert.CertPEM), - "ca_cert_pem": string(h.ca.CACertPEM()), + // Build event data for device registration. CertPEM + CACertPEM + // ride along so future replays can recover the cert bytes from + // the event log; the projector ignores them. + hostname := req.Msg.Hostname + agentVersion := req.Msg.AgentVersion + certFingerprint := cert.Fingerprint + // Preserve the legacy RFC 3339-string serialisation (no + // sub-second precision) so wire bytes stay identical to the + // pre-typed-payload emission. The projector parses the string + // back into a time.Time via parseOptionalRFC3339, which accepts + // both RFC 3339 and RFC 3339Nano. + certNotAfterStr := cert.NotAfter.Format(time.RFC3339) + registrationTokenID := token.ID + certPEM := string(cert.CertPEM) + caCertPEM := string(h.ca.CACertPEM()) + deviceData := payloads.DeviceRegistered{ + Hostname: &hostname, + AgentVersion: &agentVersion, + CertFingerprint: &certFingerprint, + CertNotAfter: &certNotAfterStr, + RegistrationTokenID: ®istrationTokenID, + CertPEM: &certPEM, + CACertPEM: &caCertPEM, } // Auto-assign device to token owner if the token has an owner if token.OwnerID != nil && *token.OwnerID != "" { - eventData["assigned_user_id"] = *token.OwnerID - logger.Info("auto-assigning device to token owner", "owner_id", *token.OwnerID) + ownerID := *token.OwnerID + deviceData.AssignedUserID = &ownerID + logger.Info("auto-assigning device to token owner", "owner_id", ownerID) } // Consume the token FIRST to prevent race conditions with one-time tokens. @@ -238,7 +254,7 @@ func (h *RegistrationHandler) Register(ctx context.Context, req *connect.Request StreamType: "device", StreamID: deviceID, EventType: string(eventtypes.DeviceRegistered), - Data: eventData, + Data: deviceData, ActorType: "system", ActorID: "registration", }); err != nil { diff --git a/internal/api/system_actions.go b/internal/api/system_actions.go index d3a93729..24adafd1 100644 --- a/internal/api/system_actions.go +++ b/internal/api/system_actions.go @@ -12,6 +12,7 @@ import ( "github.com/manchtools/power-manage/server/internal/actionparams" "github.com/manchtools/power-manage/server/internal/ca" "github.com/manchtools/power-manage/server/internal/eventtypes" + "github.com/manchtools/power-manage/server/internal/eventtypes/payloads" "github.com/manchtools/power-manage/server/internal/store" db "github.com/manchtools/power-manage/server/internal/store/generated" ) @@ -529,18 +530,19 @@ func (m *SystemActionManager) createSystemAction(ctx context.Context, name strin // assignActionToUser emits an AssignmentCreated event. func (m *SystemActionManager) assignActionToUser(ctx context.Context, actionID, userID string) error { assignmentID := newULID() - + mode := int32(0) // REQUIRED + sortOrder := int32(0) return m.store.AppendEvent(ctx, store.Event{ StreamType: "assignment", StreamID: assignmentID, EventType: string(eventtypes.AssignmentCreated), - Data: map[string]any{ - "source_type": "action", - "source_id": actionID, - "target_type": "user", - "target_id": userID, - "mode": 0, // REQUIRED - "sort_order": 0, + Data: payloads.AssignmentCreated{ + SourceType: "action", + SourceID: actionID, + TargetType: "user", + TargetID: userID, + Mode: &mode, + SortOrder: &sortOrder, }, ActorType: "system", ActorID: "system", @@ -586,9 +588,9 @@ func (m *SystemActionManager) linkSystemAction(ctx context.Context, userID, fiel StreamType: "user", StreamID: userID, EventType: string(eventtypes.UserSystemActionLinked), - Data: map[string]any{ - "field": field, - "action_id": actionID, + Data: payloads.UserSystemActionLinked{ + Field: &field, + ActionID: &actionID, }, ActorType: "system", ActorID: "system", diff --git a/internal/api/user_handler.go b/internal/api/user_handler.go index c4ac55b2..5a34d8f6 100644 --- a/internal/api/user_handler.go +++ b/internal/api/user_handler.go @@ -15,6 +15,7 @@ import ( pm "github.com/manchtools/power-manage/sdk/gen/go/pm/v1" "github.com/manchtools/power-manage/server/internal/auth" "github.com/manchtools/power-manage/server/internal/eventtypes" + "github.com/manchtools/power-manage/server/internal/eventtypes/payloads" "github.com/manchtools/power-manage/server/internal/search" "github.com/manchtools/power-manage/server/internal/store" db "github.com/manchtools/power-manage/server/internal/store/generated" @@ -130,21 +131,22 @@ func (h *UserHandler) CreateUser(ctx context.Context, req *connect.Request[pm.Cr // projector tx. The pre-#135 partial-write window between the // user row INSERT and the per-role INSERTs is no longer // reachable: either both land or neither does. + defaultRole := "user" err = h.store.AppendEvent(ctx, store.Event{ StreamType: "user", StreamID: id, EventType: string(eventtypes.UserCreatedWithRoles), - Data: map[string]any{ - "email": req.Msg.Email, - "password_hash": passwordHash, - "role": "user", - "display_name": req.Msg.DisplayName, - "given_name": req.Msg.GivenName, - "family_name": req.Msg.FamilyName, - "preferred_username": req.Msg.PreferredUsername, - "linux_username": linuxUsername, - "linux_uid": linuxUID, - "role_ids": roleIDs, + Data: payloads.UserCreatedWithRoles{ + Email: &req.Msg.Email, + PasswordHash: &passwordHash, + Role: &defaultRole, + DisplayName: &req.Msg.DisplayName, + GivenName: &req.Msg.GivenName, + FamilyName: &req.Msg.FamilyName, + PreferredUsername: &req.Msg.PreferredUsername, + LinuxUsername: &linuxUsername, + LinuxUID: &linuxUID, + RoleIDs: roleIDs, }, ActorType: "user", ActorID: userCtx.ID, @@ -171,26 +173,30 @@ func (h *UserHandler) CreateUser(ctx context.Context, req *connect.Request[pm.Cr // Auto-enable provisioning/SSH if global server settings are on if settings, err := h.store.Queries().GetServerSettings(ctx); err == nil { if settings.UserProvisioningEnabled { + provisioningEnabled := true if err := h.store.AppendEvent(ctx, store.Event{ StreamType: "user", StreamID: id, EventType: string(eventtypes.UserProvisioningSettingsUpdated), - Data: map[string]any{"user_provisioning_enabled": true}, - ActorType: "system", - ActorID: "auto", + Data: payloads.UserProvisioningSettingsUpdated{ + UserProvisioningEnabled: &provisioningEnabled, + }, + ActorType: "system", + ActorID: "auto", }); err != nil { h.logger.Warn("failed to auto-enable provisioning for new user", "user_id", id, "error", err) } } if settings.SshAccessForAll { + sshOn, sshAllow, sshPwOff := true, true, false if err := h.store.AppendEvent(ctx, store.Event{ StreamType: "user", StreamID: id, EventType: string(eventtypes.UserSshSettingsUpdated), - Data: map[string]any{ - "ssh_access_enabled": true, - "ssh_allow_pubkey": true, - "ssh_allow_password": false, + Data: payloads.UserSshSettingsUpdated{ + SshAccessEnabled: &sshOn, + SshAllowPubkey: &sshAllow, + SshAllowPassword: &sshPwOff, }, ActorType: "system", ActorID: "auto", @@ -329,8 +335,8 @@ func (h *UserHandler) UpdateUserEmail(ctx context.Context, req *connect.Request[ StreamType: "user", StreamID: req.Msg.Id, EventType: string(eventtypes.UserEmailChanged), - Data: map[string]any{ - "email": req.Msg.Email, + Data: payloads.UserEmailChanged{ + Email: &req.Msg.Email, }, ActorType: "user", ActorID: userCtx.ID, @@ -394,8 +400,8 @@ func (h *UserHandler) UpdateUserPassword(ctx context.Context, req *connect.Reque StreamType: "user", StreamID: req.Msg.Id, EventType: string(eventtypes.UserPasswordChanged), - Data: map[string]any{ - "password_hash": passwordHash, + Data: payloads.UserPasswordChanged{ + PasswordHash: &passwordHash, }, ActorType: "user", ActorID: userCtx.ID, @@ -523,13 +529,13 @@ func (h *UserHandler) UpdateUserProfile(ctx context.Context, req *connect.Reques StreamType: "user", StreamID: req.Msg.Id, EventType: string(eventtypes.UserProfileUpdated), - Data: map[string]any{ - "display_name": req.Msg.DisplayName, - "given_name": req.Msg.GivenName, - "family_name": req.Msg.FamilyName, - "preferred_username": req.Msg.PreferredUsername, - "picture": req.Msg.Picture, - "locale": req.Msg.Locale, + Data: payloads.UserProfileUpdated{ + DisplayName: &req.Msg.DisplayName, + GivenName: &req.Msg.GivenName, + FamilyName: &req.Msg.FamilyName, + PreferredUsername: &req.Msg.PreferredUsername, + Picture: &req.Msg.Picture, + Locale: &req.Msg.Locale, }, ActorType: "user", ActorID: userCtx.ID, @@ -620,12 +626,13 @@ func (h *UserHandler) SetUserProvisioningEnabled(ctx context.Context, req *conne return nil, err } + enabled := req.Msg.Enabled err = h.store.AppendEvent(ctx, store.Event{ StreamType: "user", StreamID: req.Msg.UserId, EventType: string(eventtypes.UserProvisioningSettingsUpdated), - Data: map[string]any{ - "user_provisioning_enabled": req.Msg.Enabled, + Data: payloads.UserProvisioningSettingsUpdated{ + UserProvisioningEnabled: &enabled, }, ActorType: "user", ActorID: userCtx.ID, @@ -675,8 +682,8 @@ func (h *UserHandler) UpdateUserLinuxUsername(ctx context.Context, req *connect. StreamType: "user", StreamID: req.Msg.UserId, EventType: string(eventtypes.UserLinuxUsernameChanged), - Data: map[string]any{ - "linux_username": username, + Data: payloads.UserLinuxUsernameChanged{ + LinuxUsername: &username, }, ActorType: "user", ActorID: userCtx.ID, @@ -716,16 +723,17 @@ func (h *UserHandler) AddUserSshKey(ctx context.Context, req *connect.Request[pm keyID := ulid.Make().String() now := time.Now() + addedAt := now.Format(time.RFC3339) err = h.store.AppendEvent(ctx, store.Event{ StreamType: "user", StreamID: req.Msg.UserId, EventType: string(eventtypes.UserSshKeyAdded), - Data: map[string]any{ - "key_id": keyID, - "public_key": req.Msg.PublicKey, - "comment": req.Msg.Comment, - "added_at": now.Format(time.RFC3339), + Data: payloads.UserSshKeyAdded{ + KeyID: &keyID, + PublicKey: &req.Msg.PublicKey, + Comment: &req.Msg.Comment, + AddedAt: &addedAt, }, ActorType: "user", ActorID: userCtx.ID, @@ -765,8 +773,8 @@ func (h *UserHandler) RemoveUserSshKey(ctx context.Context, req *connect.Request StreamType: "user", StreamID: req.Msg.UserId, EventType: string(eventtypes.UserSshKeyRemoved), - Data: map[string]any{ - "key_id": req.Msg.KeyId, + Data: payloads.UserSshKeyRemoved{ + KeyID: &req.Msg.KeyId, }, ActorType: "user", ActorID: userCtx.ID, @@ -795,14 +803,17 @@ func (h *UserHandler) UpdateUserSshSettings(ctx context.Context, req *connect.Re return nil, err } + sshAccess := req.Msg.SshAccessEnabled + sshPubkey := req.Msg.SshAllowPubkey + sshPassword := req.Msg.SshAllowPassword err = h.store.AppendEvent(ctx, store.Event{ StreamType: "user", StreamID: req.Msg.UserId, EventType: string(eventtypes.UserSshSettingsUpdated), - Data: map[string]any{ - "ssh_access_enabled": req.Msg.SshAccessEnabled, - "ssh_allow_pubkey": req.Msg.SshAllowPubkey, - "ssh_allow_password": req.Msg.SshAllowPassword, + Data: payloads.UserSshSettingsUpdated{ + SshAccessEnabled: &sshAccess, + SshAllowPubkey: &sshPubkey, + SshAllowPassword: &sshPassword, }, ActorType: "user", ActorID: userCtx.ID, diff --git a/internal/control/inbox_worker.go b/internal/control/inbox_worker.go index aa1d8e21..04da7ba5 100644 --- a/internal/control/inbox_worker.go +++ b/internal/control/inbox_worker.go @@ -17,6 +17,7 @@ import ( pm "github.com/manchtools/power-manage/sdk/gen/go/pm/v1" "github.com/manchtools/power-manage/server/internal/ca" "github.com/manchtools/power-manage/server/internal/eventtypes" + "github.com/manchtools/power-manage/server/internal/eventtypes/payloads" "github.com/manchtools/power-manage/server/internal/store" db "github.com/manchtools/power-manage/server/internal/store/generated" "github.com/manchtools/power-manage/server/internal/taskqueue" @@ -264,14 +265,19 @@ func (w *InboxWorker) handleExecutionResult(ctx context.Context, t *asynq.Task) } cachedAction = &action - createdData := map[string]any{ - "device_id": deviceID, - "action_id": actionID, - "action_type": action.ActionType, - "desired_state": 0, - "params": json.RawMessage(action.Params), - "timeout_seconds": action.TimeoutSeconds, - "executed_at": executedAt.Format(time.RFC3339Nano), + actionIDCopy := actionID + actionType := action.ActionType + desiredState := int32(0) + timeoutSeconds := action.TimeoutSeconds + executedAtStr := executedAt.Format(time.RFC3339Nano) + createdData := payloads.ExecutionCreated{ + DeviceID: deviceID, + ActionID: &actionIDCopy, + ActionType: &actionType, + DesiredState: &desiredState, + Params: json.RawMessage(action.Params), + TimeoutSeconds: &timeoutSeconds, + ExecutedAt: &executedAtStr, } if err := w.store.AppendEvent(ctx, store.Event{ StreamType: "execution", @@ -285,31 +291,44 @@ func (w *InboxWorker) handleExecutionResult(ctx context.Context, t *asynq.Task) } } - // Map proto status to event type + // Map proto status to event type. data holds whichever payloads.* + // struct matches — store.Event.Data accepts any JSON-marshalable + // value, so each branch picks the typed wire shape rather than + // dropping back to map[string]any. var eventType string - var data map[string]any + var data any + completedAtStr := completedAt.Format(time.RFC3339Nano) switch result.Status { case pm.ExecutionStatus_EXECUTION_STATUS_SUCCESS: eventType = "ExecutionCompleted" - data = map[string]any{ - "duration_ms": result.DurationMs, - "completed_at": completedAt.Format(time.RFC3339Nano), - "changed": result.Changed, - "compliant": result.Compliant, + durationMs := result.DurationMs + changed := result.Changed + compliant := result.Compliant + data = payloads.ExecutionTerminal{ + CompletedAt: &completedAtStr, + DurationMs: &durationMs, + Changed: &changed, + Compliant: &compliant, + Output: payloads.RawCommandOutput(commandOutputPayload(result.Output)), + DetectionOutput: payloads.RawCommandOutput(commandOutputPayload(result.DetectionOutput)), } - addCommandOutputs(data, &result) case pm.ExecutionStatus_EXECUTION_STATUS_FAILED: eventType = "ExecutionFailed" - data = map[string]any{ - "error": result.Error, - "duration_ms": result.DurationMs, - "completed_at": completedAt.Format(time.RFC3339Nano), - "changed": result.Changed, - "compliant": result.Compliant, + errStr := result.Error + durationMs := result.DurationMs + changed := result.Changed + compliant := result.Compliant + data = payloads.ExecutionTerminal{ + Error: &errStr, + CompletedAt: &completedAtStr, + DurationMs: &durationMs, + Changed: &changed, + Compliant: &compliant, + Output: payloads.RawCommandOutput(commandOutputPayload(result.Output)), + DetectionOutput: payloads.RawCommandOutput(commandOutputPayload(result.DetectionOutput)), } - addCommandOutputs(data, &result) case pm.ExecutionStatus_EXECUTION_STATUS_RUNNING: eventType = "ExecutionStarted" @@ -317,20 +336,22 @@ func (w *InboxWorker) handleExecutionResult(ctx context.Context, t *asynq.Task) case pm.ExecutionStatus_EXECUTION_STATUS_TIMEOUT: eventType = "ExecutionTimedOut" - data = map[string]any{ - "error": result.Error, - "duration_ms": result.DurationMs, - "completed_at": completedAt.Format(time.RFC3339Nano), - } - if m := commandOutputToMap(result.Output); m != nil { - data["output"] = m + errStr := result.Error + durationMs := result.DurationMs + data = payloads.ExecutionTimedOut{ + Error: &errStr, + CompletedAt: &completedAtStr, + DurationMs: &durationMs, + Output: payloads.RawCommandOutput(commandOutputPayload(result.Output)), } case pm.ExecutionStatus_EXECUTION_STATUS_SKIPPED: eventType = "ExecutionSkipped" - data = map[string]any{} if result.Error != "" { - data["reason"] = result.Error + reason := result.Error + data = payloads.ExecutionReason{Reason: &reason} + } else { + data = payloads.ExecutionReason{} } default: @@ -769,8 +790,10 @@ func (w *InboxWorker) dispatchPendingActions(ctx context.Context, deviceID strin return nil } -// commandOutputToMap converts a CommandOutput proto to a map for event data. -// Returns nil if the output is nil. +// commandOutputToMap converts a CommandOutput proto to a map for event +// data. Returns nil if the output is nil. Used by emit sites that +// still construct payloads as map[string]any (the compliance event +// shape, for now). func commandOutputToMap(o *pm.CommandOutput) map[string]any { if o == nil { return nil @@ -782,13 +805,18 @@ func commandOutputToMap(o *pm.CommandOutput) map[string]any { } } -// addCommandOutputs adds output and detection_output fields to the event data map. -func addCommandOutputs(data map[string]any, result *pm.ActionResult) { - if m := commandOutputToMap(result.Output); m != nil { - data["output"] = m +// commandOutputPayload converts a CommandOutput proto into the typed +// payloads.CommandOutput used by the execution-event payload structs. +// Returns nil for nil input so payloads.RawCommandOutput drops the +// field via omitempty. +func commandOutputPayload(o *pm.CommandOutput) *payloads.CommandOutput { + if o == nil { + return nil } - if m := commandOutputToMap(result.DetectionOutput); m != nil { - data["detection_output"] = m + return &payloads.CommandOutput{ + Stdout: o.Stdout, + Stderr: o.Stderr, + ExitCode: o.ExitCode, } } diff --git a/internal/eventtypes/payloads/action.go b/internal/eventtypes/payloads/action.go new file mode 100644 index 00000000..41e708a5 --- /dev/null +++ b/internal/eventtypes/payloads/action.go @@ -0,0 +1,39 @@ +package payloads + +import "encoding/json" + +// ActionCreated is the wire shape for ActionCreated. Name is required +// (NOT NULL column on actions_projection); it stays a non-pointer +// string so the handler can't accidentally omit it. The projector +// validates that it is non-empty. +type ActionCreated struct { + Name string `json:"name"` + Description *string `json:"description,omitempty"` + ActionType *int32 `json:"action_type,omitempty"` + DesiredState *int32 `json:"desired_state,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + TimeoutSeconds *int32 `json:"timeout_seconds,omitempty"` + IsSystem *bool `json:"is_system,omitempty"` + Schedule json.RawMessage `json:"schedule,omitempty"` +} + +// ActionRenamed is the wire shape for ActionRenamed. +type ActionRenamed struct { + Name string `json:"name"` +} + +// ActionDescriptionUpdated is the wire shape for +// ActionDescriptionUpdated. The PL/pgSQL projector wrote +// `event.data->>'description'` directly; *string preserves the +// nullable-vs-empty distinction. +type ActionDescriptionUpdated struct { + Description *string `json:"description,omitempty"` +} + +// ActionParamsUpdated is the wire shape for ActionParamsUpdated. +type ActionParamsUpdated struct { + Params json.RawMessage `json:"params,omitempty"` + TimeoutSeconds *int32 `json:"timeout_seconds,omitempty"` + DesiredState *int32 `json:"desired_state,omitempty"` + Schedule json.RawMessage `json:"schedule,omitempty"` +} diff --git a/internal/eventtypes/payloads/assignment.go b/internal/eventtypes/payloads/assignment.go new file mode 100644 index 00000000..89f74ece --- /dev/null +++ b/internal/eventtypes/payloads/assignment.go @@ -0,0 +1,27 @@ +package payloads + +// AssignmentCreated is the wire shape for AssignmentCreated. The four +// tuple fields are required (NOT NULL columns); they stay non-pointer +// strings so the handler emit site can't omit them. The projector +// validates that none are empty. +type AssignmentCreated struct { + SourceType string `json:"source_type"` + SourceID string `json:"source_id"` + TargetType string `json:"target_type"` + TargetID string `json:"target_id"` + SortOrder *int32 `json:"sort_order,omitempty"` + Mode *int32 `json:"mode,omitempty"` +} + +// AssignmentModeChanged is the wire shape for AssignmentModeChanged. +// Currently unused (assignments are immutable; mutate by +// delete-and-recreate) but the projector preserves replay parity. +type AssignmentModeChanged struct { + Mode *int32 `json:"mode,omitempty"` +} + +// AssignmentSortOrderChanged is the wire shape for +// AssignmentSortOrderChanged. Currently unused; preserved for replay. +type AssignmentSortOrderChanged struct { + SortOrder *int32 `json:"sort_order,omitempty"` +} diff --git a/internal/eventtypes/payloads/device.go b/internal/eventtypes/payloads/device.go new file mode 100644 index 00000000..4b1f6685 --- /dev/null +++ b/internal/eventtypes/payloads/device.go @@ -0,0 +1,84 @@ +package payloads + +import "encoding/json" + +// DeviceRegistered is the wire shape for DeviceRegistered. Pointer +// fields preserve the absent-vs-explicit distinction the listener's +// SQL UPSERT relies on (omitted CertFingerprint => SQL NULL, +// permitted by the column's UNIQUE constraint that allows multiple +// NULLs). CertNotAfter is the RFC 3339 string emitted by the +// registration handler — kept as *string so the wire bytes stay +// byte-identical with the legacy map-literal emission (which used +// time.Format(time.RFC3339), no sub-second precision). The projector +// parses the string back into a time.Time before writing. +type DeviceRegistered struct { + Hostname *string `json:"hostname,omitempty"` + AgentVersion *string `json:"agent_version,omitempty"` + CertFingerprint *string `json:"cert_fingerprint,omitempty"` + CertNotAfter *string `json:"cert_not_after,omitempty"` + RegistrationTokenID *string `json:"registration_token_id,omitempty"` + Labels json.RawMessage `json:"labels,omitempty"` + AssignedUserID *string `json:"assigned_user_id,omitempty"` + // CertPEM and CACertPEM are emitted by the registration handler so + // the response can return signed cert bytes; the projector + // ignores them. Keeping them here lets the handler pass one typed + // payload to AppendEvent without dropping back to a map literal. + CertPEM *string `json:"cert_pem,omitempty"` + CACertPEM *string `json:"ca_cert_pem,omitempty"` +} + +// DeviceSeen is the wire shape for DeviceSeen. +type DeviceSeen struct { + AgentVersion *string `json:"agent_version,omitempty"` + Hostname *string `json:"hostname,omitempty"` +} + +// DeviceHeartbeat is the wire shape for DeviceHeartbeat. +type DeviceHeartbeat struct { + AgentVersion *string `json:"agent_version,omitempty"` +} + +// DeviceCertRenewed is the wire shape for DeviceCertRenewed. +// CertNotAfter is the RFC 3339 string emitted by the renewal handler; +// matches the legacy emit shape and the projector parses back into a +// time.Time before writing. +type DeviceCertRenewed struct { + CertFingerprint *string `json:"cert_fingerprint,omitempty"` + CertNotAfter *string `json:"cert_not_after,omitempty"` +} + +// DeviceLabelsUpdated is the wire shape for DeviceLabelsUpdated. +type DeviceLabelsUpdated struct { + Labels json.RawMessage `json:"labels,omitempty"` +} + +// DeviceLabelSet is the wire shape for DeviceLabelSet. +type DeviceLabelSet struct { + Key *string `json:"key,omitempty"` + Value *string `json:"value,omitempty"` +} + +// DeviceLabelRemoved is the wire shape for DeviceLabelRemoved. +type DeviceLabelRemoved struct { + Key *string `json:"key,omitempty"` +} + +// DeviceUserAssignment is the wire shape for DeviceAssigned and +// DeviceUnassigned. user_id is required (composite-PK column on the +// projection) but stays a pointer here so the wire encoding matches +// the legacy map[string]any verbatim — the projector validates +// presence on the read side. +type DeviceUserAssignment struct { + UserID *string `json:"user_id,omitempty"` +} + +// DeviceGroupAssignment is the wire shape for DeviceGroupAssigned and +// DeviceGroupUnassigned. +type DeviceGroupAssignment struct { + GroupID *string `json:"group_id,omitempty"` +} + +// DeviceSyncIntervalSet is the wire shape for DeviceSyncIntervalSet. +type DeviceSyncIntervalSet struct { + SyncIntervalMinutes *int32 `json:"sync_interval_minutes,omitempty"` +} diff --git a/internal/eventtypes/payloads/doc.go b/internal/eventtypes/payloads/doc.go new file mode 100644 index 00000000..e6da62b7 --- /dev/null +++ b/internal/eventtypes/payloads/doc.go @@ -0,0 +1,33 @@ +// Package payloads holds the shared JSON-shape structs that travel in +// the events table's data column. Each struct is the single source of +// truth for one event type's wire format: the handler emit site +// constructs an instance, AppendEvent json.Marshals it into JSONB, and +// the projector decoder json.Unmarshals it back out. Sharing one +// struct between emit and decode catches schema drift (renamed field, +// typo'd key, dropped field) at compile time instead of at projection +// replay. +// +// Field-tag rules: +// +// - Preserve the json tag exactly as it appears in the legacy +// handler-side map[string]any literal. Any change to the wire key +// is a migration, not a refactor — old events in the events table +// must keep decoding cleanly. +// - omitempty matches what the deleted PL/pgSQL projector expected. +// A field that the projector treats as "missing key falls back to +// existing column value" must use omitempty so the struct's zero +// value round-trips as an absent key, not as a JSON null that +// overwrites the column. +// - Pointer types preserve the absent-vs-explicit-zero distinction. +// Use them when the projector's COALESCE semantics depend on it +// (DeviceSeen.AgentVersion, UserSshSettingsUpdated.SshAccessEnabled, +// etc.). +// - json.RawMessage is used for nested JSONB blobs (action params, +// device labels, schedule) so wire bytes pass through verbatim +// without a marshal/unmarshal round trip. +// +// Roundtrip tests in payloads_test.go assert that every payload struct +// survives a json.Marshal -> json.Unmarshal cycle byte-identical, so a +// future field-tag typo or accidental encoding-changing edit fails CI +// before reaching production. +package payloads diff --git a/internal/eventtypes/payloads/execution.go b/internal/eventtypes/payloads/execution.go new file mode 100644 index 00000000..b6ff66f1 --- /dev/null +++ b/internal/eventtypes/payloads/execution.go @@ -0,0 +1,93 @@ +package payloads + +import "encoding/json" + +// CommandOutput is the wire shape for the nested output / detection_output +// JSONB blobs that ride along on terminal execution events. +// Mirrors the legacy commandOutputToMap helper. +type CommandOutput struct { + Stdout string `json:"stdout"` + Stderr string `json:"stderr"` + ExitCode int32 `json:"exit_code"` +} + +// ExecutionCreated is the wire shape for ExecutionCreated emitted by +// the gateway inbox worker when a derived execution from an offline +// agent result needs a parent row. The projector requires DeviceID +// and ActionType; the rest fall back to PL/pgSQL-equivalent defaults. +type ExecutionCreated struct { + DeviceID string `json:"device_id"` + ActionID *string `json:"action_id,omitempty"` + DefinitionID *string `json:"definition_id,omitempty"` + ActionType *int32 `json:"action_type,omitempty"` + DesiredState *int32 `json:"desired_state,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + TimeoutSeconds *int32 `json:"timeout_seconds,omitempty"` + ExecutedAt *string `json:"executed_at,omitempty"` +} + +// ExecutionScheduled is the wire shape for ExecutionScheduled. +// scheduled_for is required (the only emitter populates it +// unconditionally — a missing key is an emitter bug). +type ExecutionScheduled struct { + DeviceID string `json:"device_id"` + ActionID *string `json:"action_id,omitempty"` + DefinitionID *string `json:"definition_id,omitempty"` + ActionType *int32 `json:"action_type,omitempty"` + DesiredState *int32 `json:"desired_state,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + TimeoutSeconds *int32 `json:"timeout_seconds,omitempty"` + ScheduledFor string `json:"scheduled_for"` +} + +// ExecutionTerminal is the wire shape shared by ExecutionCompleted and +// ExecutionFailed. Output and DetectionOutput ride as JSONB blobs so +// the projector can pass the bytes straight to the JSONB column without +// a marshal round-trip; the emit-side helper RawCommandOutput converts +// a typed CommandOutput value into the json.RawMessage form. Error is +// populated only on the Failed variant. +type ExecutionTerminal struct { + CompletedAt *string `json:"completed_at,omitempty"` + Error *string `json:"error,omitempty"` + Output json.RawMessage `json:"output,omitempty"` + DurationMs *int64 `json:"duration_ms,omitempty"` + Changed *bool `json:"changed,omitempty"` + Compliant *bool `json:"compliant,omitempty"` + DetectionOutput json.RawMessage `json:"detection_output,omitempty"` +} + +// ExecutionTimedOut is the wire shape for ExecutionTimedOut. Subset of +// the terminal shape — no changed / compliant / detection_output. +type ExecutionTimedOut struct { + CompletedAt *string `json:"completed_at,omitempty"` + Error *string `json:"error,omitempty"` + Output json.RawMessage `json:"output,omitempty"` + DurationMs *int64 `json:"duration_ms,omitempty"` +} + +// RawCommandOutput marshals a CommandOutput into the json.RawMessage +// form used by the terminal-event payloads. Returns nil when the input +// is nil so the omitempty tag fires and the field is dropped from the +// wire payload (matches the legacy commandOutputToMap helper that +// returned nil for a nil proto). +func RawCommandOutput(o *CommandOutput) json.RawMessage { + if o == nil { + return nil + } + b, err := json.Marshal(o) + if err != nil { + // CommandOutput has only string and int32 fields; encoding + // can't fail. The error path exists only to satisfy + // json.Marshal's signature — surfacing it would force every + // emit site to handle an impossible error. + return nil + } + return b +} + +// ExecutionReason is the wire shape shared by ExecutionSkipped and +// ExecutionCancelled. The reason rides on the wire as `reason` and is +// projected into the error column. +type ExecutionReason struct { + Reason *string `json:"reason,omitempty"` +} diff --git a/internal/eventtypes/payloads/lps_password.go b/internal/eventtypes/payloads/lps_password.go new file mode 100644 index 00000000..60b0a43d --- /dev/null +++ b/internal/eventtypes/payloads/lps_password.go @@ -0,0 +1,39 @@ +package payloads + +import ( + "log/slog" + "time" +) + +// LpsPasswordRotated is the wire shape for LpsPasswordRotated. The +// projector struct in internal/projectors/lps_password.go decodes into +// the same JSON shape; sharing this type at the emit site lets a +// future field rename catch at compile time. +// +// Password is encrypted ciphertext (AES-GCM via internal/crypto) so +// the emitter never holds plaintext at this layer. LogValue still +// masks it defensively in case a future call site routes the whole +// payload through slog. +type LpsPasswordRotated struct { + DeviceID string `json:"device_id"` + ActionID string `json:"action_id"` + Username string `json:"username"` + Password string `json:"password"` + RotatedAt time.Time `json:"rotated_at"` + RotationReason string `json:"rotation_reason"` +} + +// LogValue masks the encrypted Password so a future +// `logger.Warn("…", "payload", payload)` or `fmt.Sprintf("%+v", p)` +// routed through slog cannot leak the credential. Mirrors the +// LpsPasswordRotatedPayload masking in the projector. +func (p LpsPasswordRotated) LogValue() slog.Value { + return slog.GroupValue( + slog.String("device_id", p.DeviceID), + slog.String("action_id", p.ActionID), + slog.String("username", p.Username), + slog.String("password", "[REDACTED]"), + slog.Time("rotated_at", p.RotatedAt), + slog.String("rotation_reason", p.RotationReason), + ) +} diff --git a/internal/eventtypes/payloads/luks_key.go b/internal/eventtypes/payloads/luks_key.go new file mode 100644 index 00000000..737daa6c --- /dev/null +++ b/internal/eventtypes/payloads/luks_key.go @@ -0,0 +1,30 @@ +package payloads + +import ( + "log/slog" + "time" +) + +// LuksKeyRotated is the wire shape for LuksKeyRotated. Same pattern as +// LpsPasswordRotated but partitioned by device_path. +type LuksKeyRotated struct { + DeviceID string `json:"device_id"` + ActionID string `json:"action_id"` + DevicePath string `json:"device_path"` + Passphrase string `json:"passphrase"` + RotatedAt time.Time `json:"rotated_at"` + RotationReason string `json:"rotation_reason"` +} + +// LogValue masks the encrypted Passphrase. See LpsPasswordRotated for +// the rationale. +func (p LuksKeyRotated) LogValue() slog.Value { + return slog.GroupValue( + slog.String("device_id", p.DeviceID), + slog.String("action_id", p.ActionID), + slog.String("device_path", p.DevicePath), + slog.String("passphrase", "[REDACTED]"), + slog.Time("rotated_at", p.RotatedAt), + slog.String("rotation_reason", p.RotationReason), + ) +} diff --git a/internal/eventtypes/payloads/payloads_test.go b/internal/eventtypes/payloads/payloads_test.go new file mode 100644 index 00000000..60db7562 --- /dev/null +++ b/internal/eventtypes/payloads/payloads_test.go @@ -0,0 +1,601 @@ +package payloads_test + +import ( + "encoding/json" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/manchtools/power-manage/server/internal/eventtypes/payloads" +) + +// All roundtrip tests assert json.Marshal -> json.Unmarshal preserves +// the struct byte-identically. A field-tag typo on either side fails +// the test loudly: the marshal writes one key, the unmarshal looks for +// another, the assert.Equal catches the missing fields. This guards +// the wire contract that the projector decoder + handler emit site +// share — exactly the regression class PR F was created to prevent. + +func ptr[T any](v T) *T { return &v } + +func TestRoundtrip_UserCreatedWithRoles(t *testing.T) { + in := payloads.UserCreatedWithRoles{ + Email: ptr("a@b.com"), + PasswordHash: ptr("hash"), + Role: ptr("admin"), + DisplayName: ptr("Alice"), + GivenName: ptr("Alice"), + FamilyName: ptr("Example"), + PreferredUsername: ptr("alice"), + Picture: ptr("https://example.com/a.png"), + Locale: ptr("en-US"), + LinuxUsername: ptr("alice"), + LinuxUID: ptr(int32(1001)), + RoleIDs: []string{"role-1", "role-2"}, + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.UserCreatedWithRoles + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out, "must roundtrip cleanly through the events table JSONB") +} + +func TestRoundtrip_UserProfileUpdated(t *testing.T) { + in := payloads.UserProfileUpdated{ + DisplayName: ptr("Alice"), + GivenName: ptr("Alice"), + FamilyName: ptr("Example"), + PreferredUsername: ptr("alice"), + Picture: ptr("https://example.com/a.png"), + Locale: ptr("en-US"), + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.UserProfileUpdated + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_UserEmailChanged(t *testing.T) { + in := payloads.UserEmailChanged{Email: ptr("new@example.com")} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.UserEmailChanged + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_UserPasswordChanged(t *testing.T) { + in := payloads.UserPasswordChanged{PasswordHash: ptr("argon2id$...")} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.UserPasswordChanged + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_UserRoleChanged(t *testing.T) { + in := payloads.UserRoleChanged{Role: ptr("admin")} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.UserRoleChanged + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_UserSshKeyAdded(t *testing.T) { + in := payloads.UserSshKeyAdded{ + KeyID: ptr("01H..."), + PublicKey: ptr("ssh-ed25519 AAAA..."), + Comment: ptr("alice@laptop"), + AddedAt: ptr("2026-05-08T12:00:00Z"), + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.UserSshKeyAdded + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_UserSshKeyRemoved(t *testing.T) { + in := payloads.UserSshKeyRemoved{KeyID: ptr("01H...")} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.UserSshKeyRemoved + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_UserSshSettingsUpdated(t *testing.T) { + in := payloads.UserSshSettingsUpdated{ + SshAccessEnabled: ptr(true), + SshAllowPubkey: ptr(true), + SshAllowPassword: ptr(false), + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.UserSshSettingsUpdated + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_UserLinuxUsernameChanged(t *testing.T) { + in := payloads.UserLinuxUsernameChanged{LinuxUsername: ptr("alice")} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.UserLinuxUsernameChanged + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_UserSystemActionLinked(t *testing.T) { + in := payloads.UserSystemActionLinked{ + Field: ptr("system_user_action_id"), + ActionID: ptr("01H..."), + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.UserSystemActionLinked + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_UserProvisioningSettingsUpdated(t *testing.T) { + in := payloads.UserProvisioningSettingsUpdated{ + UserProvisioningEnabled: ptr(true), + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.UserProvisioningSettingsUpdated + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_DeviceRegistered(t *testing.T) { + in := payloads.DeviceRegistered{ + Hostname: ptr("host-01"), + AgentVersion: ptr("2026.06.0"), + CertFingerprint: ptr("aabbcc"), + CertNotAfter: ptr("2027-05-08T12:00:00Z"), + RegistrationTokenID: ptr("token-1"), + Labels: json.RawMessage(`{"env":"prod"}`), + AssignedUserID: ptr("user-1"), + CertPEM: ptr("-----BEGIN CERTIFICATE-----..."), + CACertPEM: ptr("-----BEGIN CERTIFICATE-----..."), + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.DeviceRegistered + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_DeviceSeen(t *testing.T) { + in := payloads.DeviceSeen{ + AgentVersion: ptr("2026.06.0"), + Hostname: ptr("host-01"), + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.DeviceSeen + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_DeviceHeartbeat(t *testing.T) { + in := payloads.DeviceHeartbeat{AgentVersion: ptr("2026.06.0")} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.DeviceHeartbeat + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_DeviceCertRenewed(t *testing.T) { + in := payloads.DeviceCertRenewed{ + CertFingerprint: ptr("aabbcc"), + CertNotAfter: ptr("2027-05-08T12:00:00Z"), + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.DeviceCertRenewed + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_DeviceLabelsUpdated(t *testing.T) { + in := payloads.DeviceLabelsUpdated{Labels: json.RawMessage(`{"env":"prod"}`)} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.DeviceLabelsUpdated + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_DeviceLabelSet(t *testing.T) { + in := payloads.DeviceLabelSet{Key: ptr("env"), Value: ptr("prod")} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.DeviceLabelSet + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_DeviceLabelRemoved(t *testing.T) { + in := payloads.DeviceLabelRemoved{Key: ptr("env")} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.DeviceLabelRemoved + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_DeviceUserAssignment(t *testing.T) { + in := payloads.DeviceUserAssignment{UserID: ptr("user-1")} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.DeviceUserAssignment + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_DeviceGroupAssignment(t *testing.T) { + in := payloads.DeviceGroupAssignment{GroupID: ptr("group-1")} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.DeviceGroupAssignment + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_DeviceSyncIntervalSet(t *testing.T) { + in := payloads.DeviceSyncIntervalSet{SyncIntervalMinutes: ptr(int32(15))} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.DeviceSyncIntervalSet + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_ActionCreated(t *testing.T) { + in := payloads.ActionCreated{ + Name: "install-nginx", + Description: ptr("install nginx package"), + ActionType: ptr(int32(1)), + DesiredState: ptr(int32(1)), + Params: json.RawMessage(`{"package":"nginx"}`), + TimeoutSeconds: ptr(int32(300)), + IsSystem: ptr(false), + Schedule: json.RawMessage(`{"cron":"0 0 * * *"}`), + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.ActionCreated + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_ActionRenamed(t *testing.T) { + in := payloads.ActionRenamed{Name: "install-nginx-v2"} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.ActionRenamed + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_ActionDescriptionUpdated(t *testing.T) { + in := payloads.ActionDescriptionUpdated{Description: ptr("new description")} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.ActionDescriptionUpdated + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_ActionParamsUpdated(t *testing.T) { + in := payloads.ActionParamsUpdated{ + Params: json.RawMessage(`{"package":"nginx","version":"1.24"}`), + TimeoutSeconds: ptr(int32(600)), + DesiredState: ptr(int32(0)), + Schedule: json.RawMessage(`{"cron":"0 6 * * *"}`), + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.ActionParamsUpdated + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_AssignmentCreated(t *testing.T) { + in := payloads.AssignmentCreated{ + SourceType: "action", + SourceID: "act-1", + TargetType: "device", + TargetID: "dev-1", + SortOrder: ptr(int32(0)), + Mode: ptr(int32(1)), + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.AssignmentCreated + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_AssignmentModeChanged(t *testing.T) { + in := payloads.AssignmentModeChanged{Mode: ptr(int32(2))} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.AssignmentModeChanged + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_AssignmentSortOrderChanged(t *testing.T) { + in := payloads.AssignmentSortOrderChanged{SortOrder: ptr(int32(7))} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.AssignmentSortOrderChanged + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_ExecutionCreated(t *testing.T) { + in := payloads.ExecutionCreated{ + DeviceID: "dev-1", + ActionID: ptr("act-1"), + DefinitionID: ptr("def-1"), + ActionType: ptr(int32(1)), + DesiredState: ptr(int32(1)), + Params: json.RawMessage(`{"package":"nginx"}`), + TimeoutSeconds: ptr(int32(300)), + ExecutedAt: ptr("2026-05-08T12:00:00Z"), + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.ExecutionCreated + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_ExecutionScheduled(t *testing.T) { + in := payloads.ExecutionScheduled{ + DeviceID: "dev-1", + ActionID: ptr("act-1"), + ActionType: ptr(int32(1)), + DesiredState: ptr(int32(1)), + Params: json.RawMessage(`{"package":"nginx"}`), + TimeoutSeconds: ptr(int32(300)), + ScheduledFor: "2026-05-09T12:00:00Z", + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.ExecutionScheduled + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_ExecutionTerminal(t *testing.T) { + in := payloads.ExecutionTerminal{ + CompletedAt: ptr("2026-05-08T12:00:00Z"), + Error: ptr("non-zero exit"), + Output: json.RawMessage(`{"stdout":"ok","stderr":"","exit_code":0}`), + DurationMs: ptr(int64(1500)), + Changed: ptr(true), + Compliant: ptr(false), + DetectionOutput: json.RawMessage(`{"stdout":"compliant","stderr":"","exit_code":0}`), + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.ExecutionTerminal + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_ExecutionTimedOut(t *testing.T) { + in := payloads.ExecutionTimedOut{ + CompletedAt: ptr("2026-05-08T12:00:00Z"), + Error: ptr("timeout"), + Output: json.RawMessage(`{"stdout":"","stderr":"killed","exit_code":-1}`), + DurationMs: ptr(int64(300000)), + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.ExecutionTimedOut + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_ExecutionReason(t *testing.T) { + in := payloads.ExecutionReason{Reason: ptr("device offline during scheduled window")} + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.ExecutionReason + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) +} + +func TestRoundtrip_LpsPasswordRotated(t *testing.T) { + rotated := time.Date(2026, 5, 8, 12, 0, 0, 0, time.UTC) + in := payloads.LpsPasswordRotated{ + DeviceID: "dev-1", + ActionID: "act-1", + Username: "alice", + Password: "ENCRYPTED-CIPHERTEXT", + RotatedAt: rotated, + RotationReason: "scheduled", + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.LpsPasswordRotated + require.NoError(t, json.Unmarshal(raw, &out)) + assert.True(t, in.RotatedAt.Equal(out.RotatedAt), "RotatedAt must round-trip the same instant") + // Compare structurally with a normalised time (json round trip + // drops monotonic clock readings so == fails even when the + // instant matches). + in.RotatedAt = out.RotatedAt + assert.Equal(t, in, out) +} + +func TestRoundtrip_LuksKeyRotated(t *testing.T) { + rotated := time.Date(2026, 5, 8, 12, 0, 0, 0, time.UTC) + in := payloads.LuksKeyRotated{ + DeviceID: "dev-1", + ActionID: "act-1", + DevicePath: "/dev/sda1", + Passphrase: "ENCRYPTED-CIPHERTEXT", + RotatedAt: rotated, + RotationReason: "scheduled", + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.LuksKeyRotated + require.NoError(t, json.Unmarshal(raw, &out)) + assert.True(t, in.RotatedAt.Equal(out.RotatedAt)) + in.RotatedAt = out.RotatedAt + assert.Equal(t, in, out) +} + +// TestRoundtrip_CommandOutput covers the nested JSONB sub-shape used +// by the terminal-execution payloads. The wire format must exactly +// match the legacy commandOutputToMap output so historical events +// continue to decode through the same key set. +func TestRoundtrip_CommandOutput(t *testing.T) { + in := payloads.CommandOutput{ + Stdout: "hello\n", + Stderr: "warn: foo\n", + ExitCode: 0, + } + raw, err := json.Marshal(in) + require.NoError(t, err) + var out payloads.CommandOutput + require.NoError(t, json.Unmarshal(raw, &out)) + assert.Equal(t, in, out) + // Wire-shape sanity: keys must be EXACTLY the legacy ones so an + // extra key would break historical decoders (CR catch on PR #192). + var asMap map[string]any + require.NoError(t, json.Unmarshal(raw, &asMap)) + assert.Contains(t, asMap, "stdout") + assert.Contains(t, asMap, "stderr") + assert.Contains(t, asMap, "exit_code") + assert.Len(t, asMap, 3, "wire shape must keep the exact legacy key set") +} + +// TestRawCommandOutput_NilOmitted locks the contract used by +// ExecutionTerminal / ExecutionTimedOut: a nil *CommandOutput +// produces nil json.RawMessage so the surrounding payload's +// omitempty tag drops the field from the wire entirely. +func TestRawCommandOutput_NilOmitted(t *testing.T) { + assert.Nil(t, payloads.RawCommandOutput(nil), + "nil input must return nil so omitempty fires on the parent payload") + + // End-to-end omitempty contract: marshalling the parent payload + // with nil Output must NOT include the `output` key in the wire + // JSON. Without this assertion the helper could regress (e.g. + // to []byte("null")) and we'd only catch it via downstream + // projector failures (CR catch on PR #192). + completedAt := time.Date(2026, 5, 9, 0, 0, 0, 0, time.UTC).Format(time.RFC3339) + host := payloads.ExecutionTerminal{ + CompletedAt: &completedAt, + Output: payloads.RawCommandOutput(nil), + } + wire, err := json.Marshal(host) + require.NoError(t, err) + var asMap map[string]any + require.NoError(t, json.Unmarshal(wire, &asMap)) + assert.NotContains(t, asMap, "output", "omitempty contract must omit output when nil") + assert.NotContains(t, asMap, "detection_output", "omitempty must also drop detection_output when nil") +} + +// TestRawCommandOutput_ProducesObject locks that the helper writes +// a JSON object (not a quoted string), and exact legacy wire keys. +// A regression where the helper started emitting a quoted JSON +// string would silently break every consumer of historical +// ExecutionCompleted/ExecutionFailed events. +func TestRawCommandOutput_ProducesObject(t *testing.T) { + in := &payloads.CommandOutput{ + Stdout: "out\n", + Stderr: "err\n", + ExitCode: 1, + } + raw := payloads.RawCommandOutput(in) + require.NotNil(t, raw) + require.Greater(t, len(raw), 0) + // First byte must be '{', NOT '"' — distinguishes a JSON object + // from a quoted JSON string. + assert.Equal(t, byte('{'), raw[0], + "output must be a JSON object, not a quoted string") + + // Wire keys must be EXACTLY the legacy set — extra keys would + // break historical decoders (CR catch on PR #192). + var asMap map[string]any + require.NoError(t, json.Unmarshal(raw, &asMap)) + assert.Contains(t, asMap, "stdout") + assert.Contains(t, asMap, "stderr") + assert.Contains(t, asMap, "exit_code") + assert.Len(t, asMap, 3, "wire shape must keep the exact legacy key set") + + // Round-trip back through ExecutionTerminal so the test mirrors + // how the field actually rides on the wire. + completedAt := time.Date(2026, 5, 9, 0, 0, 0, 0, time.UTC).Format(time.RFC3339) + host := payloads.ExecutionTerminal{ + CompletedAt: &completedAt, + Output: raw, + } + wire, err := json.Marshal(host) + require.NoError(t, err) + var decoded payloads.ExecutionTerminal + require.NoError(t, json.Unmarshal(wire, &decoded)) + assert.JSONEq(t, string(raw), string(decoded.Output)) +} + +// TestLpsPasswordRotated_LogValueMasksPassword locks the security +// guarantee: routing the payload through slog must NEVER emit the +// raw Password value, even if a future call site does +// `slog.Warn("...", "payload", payload)` directly. +func TestLpsPasswordRotated_LogValueMasksPassword(t *testing.T) { + const secret = "super-secret-cipher-bytes" + p := payloads.LpsPasswordRotated{ + DeviceID: "dev-1", + ActionID: "act-1", + Username: "alice", + Password: secret, + RotatedAt: time.Date(2026, 5, 9, 12, 0, 0, 0, time.UTC), + RotationReason: "scheduled", + } + // Render the slog.Value into a string and grep for the secret. + v := p.LogValue() + rendered := v.String() + assert.NotContains(t, rendered, secret, + "LpsPasswordRotated.LogValue() must NOT contain the raw Password value") + assert.Contains(t, rendered, "[REDACTED]", + "LpsPasswordRotated.LogValue() must emit the explicit redaction marker") + // The non-secret context fields must still be present so logs + // remain actionable. + assert.Contains(t, rendered, "alice") + assert.Contains(t, rendered, "scheduled") +} + +// TestLuksKeyRotated_LogValueMasksPassphrase — sibling to the LPS +// test for LuksKeyRotated. Same security contract. +func TestLuksKeyRotated_LogValueMasksPassphrase(t *testing.T) { + const secret = "wXr-luks-passphrase-blob" + p := payloads.LuksKeyRotated{ + DeviceID: "dev-1", + ActionID: "act-1", + DevicePath: "/dev/sda1", + Passphrase: secret, + RotatedAt: time.Date(2026, 5, 9, 12, 0, 0, 0, time.UTC), + RotationReason: "initial", + } + v := p.LogValue() + rendered := v.String() + assert.NotContains(t, rendered, secret, + "LuksKeyRotated.LogValue() must NOT contain the raw Passphrase value") + assert.Contains(t, rendered, "[REDACTED]", + "LuksKeyRotated.LogValue() must emit the explicit redaction marker") + assert.Contains(t, rendered, "/dev/sda1") +} diff --git a/internal/eventtypes/payloads/user.go b/internal/eventtypes/payloads/user.go new file mode 100644 index 00000000..113caba1 --- /dev/null +++ b/internal/eventtypes/payloads/user.go @@ -0,0 +1,97 @@ +package payloads + +// UserCreatedWithRoles is the wire shape for the UserCreatedWithRoles +// compound event (issue #135). Pointer fields preserve the +// absent-vs-explicit distinction the PL/pgSQL projector's COALESCE +// chain relied on — see internal/projectors/user.go for the per-field +// fallback semantics. +type UserCreatedWithRoles struct { + Email *string `json:"email,omitempty"` + PasswordHash *string `json:"password_hash,omitempty"` + Role *string `json:"role,omitempty"` + DisplayName *string `json:"display_name,omitempty"` + GivenName *string `json:"given_name,omitempty"` + FamilyName *string `json:"family_name,omitempty"` + PreferredUsername *string `json:"preferred_username,omitempty"` + Picture *string `json:"picture,omitempty"` + Locale *string `json:"locale,omitempty"` + LinuxUsername *string `json:"linux_username,omitempty"` + LinuxUID *int32 `json:"linux_uid,omitempty"` + RoleIDs []string `json:"role_ids,omitempty"` +} + +// UserProfileUpdated is the wire shape for UserProfileUpdated. +type UserProfileUpdated struct { + DisplayName *string `json:"display_name,omitempty"` + GivenName *string `json:"given_name,omitempty"` + FamilyName *string `json:"family_name,omitempty"` + PreferredUsername *string `json:"preferred_username,omitempty"` + Picture *string `json:"picture,omitempty"` + Locale *string `json:"locale,omitempty"` +} + +// UserEmailChanged is the wire shape for UserEmailChanged. +type UserEmailChanged struct { + Email *string `json:"email,omitempty"` +} + +// UserPasswordChanged is the wire shape for UserPasswordChanged. +type UserPasswordChanged struct { + PasswordHash *string `json:"password_hash,omitempty"` +} + +// UserRoleChanged is the wire shape for UserRoleChanged. +type UserRoleChanged struct { + Role *string `json:"role,omitempty"` +} + +// UserSshKeyAdded is the wire shape for UserSshKeyAdded. PublicKey + +// Comment stay pointers so an omitted key marshals as JSON null +// (matches PL/pgSQL `event.data->>'public_key'` writing NULL into the +// JSONB element). AddedAt is the RFC 3339 string the legacy emit site +// stuffed into the map; the projector reads added-at from +// event.occurred_at instead so the field is not required, but keeping +// it on the wire preserves byte-identical event payloads with the +// pre-typed-payload emission for replay safety. +type UserSshKeyAdded struct { + KeyID *string `json:"key_id,omitempty"` + PublicKey *string `json:"public_key,omitempty"` + Comment *string `json:"comment,omitempty"` + AddedAt *string `json:"added_at,omitempty"` +} + +// UserSshKeyRemoved is the wire shape for UserSshKeyRemoved. +type UserSshKeyRemoved struct { + KeyID *string `json:"key_id,omitempty"` +} + +// UserSshSettingsUpdated is the wire shape for UserSshSettingsUpdated. +// Pointer bools preserve the COALESCE-on-missing semantics the +// projector relies on — a nil pointer means "preserve existing column +// value", a non-nil pointer means "set to value". +type UserSshSettingsUpdated struct { + SshAccessEnabled *bool `json:"ssh_access_enabled,omitempty"` + SshAllowPubkey *bool `json:"ssh_allow_pubkey,omitempty"` + SshAllowPassword *bool `json:"ssh_allow_password,omitempty"` +} + +// UserLinuxUsernameChanged is the wire shape for +// UserLinuxUsernameChanged. +type UserLinuxUsernameChanged struct { + LinuxUsername *string `json:"linux_username,omitempty"` +} + +// UserSystemActionLinked is the wire shape for UserSystemActionLinked. +// Field selects which of the three system_*_action_id columns gets +// the supplied action_id (see SystemActionField* constants in the +// projector package). +type UserSystemActionLinked struct { + Field *string `json:"field,omitempty"` + ActionID *string `json:"action_id,omitempty"` +} + +// UserProvisioningSettingsUpdated is the wire shape for +// UserProvisioningSettingsUpdated. +type UserProvisioningSettingsUpdated struct { + UserProvisioningEnabled *bool `json:"user_provisioning_enabled,omitempty"` +} diff --git a/internal/projectors/action.go b/internal/projectors/action.go index 6dbab38e..46319862 100644 --- a/internal/projectors/action.go +++ b/internal/projectors/action.go @@ -5,6 +5,7 @@ import ( "fmt" "github.com/manchtools/power-manage/server/internal/eventtypes" + "github.com/manchtools/power-manage/server/internal/eventtypes/payloads" "github.com/manchtools/power-manage/server/internal/store" ) @@ -48,17 +49,6 @@ type ActionCreatedPayload struct { CreatedBy string } -type actionCreatedRaw struct { - Name string `json:"name"` - Description *string `json:"description,omitempty"` - ActionType *int32 `json:"action_type,omitempty"` - DesiredState *int32 `json:"desired_state,omitempty"` - Params json.RawMessage `json:"params,omitempty"` - TimeoutSeconds *int32 `json:"timeout_seconds,omitempty"` - IsSystem *bool `json:"is_system,omitempty"` - Schedule json.RawMessage `json:"schedule,omitempty"` -} - // ActionCreatedFromEvent decodes ActionCreated. Returns ErrIgnoredEvent // for any other (stream, event_type) so the listener wrapper can // silently no-op. @@ -69,7 +59,7 @@ func ActionCreatedFromEvent(e store.PersistedEvent) (ActionCreatedPayload, error if len(e.Data) == 0 { return ActionCreatedPayload{}, fmt.Errorf("projector: empty ActionCreated payload") } - var raw actionCreatedRaw + var raw payloads.ActionCreated if err := json.Unmarshal(e.Data, &raw); err != nil { return ActionCreatedPayload{}, fmt.Errorf("projector: invalid ActionCreated payload: %w", err) } @@ -116,10 +106,6 @@ type ActionRenamedPayload struct { Name string } -type actionRenamedRaw struct { - Name string `json:"name"` -} - // ActionRenamedFromEvent decodes ActionRenamed. func ActionRenamedFromEvent(e store.PersistedEvent) (ActionRenamedPayload, error) { if e.StreamType != "action" || e.EventType != string(eventtypes.ActionRenamed) { @@ -128,7 +114,7 @@ func ActionRenamedFromEvent(e store.PersistedEvent) (ActionRenamedPayload, error if len(e.Data) == 0 { return ActionRenamedPayload{}, fmt.Errorf("projector: empty ActionRenamed payload") } - var raw actionRenamedRaw + var raw payloads.ActionRenamed if err := json.Unmarshal(e.Data, &raw); err != nil { return ActionRenamedPayload{}, fmt.Errorf("projector: invalid ActionRenamed payload: %w", err) } @@ -147,10 +133,6 @@ type ActionDescriptionUpdatedPayload struct { Description *string } -type actionDescriptionUpdatedRaw struct { - Description *string `json:"description,omitempty"` -} - // ActionDescriptionUpdatedFromEvent decodes ActionDescriptionUpdated. func ActionDescriptionUpdatedFromEvent(e store.PersistedEvent) (ActionDescriptionUpdatedPayload, error) { if e.StreamType != "action" || e.EventType != string(eventtypes.ActionDescriptionUpdated) { @@ -160,7 +142,7 @@ func ActionDescriptionUpdatedFromEvent(e store.PersistedEvent) (ActionDescriptio if len(e.Data) == 0 { return out, nil } - var raw actionDescriptionUpdatedRaw + var raw payloads.ActionDescriptionUpdated if err := json.Unmarshal(e.Data, &raw); err != nil { return ActionDescriptionUpdatedPayload{}, fmt.Errorf("projector: invalid ActionDescriptionUpdated payload: %w", err) } @@ -181,13 +163,6 @@ type ActionParamsUpdatedPayload struct { Schedule []byte } -type actionParamsUpdatedRaw struct { - Params json.RawMessage `json:"params,omitempty"` - TimeoutSeconds *int32 `json:"timeout_seconds,omitempty"` - DesiredState *int32 `json:"desired_state,omitempty"` - Schedule json.RawMessage `json:"schedule,omitempty"` -} - // ActionParamsUpdatedFromEvent decodes ActionParamsUpdated. func ActionParamsUpdatedFromEvent(e store.PersistedEvent) (ActionParamsUpdatedPayload, error) { if e.StreamType != "action" || e.EventType != string(eventtypes.ActionParamsUpdated) { @@ -197,7 +172,7 @@ func ActionParamsUpdatedFromEvent(e store.PersistedEvent) (ActionParamsUpdatedPa if len(e.Data) == 0 { return out, nil } - var raw actionParamsUpdatedRaw + var raw payloads.ActionParamsUpdated if err := json.Unmarshal(e.Data, &raw); err != nil { return ActionParamsUpdatedPayload{}, fmt.Errorf("projector: invalid ActionParamsUpdated payload: %w", err) } diff --git a/internal/projectors/assignment.go b/internal/projectors/assignment.go index 3243da99..4ac1002d 100644 --- a/internal/projectors/assignment.go +++ b/internal/projectors/assignment.go @@ -5,6 +5,7 @@ import ( "fmt" "github.com/manchtools/power-manage/server/internal/eventtypes" + "github.com/manchtools/power-manage/server/internal/eventtypes/payloads" "github.com/manchtools/power-manage/server/internal/store" ) @@ -28,15 +29,6 @@ type AssignmentCreatedPayload struct { CreatedBy string } -type assignmentCreatedRaw struct { - SourceType string `json:"source_type"` - SourceID string `json:"source_id"` - TargetType string `json:"target_type"` - TargetID string `json:"target_id"` - SortOrder *int32 `json:"sort_order,omitempty"` - Mode *int32 `json:"mode,omitempty"` -} - // AssignmentCreatedFromEvent decodes AssignmentCreated. Returns // ErrIgnoredEvent for any other (stream, event_type) so the listener // wrapper can silently no-op. @@ -54,7 +46,7 @@ func AssignmentCreatedFromEvent(e store.PersistedEvent) (AssignmentCreatedPayloa if len(e.Data) == 0 { return AssignmentCreatedPayload{}, fmt.Errorf("projector: empty AssignmentCreated payload") } - var raw assignmentCreatedRaw + var raw payloads.AssignmentCreated if err := json.Unmarshal(e.Data, &raw); err != nil { return AssignmentCreatedPayload{}, fmt.Errorf("projector: invalid AssignmentCreated payload: %w", err) } @@ -99,10 +91,6 @@ type AssignmentModeChangedPayload struct { Mode int32 } -type assignmentModeChangedRaw struct { - Mode *int32 `json:"mode,omitempty"` -} - // AssignmentModeChangedFromEvent decodes AssignmentModeChanged. func AssignmentModeChangedFromEvent(e store.PersistedEvent) (AssignmentModeChangedPayload, error) { if e.StreamType != "assignment" || e.EventType != string(eventtypes.AssignmentModeChanged) { @@ -112,7 +100,7 @@ func AssignmentModeChangedFromEvent(e store.PersistedEvent) (AssignmentModeChang if len(e.Data) == 0 { return out, nil } - var raw assignmentModeChangedRaw + var raw payloads.AssignmentModeChanged if err := json.Unmarshal(e.Data, &raw); err != nil { return AssignmentModeChangedPayload{}, fmt.Errorf("projector: invalid AssignmentModeChanged payload: %w", err) } @@ -133,10 +121,6 @@ type AssignmentSortOrderChangedPayload struct { SortOrder int32 } -type assignmentSortOrderChangedRaw struct { - SortOrder *int32 `json:"sort_order,omitempty"` -} - // AssignmentSortOrderChangedFromEvent decodes // AssignmentSortOrderChanged. func AssignmentSortOrderChangedFromEvent(e store.PersistedEvent) (AssignmentSortOrderChangedPayload, error) { @@ -147,7 +131,7 @@ func AssignmentSortOrderChangedFromEvent(e store.PersistedEvent) (AssignmentSort if len(e.Data) == 0 { return out, nil } - var raw assignmentSortOrderChangedRaw + var raw payloads.AssignmentSortOrderChanged if err := json.Unmarshal(e.Data, &raw); err != nil { return AssignmentSortOrderChangedPayload{}, fmt.Errorf("projector: invalid AssignmentSortOrderChanged payload: %w", err) } diff --git a/internal/projectors/device.go b/internal/projectors/device.go index 22c73dcb..6726f3ff 100644 --- a/internal/projectors/device.go +++ b/internal/projectors/device.go @@ -6,6 +6,7 @@ import ( "time" "github.com/manchtools/power-manage/server/internal/eventtypes" + "github.com/manchtools/power-manage/server/internal/eventtypes/payloads" "github.com/manchtools/power-manage/server/internal/store" ) @@ -41,15 +42,6 @@ type DeviceRegisteredPayload struct { AssignedUserID *string } -type deviceRegisteredRaw struct { - Hostname *string `json:"hostname,omitempty"` - CertFingerprint *string `json:"cert_fingerprint,omitempty"` - CertNotAfter *time.Time `json:"cert_not_after,omitempty"` - RegistrationTokenID *string `json:"registration_token_id,omitempty"` - Labels json.RawMessage `json:"labels,omitempty"` - AssignedUserID *string `json:"assigned_user_id,omitempty"` -} - // DeviceRegisteredFromEvent decodes DeviceRegistered. Returns // ErrIgnoredEvent for any other (stream, event_type) so the listener // wrapper can silently no-op. @@ -60,18 +52,22 @@ func DeviceRegisteredFromEvent(e store.PersistedEvent) (DeviceRegisteredPayload, if len(e.Data) == 0 { return DeviceRegisteredPayload{}, fmt.Errorf("projector: empty DeviceRegistered payload") } - var raw deviceRegisteredRaw + var raw payloads.DeviceRegistered if err := json.Unmarshal(e.Data, &raw); err != nil { return DeviceRegisteredPayload{}, fmt.Errorf("projector: invalid DeviceRegistered payload: %w", err) } out := DeviceRegisteredPayload{ ID: e.StreamID, CertFingerprint: raw.CertFingerprint, - CertNotAfter: raw.CertNotAfter, RegistrationTokenID: raw.RegistrationTokenID, AssignedUserID: raw.AssignedUserID, Labels: defaultDeviceLabels, } + notAfter, err := parseOptionalRFC3339(raw.CertNotAfter) + if err != nil { + return DeviceRegisteredPayload{}, fmt.Errorf("projector: invalid cert_not_after on DeviceRegistered: %w", err) + } + out.CertNotAfter = notAfter if raw.Hostname != nil { out.Hostname = *raw.Hostname } @@ -84,6 +80,28 @@ func DeviceRegisteredFromEvent(e store.PersistedEvent) (DeviceRegisteredPayload, return out, nil } +// parseOptionalRFC3339 parses an RFC 3339 (or RFC 3339Nano) timestamp +// string into a *time.Time, treating nil and empty as "absent" (returns +// nil with no error). Centralised here because every cert-related +// payload that emits as a string-formatted timestamp needs the same +// dual-format tolerant parse on the way back into a column. +func parseOptionalRFC3339(s *string) (*time.Time, error) { + if s == nil || *s == "" { + return nil, nil + } + t, err := time.Parse(time.RFC3339, *s) + if err != nil { + // Fall back to RFC 3339Nano so an emitter that switches + // formatters in the future doesn't silently corrupt + // downstream time arithmetic. + t, err = time.Parse(time.RFC3339Nano, *s) + if err != nil { + return nil, err + } + } + return &t, nil +} + // DeviceSeenPayload mirrors the PL/pgSQL projector's COALESCE-on- // missing semantics for DeviceSeen: // @@ -101,11 +119,6 @@ type DeviceSeenPayload struct { Hostname *string } -type deviceSeenRaw struct { - AgentVersion *string `json:"agent_version,omitempty"` - Hostname *string `json:"hostname,omitempty"` -} - // DeviceSeenFromEvent decodes DeviceSeen. Empty payload is valid (a // pure heartbeat-style ping that only refreshes last_seen_at). func DeviceSeenFromEvent(e store.PersistedEvent) (DeviceSeenPayload, error) { @@ -116,7 +129,7 @@ func DeviceSeenFromEvent(e store.PersistedEvent) (DeviceSeenPayload, error) { if len(e.Data) == 0 { return out, nil } - var raw deviceSeenRaw + var raw payloads.DeviceSeen if err := json.Unmarshal(e.Data, &raw); err != nil { return DeviceSeenPayload{}, fmt.Errorf("projector: invalid DeviceSeen payload: %w", err) } @@ -140,10 +153,6 @@ type DeviceHeartbeatPayload struct { AgentVersion *string } -type deviceHeartbeatRaw struct { - AgentVersion *string `json:"agent_version,omitempty"` -} - // DeviceHeartbeatFromEvent decodes DeviceHeartbeat. Empty payload is // a valid bare ping. func DeviceHeartbeatFromEvent(e store.PersistedEvent) (DeviceHeartbeatPayload, error) { @@ -154,7 +163,7 @@ func DeviceHeartbeatFromEvent(e store.PersistedEvent) (DeviceHeartbeatPayload, e if len(e.Data) == 0 { return out, nil } - var raw deviceHeartbeatRaw + var raw payloads.DeviceHeartbeat if err := json.Unmarshal(e.Data, &raw); err != nil { return DeviceHeartbeatPayload{}, fmt.Errorf("projector: invalid DeviceHeartbeat payload: %w", err) } @@ -172,11 +181,6 @@ type DeviceCertRenewedPayload struct { CertNotAfter *time.Time } -type deviceCertRenewedRaw struct { - CertFingerprint *string `json:"cert_fingerprint,omitempty"` - CertNotAfter *time.Time `json:"cert_not_after,omitempty"` -} - // DeviceCertRenewedFromEvent decodes DeviceCertRenewed. func DeviceCertRenewedFromEvent(e store.PersistedEvent) (DeviceCertRenewedPayload, error) { if e.StreamType != "device" || e.EventType != string(eventtypes.DeviceCertRenewed) { @@ -185,17 +189,21 @@ func DeviceCertRenewedFromEvent(e store.PersistedEvent) (DeviceCertRenewedPayloa if len(e.Data) == 0 { return DeviceCertRenewedPayload{}, fmt.Errorf("projector: empty DeviceCertRenewed payload") } - var raw deviceCertRenewedRaw + var raw payloads.DeviceCertRenewed if err := json.Unmarshal(e.Data, &raw); err != nil { return DeviceCertRenewedPayload{}, fmt.Errorf("projector: invalid DeviceCertRenewed payload: %w", err) } if raw.CertFingerprint == nil || *raw.CertFingerprint == "" { return DeviceCertRenewedPayload{}, fmt.Errorf("projector: DeviceCertRenewed requires cert_fingerprint") } + notAfter, err := parseOptionalRFC3339(raw.CertNotAfter) + if err != nil { + return DeviceCertRenewedPayload{}, fmt.Errorf("projector: invalid cert_not_after on DeviceCertRenewed: %w", err) + } return DeviceCertRenewedPayload{ ID: e.StreamID, CertFingerprint: *raw.CertFingerprint, - CertNotAfter: raw.CertNotAfter, + CertNotAfter: notAfter, }, nil } @@ -208,10 +216,6 @@ type DeviceLabelsUpdatedPayload struct { Labels []byte } -type deviceLabelsUpdatedRaw struct { - Labels json.RawMessage `json:"labels,omitempty"` -} - // DeviceLabelsUpdatedFromEvent decodes DeviceLabelsUpdated. Empty // labels => nil byte slice; the SQL COALESCE will preserve the // existing row value. @@ -223,7 +227,7 @@ func DeviceLabelsUpdatedFromEvent(e store.PersistedEvent) (DeviceLabelsUpdatedPa if len(e.Data) == 0 { return out, nil } - var raw deviceLabelsUpdatedRaw + var raw payloads.DeviceLabelsUpdated if err := json.Unmarshal(e.Data, &raw); err != nil { return DeviceLabelsUpdatedPayload{}, fmt.Errorf("projector: invalid DeviceLabelsUpdated payload: %w", err) } @@ -242,11 +246,6 @@ type DeviceLabelSetPayload struct { Value string } -type deviceLabelSetRaw struct { - Key *string `json:"key,omitempty"` - Value *string `json:"value,omitempty"` -} - // DeviceLabelSetFromEvent decodes DeviceLabelSet. func DeviceLabelSetFromEvent(e store.PersistedEvent) (DeviceLabelSetPayload, error) { if e.StreamType != "device" || e.EventType != string(eventtypes.DeviceLabelSet) { @@ -255,7 +254,7 @@ func DeviceLabelSetFromEvent(e store.PersistedEvent) (DeviceLabelSetPayload, err if len(e.Data) == 0 { return DeviceLabelSetPayload{}, fmt.Errorf("projector: empty DeviceLabelSet payload") } - var raw deviceLabelSetRaw + var raw payloads.DeviceLabelSet if err := json.Unmarshal(e.Data, &raw); err != nil { return DeviceLabelSetPayload{}, fmt.Errorf("projector: invalid DeviceLabelSet payload: %w", err) } @@ -276,10 +275,6 @@ type DeviceLabelRemovedPayload struct { Key string } -type deviceLabelRemovedRaw struct { - Key *string `json:"key,omitempty"` -} - // DeviceLabelRemovedFromEvent decodes DeviceLabelRemoved. func DeviceLabelRemovedFromEvent(e store.PersistedEvent) (DeviceLabelRemovedPayload, error) { if e.StreamType != "device" || e.EventType != string(eventtypes.DeviceLabelRemoved) { @@ -288,7 +283,7 @@ func DeviceLabelRemovedFromEvent(e store.PersistedEvent) (DeviceLabelRemovedPayl if len(e.Data) == 0 { return DeviceLabelRemovedPayload{}, fmt.Errorf("projector: empty DeviceLabelRemoved payload") } - var raw deviceLabelRemovedRaw + var raw payloads.DeviceLabelRemoved if err := json.Unmarshal(e.Data, &raw); err != nil { return DeviceLabelRemovedPayload{}, fmt.Errorf("projector: invalid DeviceLabelRemoved payload: %w", err) } @@ -309,10 +304,6 @@ type DeviceUserAssignmentPayload struct { UserID string } -type deviceUserAssignmentRaw struct { - UserID *string `json:"user_id,omitempty"` -} - // DeviceAssignedFromEvent decodes DeviceAssigned. func DeviceAssignedFromEvent(e store.PersistedEvent) (DeviceUserAssignmentPayload, error) { if e.StreamType != "device" || e.EventType != string(eventtypes.DeviceAssigned) { @@ -333,7 +324,7 @@ func decodeDeviceUserAssignment(e store.PersistedEvent) (DeviceUserAssignmentPay if len(e.Data) == 0 { return DeviceUserAssignmentPayload{}, fmt.Errorf("projector: empty %s payload", e.EventType) } - var raw deviceUserAssignmentRaw + var raw payloads.DeviceUserAssignment if err := json.Unmarshal(e.Data, &raw); err != nil { return DeviceUserAssignmentPayload{}, fmt.Errorf("projector: invalid %s payload: %w", e.EventType, err) } @@ -351,10 +342,6 @@ type DeviceGroupAssignmentPayload struct { GroupID string } -type deviceGroupAssignmentRaw struct { - GroupID *string `json:"group_id,omitempty"` -} - // DeviceGroupAssignedFromEvent decodes DeviceGroupAssigned. func DeviceGroupAssignedFromEvent(e store.PersistedEvent) (DeviceGroupAssignmentPayload, error) { if e.StreamType != "device" || e.EventType != string(eventtypes.DeviceGroupAssigned) { @@ -375,7 +362,7 @@ func decodeDeviceGroupAssignment(e store.PersistedEvent) (DeviceGroupAssignmentP if len(e.Data) == 0 { return DeviceGroupAssignmentPayload{}, fmt.Errorf("projector: empty %s payload", e.EventType) } - var raw deviceGroupAssignmentRaw + var raw payloads.DeviceGroupAssignment if err := json.Unmarshal(e.Data, &raw); err != nil { return DeviceGroupAssignmentPayload{}, fmt.Errorf("projector: invalid %s payload: %w", e.EventType, err) } @@ -393,10 +380,6 @@ type DeviceSyncIntervalSetPayload struct { SyncIntervalMinutes int32 } -type deviceSyncIntervalSetRaw struct { - SyncIntervalMinutes *int32 `json:"sync_interval_minutes,omitempty"` -} - // DeviceSyncIntervalSetFromEvent decodes DeviceSyncIntervalSet. func DeviceSyncIntervalSetFromEvent(e store.PersistedEvent) (DeviceSyncIntervalSetPayload, error) { if e.StreamType != "device" || e.EventType != string(eventtypes.DeviceSyncIntervalSet) { @@ -406,7 +389,7 @@ func DeviceSyncIntervalSetFromEvent(e store.PersistedEvent) (DeviceSyncIntervalS if len(e.Data) == 0 { return out, nil } - var raw deviceSyncIntervalSetRaw + var raw payloads.DeviceSyncIntervalSet if err := json.Unmarshal(e.Data, &raw); err != nil { return DeviceSyncIntervalSetPayload{}, fmt.Errorf("projector: invalid DeviceSyncIntervalSet payload: %w", err) } diff --git a/internal/projectors/execution.go b/internal/projectors/execution.go index 09ef94de..a4fd860f 100644 --- a/internal/projectors/execution.go +++ b/internal/projectors/execution.go @@ -6,6 +6,7 @@ import ( "time" "github.com/manchtools/power-manage/server/internal/eventtypes" + "github.com/manchtools/power-manage/server/internal/eventtypes/payloads" "github.com/manchtools/power-manage/server/internal/store" ) @@ -49,17 +50,6 @@ type ExecutionCreatedPayload struct { CreatedByID string } -type executionCreatedRaw struct { - DeviceID string `json:"device_id"` - ActionID *string `json:"action_id,omitempty"` - DefinitionID *string `json:"definition_id,omitempty"` - ActionType *int32 `json:"action_type,omitempty"` - DesiredState *int32 `json:"desired_state,omitempty"` - Params json.RawMessage `json:"params,omitempty"` - TimeoutSeconds *int32 `json:"timeout_seconds,omitempty"` - ExecutedAt *string `json:"executed_at,omitempty"` -} - // ExecutionCreatedFromEvent decodes ExecutionCreated. Returns // ErrIgnoredEvent for any other (stream, event_type) so the listener // wrapper can silently no-op. @@ -70,7 +60,7 @@ func ExecutionCreatedFromEvent(e store.PersistedEvent) (ExecutionCreatedPayload, if len(e.Data) == 0 { return ExecutionCreatedPayload{}, fmt.Errorf("projector: empty ExecutionCreated payload") } - var raw executionCreatedRaw + var raw payloads.ExecutionCreated if err := json.Unmarshal(e.Data, &raw); err != nil { return ExecutionCreatedPayload{}, fmt.Errorf("projector: invalid ExecutionCreated payload: %w", err) } @@ -125,17 +115,6 @@ type ExecutionScheduledPayload struct { CreatedByID string } -type executionScheduledRaw struct { - DeviceID string `json:"device_id"` - ActionID *string `json:"action_id,omitempty"` - DefinitionID *string `json:"definition_id,omitempty"` - ActionType *int32 `json:"action_type,omitempty"` - DesiredState *int32 `json:"desired_state,omitempty"` - Params json.RawMessage `json:"params,omitempty"` - TimeoutSeconds *int32 `json:"timeout_seconds,omitempty"` - ScheduledFor string `json:"scheduled_for"` -} - // ExecutionScheduledFromEvent decodes ExecutionScheduled. scheduled_for // is REQUIRED — the PL/pgSQL projector cast it directly without a // COALESCE, so an absent key would have produced a NULL column write @@ -149,7 +128,7 @@ func ExecutionScheduledFromEvent(e store.PersistedEvent) (ExecutionScheduledPayl if len(e.Data) == 0 { return ExecutionScheduledPayload{}, fmt.Errorf("projector: empty ExecutionScheduled payload") } - var raw executionScheduledRaw + var raw payloads.ExecutionScheduled if err := json.Unmarshal(e.Data, &raw); err != nil { return ExecutionScheduledPayload{}, fmt.Errorf("projector: invalid ExecutionScheduled payload: %w", err) } @@ -208,16 +187,6 @@ type ExecutionTerminalPayload struct { DetectionOutput []byte } -type executionTerminalRaw struct { - CompletedAt *string `json:"completed_at,omitempty"` - Error *string `json:"error,omitempty"` - Output json.RawMessage `json:"output,omitempty"` - DurationMs *int64 `json:"duration_ms,omitempty"` - Changed *bool `json:"changed,omitempty"` - Compliant *bool `json:"compliant,omitempty"` - DetectionOutput json.RawMessage `json:"detection_output,omitempty"` -} - // ExecutionCompletedFromEvent decodes ExecutionCompleted. func ExecutionCompletedFromEvent(e store.PersistedEvent) (ExecutionTerminalPayload, error) { if e.StreamType != "execution" || e.EventType != string(eventtypes.ExecutionCompleted) { @@ -246,7 +215,7 @@ func decodeTerminal(e store.PersistedEvent, label string) (ExecutionTerminalPayl if len(e.Data) == 0 { return out, nil } - var raw executionTerminalRaw + var raw payloads.ExecutionTerminal if err := json.Unmarshal(e.Data, &raw); err != nil { return ExecutionTerminalPayload{}, fmt.Errorf("projector: invalid %s payload: %w", label, err) } @@ -283,13 +252,6 @@ type ExecutionTimedOutPayload struct { DurationMs *int64 } -type executionTimedOutRaw struct { - CompletedAt *string `json:"completed_at,omitempty"` - Error *string `json:"error,omitempty"` - Output json.RawMessage `json:"output,omitempty"` - DurationMs *int64 `json:"duration_ms,omitempty"` -} - // ExecutionTimedOutFromEvent decodes ExecutionTimedOut. func ExecutionTimedOutFromEvent(e store.PersistedEvent) (ExecutionTimedOutPayload, error) { if e.StreamType != "execution" || e.EventType != string(eventtypes.ExecutionTimedOut) { @@ -302,7 +264,7 @@ func ExecutionTimedOutFromEvent(e store.PersistedEvent) (ExecutionTimedOutPayloa if len(e.Data) == 0 { return out, nil } - var raw executionTimedOutRaw + var raw payloads.ExecutionTimedOut if err := json.Unmarshal(e.Data, &raw); err != nil { return ExecutionTimedOutPayload{}, fmt.Errorf("projector: invalid ExecutionTimedOut payload: %w", err) } @@ -329,10 +291,6 @@ type ExecutionReasonPayload struct { Reason *string } -type executionReasonRaw struct { - Reason *string `json:"reason,omitempty"` -} - // ExecutionSkippedFromEvent decodes ExecutionSkipped. func ExecutionSkippedFromEvent(e store.PersistedEvent) (ExecutionReasonPayload, error) { if e.StreamType != "execution" || e.EventType != string(eventtypes.ExecutionSkipped) { @@ -357,7 +315,7 @@ func decodeReason(e store.PersistedEvent, label string) (ExecutionReasonPayload, if len(e.Data) == 0 { return out, nil } - var raw executionReasonRaw + var raw payloads.ExecutionReason if err := json.Unmarshal(e.Data, &raw); err != nil { return ExecutionReasonPayload{}, fmt.Errorf("projector: invalid %s payload: %w", label, err) } diff --git a/internal/projectors/lps_password.go b/internal/projectors/lps_password.go index 11d3be75..af716ca9 100644 --- a/internal/projectors/lps_password.go +++ b/internal/projectors/lps_password.go @@ -3,49 +3,17 @@ package projectors import ( "encoding/json" "fmt" - "log/slog" - "time" "github.com/manchtools/power-manage/server/internal/eventtypes" + "github.com/manchtools/power-manage/server/internal/eventtypes/payloads" "github.com/manchtools/power-manage/server/internal/store" ) -// LpsPasswordRotatedPayload covers every field the lps_password -// projector reads from the LpsPasswordRotated event. The agent sends -// rotated_at as an RFC 3339 string (see sdk/proto/pm/v1/internal.proto -// LpsRotation.rotated_at), so json.Unmarshal into time.Time round-trips -// natively. -// -// The deleted PL/pgSQL projector cast `(event.data->>'rotated_at')::TIMESTAMPTZ` -// directly. Pre-parsing in Go preserves the same shape: a parse error -// here is the analogue of a Postgres cast failure, which the PL/pgSQL -// version would have surfaced via the plpgsql_projection_errors table. -type LpsPasswordRotatedPayload struct { - DeviceID string `json:"device_id"` - ActionID string `json:"action_id"` - Username string `json:"username"` - Password string `json:"password"` - RotatedAt time.Time `json:"rotated_at"` - RotationReason string `json:"rotation_reason"` -} - -// LogValue implements slog.LogValuer so the encrypted Password is -// never written verbatim by structured logs. The listener body -// already logs only individual non-secret fields, but a future -// `logger.Warn("…", "payload", payload)` or `fmt.Sprintf("%+v", p)` -// routed through slog would otherwise leak the credential. Mask at -// the type level so the safety holds regardless of caller -// discipline. -func (p LpsPasswordRotatedPayload) LogValue() slog.Value { - return slog.GroupValue( - slog.String("device_id", p.DeviceID), - slog.String("action_id", p.ActionID), - slog.String("username", p.Username), - slog.String("password", "[REDACTED]"), - slog.Time("rotated_at", p.RotatedAt), - slog.String("rotation_reason", p.RotationReason), - ) -} +// LpsPasswordRotatedPayload aliases the shared wire struct so existing +// projector callers keep their import + symbol. The Payload-suffix name +// stays for projector-side code; the bare payloads.LpsPasswordRotated +// is the canonical handle for handler emit sites. +type LpsPasswordRotatedPayload = payloads.LpsPasswordRotated // LpsPasswordRotatedFromEvent decodes the event payload into the // typed shape the listener writes. Returns ErrIgnoredEvent for any diff --git a/internal/projectors/luks_key.go b/internal/projectors/luks_key.go index f14b40ac..70f8b5c7 100644 --- a/internal/projectors/luks_key.go +++ b/internal/projectors/luks_key.go @@ -3,45 +3,17 @@ package projectors import ( "encoding/json" "fmt" - "log/slog" "time" "github.com/manchtools/power-manage/server/internal/eventtypes" + "github.com/manchtools/power-manage/server/internal/eventtypes/payloads" "github.com/manchtools/power-manage/server/internal/store" ) -// LuksKeyRotatedPayload covers every field the luks_key projector -// reads from the LuksKeyRotated event. Same shape as -// LpsPasswordRotatedPayload but with `device_path` in the partition -// key — LUKS rotates per (device, action, device_path) where LPS -// rotates per (device, username). -type LuksKeyRotatedPayload struct { - DeviceID string `json:"device_id"` - ActionID string `json:"action_id"` - DevicePath string `json:"device_path"` - Passphrase string `json:"passphrase"` - RotatedAt time.Time `json:"rotated_at"` - RotationReason string `json:"rotation_reason"` -} - -// LogValue implements slog.LogValuer so the encrypted Passphrase is -// never written verbatim by structured logs. Mirrors the -// LpsPasswordRotatedPayload masking — the listener body already logs -// only individual non-secret fields, but a future -// `logger.Warn("…", "payload", payload)` or `fmt.Sprintf("%+v", p)` -// routed through slog would otherwise leak the credential. Mask at -// the type level so the safety holds regardless of caller -// discipline. -func (p LuksKeyRotatedPayload) LogValue() slog.Value { - return slog.GroupValue( - slog.String("device_id", p.DeviceID), - slog.String("action_id", p.ActionID), - slog.String("device_path", p.DevicePath), - slog.String("passphrase", "[REDACTED]"), - slog.Time("rotated_at", p.RotatedAt), - slog.String("rotation_reason", p.RotationReason), - ) -} +// LuksKeyRotatedPayload aliases the shared wire struct so projector +// callers keep their import + symbol; payloads.LuksKeyRotated is the +// canonical handle for handler emit sites. +type LuksKeyRotatedPayload = payloads.LuksKeyRotated // LuksRevocationPayload is the union shape for the three revocation // event types (Dispatched, Revoked, Failed). They all UPDATE the diff --git a/internal/projectors/user.go b/internal/projectors/user.go index 71e4ab47..388f2f65 100644 --- a/internal/projectors/user.go +++ b/internal/projectors/user.go @@ -6,6 +6,7 @@ import ( "time" "github.com/manchtools/power-manage/server/internal/eventtypes" + "github.com/manchtools/power-manage/server/internal/eventtypes/payloads" "github.com/manchtools/power-manage/server/internal/store" ) @@ -58,21 +59,6 @@ type UserCreatedWithRolesPayload struct { RoleIDs []string } -type userCreatedWithRolesRaw struct { - Email *string `json:"email,omitempty"` - PasswordHash *string `json:"password_hash,omitempty"` - Role *string `json:"role,omitempty"` - DisplayName *string `json:"display_name,omitempty"` - GivenName *string `json:"given_name,omitempty"` - FamilyName *string `json:"family_name,omitempty"` - PreferredUsername *string `json:"preferred_username,omitempty"` - Picture *string `json:"picture,omitempty"` - Locale *string `json:"locale,omitempty"` - LinuxUsername *string `json:"linux_username,omitempty"` - LinuxUID *int32 `json:"linux_uid,omitempty"` - RoleIDs []string `json:"role_ids,omitempty"` -} - // UserCreatedWithRolesFromEvent decodes UserCreatedWithRoles. Returns // ErrIgnoredEvent for any other (stream, event_type) so the listener // wrapper can silently no-op. @@ -83,7 +69,7 @@ func UserCreatedWithRolesFromEvent(e store.PersistedEvent) (UserCreatedWithRoles if len(e.Data) == 0 { return UserCreatedWithRolesPayload{}, fmt.Errorf("projector: empty UserCreatedWithRoles payload") } - var raw userCreatedWithRolesRaw + var raw payloads.UserCreatedWithRoles if err := json.Unmarshal(e.Data, &raw); err != nil { return UserCreatedWithRolesPayload{}, fmt.Errorf("projector: invalid UserCreatedWithRoles payload: %w", err) } @@ -153,15 +139,6 @@ type UserProfileUpdatedPayload struct { Locale string } -type userProfileUpdatedRaw struct { - DisplayName *string `json:"display_name,omitempty"` - GivenName *string `json:"given_name,omitempty"` - FamilyName *string `json:"family_name,omitempty"` - PreferredUsername *string `json:"preferred_username,omitempty"` - Picture *string `json:"picture,omitempty"` - Locale *string `json:"locale,omitempty"` -} - // UserProfileUpdatedFromEvent decodes UserProfileUpdated. func UserProfileUpdatedFromEvent(e store.PersistedEvent) (UserProfileUpdatedPayload, error) { if e.StreamType != "user" || e.EventType != string(eventtypes.UserProfileUpdated) { @@ -171,7 +148,7 @@ func UserProfileUpdatedFromEvent(e store.PersistedEvent) (UserProfileUpdatedPayl if len(e.Data) == 0 { return out, nil } - var raw userProfileUpdatedRaw + var raw payloads.UserProfileUpdated if err := json.Unmarshal(e.Data, &raw); err != nil { return UserProfileUpdatedPayload{}, fmt.Errorf("projector: invalid UserProfileUpdated payload: %w", err) } @@ -203,10 +180,6 @@ type UserEmailChangedPayload struct { Email string } -type userEmailChangedRaw struct { - Email *string `json:"email,omitempty"` -} - // UserEmailChangedFromEvent decodes UserEmailChanged. The PL/pgSQL // projector wrote `event.data->>"email"` directly — if the key was // missing the column would land as SQL NULL, but the column is @@ -219,7 +192,7 @@ func UserEmailChangedFromEvent(e store.PersistedEvent) (UserEmailChangedPayload, if len(e.Data) == 0 { return UserEmailChangedPayload{}, fmt.Errorf("projector: empty UserEmailChanged payload") } - var raw userEmailChangedRaw + var raw payloads.UserEmailChanged if err := json.Unmarshal(e.Data, &raw); err != nil { return UserEmailChangedPayload{}, fmt.Errorf("projector: invalid UserEmailChanged payload: %w", err) } @@ -241,10 +214,6 @@ type UserPasswordChangedPayload struct { PasswordHash string } -type userPasswordChangedRaw struct { - PasswordHash *string `json:"password_hash,omitempty"` -} - // UserPasswordChangedFromEvent decodes UserPasswordChanged. func UserPasswordChangedFromEvent(e store.PersistedEvent) (UserPasswordChangedPayload, error) { if e.StreamType != "user" || e.EventType != string(eventtypes.UserPasswordChanged) { @@ -253,7 +222,7 @@ func UserPasswordChangedFromEvent(e store.PersistedEvent) (UserPasswordChangedPa if len(e.Data) == 0 { return UserPasswordChangedPayload{}, fmt.Errorf("projector: empty UserPasswordChanged payload") } - var raw userPasswordChangedRaw + var raw payloads.UserPasswordChanged if err := json.Unmarshal(e.Data, &raw); err != nil { return UserPasswordChangedPayload{}, fmt.Errorf("projector: invalid UserPasswordChanged payload: %w", err) } @@ -270,10 +239,6 @@ type UserRoleChangedPayload struct { Role string } -type userRoleChangedRaw struct { - Role *string `json:"role,omitempty"` -} - // UserRoleChangedFromEvent decodes UserRoleChanged. func UserRoleChangedFromEvent(e store.PersistedEvent) (UserRoleChangedPayload, error) { if e.StreamType != "user" || e.EventType != string(eventtypes.UserRoleChanged) { @@ -282,7 +247,7 @@ func UserRoleChangedFromEvent(e store.PersistedEvent) (UserRoleChangedPayload, e if len(e.Data) == 0 { return UserRoleChangedPayload{}, fmt.Errorf("projector: empty UserRoleChanged payload") } - var raw userRoleChangedRaw + var raw payloads.UserRoleChanged if err := json.Unmarshal(e.Data, &raw); err != nil { return UserRoleChangedPayload{}, fmt.Errorf("projector: invalid UserRoleChanged payload: %w", err) } @@ -311,12 +276,6 @@ type UserSshKeyAddedPayload struct { AddedAt time.Time } -type userSshKeyAddedRaw struct { - KeyID *string `json:"key_id,omitempty"` - PublicKey *string `json:"public_key,omitempty"` - Comment *string `json:"comment,omitempty"` -} - // UserSshKeyAddedFromEvent decodes UserSshKeyAdded. key_id is // required because the JSONB element is the only addressable handle // for the matching UserSshKeyRemoved event. public_key + comment @@ -329,7 +288,7 @@ func UserSshKeyAddedFromEvent(e store.PersistedEvent) (UserSshKeyAddedPayload, e if len(e.Data) == 0 { return UserSshKeyAddedPayload{}, fmt.Errorf("projector: empty UserSshKeyAdded payload") } - var raw userSshKeyAddedRaw + var raw payloads.UserSshKeyAdded if err := json.Unmarshal(e.Data, &raw); err != nil { return UserSshKeyAddedPayload{}, fmt.Errorf("projector: invalid UserSshKeyAdded payload: %w", err) } @@ -355,10 +314,6 @@ type UserSshKeyRemovedPayload struct { KeyID string } -type userSshKeyRemovedRaw struct { - KeyID *string `json:"key_id,omitempty"` -} - // UserSshKeyRemovedFromEvent decodes UserSshKeyRemoved. func UserSshKeyRemovedFromEvent(e store.PersistedEvent) (UserSshKeyRemovedPayload, error) { if e.StreamType != "user" || e.EventType != string(eventtypes.UserSshKeyRemoved) { @@ -367,7 +322,7 @@ func UserSshKeyRemovedFromEvent(e store.PersistedEvent) (UserSshKeyRemovedPayloa if len(e.Data) == 0 { return UserSshKeyRemovedPayload{}, fmt.Errorf("projector: empty UserSshKeyRemoved payload") } - var raw userSshKeyRemovedRaw + var raw payloads.UserSshKeyRemoved if err := json.Unmarshal(e.Data, &raw); err != nil { return UserSshKeyRemovedPayload{}, fmt.Errorf("projector: invalid UserSshKeyRemoved payload: %w", err) } @@ -389,12 +344,6 @@ type UserSshSettingsUpdatedPayload struct { SshAllowPassword *bool } -type userSshSettingsUpdatedRaw struct { - SshAccessEnabled *bool `json:"ssh_access_enabled,omitempty"` - SshAllowPubkey *bool `json:"ssh_allow_pubkey,omitempty"` - SshAllowPassword *bool `json:"ssh_allow_password,omitempty"` -} - // UserSshSettingsUpdatedFromEvent decodes UserSshSettingsUpdated. // Empty payload is valid (a no-op event that only bumps // projection_version + updated_at, leaving every column as-is). @@ -406,7 +355,7 @@ func UserSshSettingsUpdatedFromEvent(e store.PersistedEvent) (UserSshSettingsUpd if len(e.Data) == 0 { return out, nil } - var raw userSshSettingsUpdatedRaw + var raw payloads.UserSshSettingsUpdated if err := json.Unmarshal(e.Data, &raw); err != nil { return UserSshSettingsUpdatedPayload{}, fmt.Errorf("projector: invalid UserSshSettingsUpdated payload: %w", err) } @@ -425,10 +374,6 @@ type UserLinuxUsernameChangedPayload struct { LinuxUsername string } -type userLinuxUsernameChangedRaw struct { - LinuxUsername *string `json:"linux_username,omitempty"` -} - // UserLinuxUsernameChangedFromEvent decodes UserLinuxUsernameChanged. func UserLinuxUsernameChangedFromEvent(e store.PersistedEvent) (UserLinuxUsernameChangedPayload, error) { if e.StreamType != "user" || e.EventType != string(eventtypes.UserLinuxUsernameChanged) { @@ -437,7 +382,7 @@ func UserLinuxUsernameChangedFromEvent(e store.PersistedEvent) (UserLinuxUsernam if len(e.Data) == 0 { return UserLinuxUsernameChangedPayload{}, fmt.Errorf("projector: empty UserLinuxUsernameChanged payload") } - var raw userLinuxUsernameChangedRaw + var raw payloads.UserLinuxUsernameChanged if err := json.Unmarshal(e.Data, &raw); err != nil { return UserLinuxUsernameChangedPayload{}, fmt.Errorf("projector: invalid UserLinuxUsernameChanged payload: %w", err) } @@ -458,11 +403,6 @@ type UserSystemActionLinkedPayload struct { ActionID string } -type userSystemActionLinkedRaw struct { - Field *string `json:"field,omitempty"` - ActionID *string `json:"action_id,omitempty"` -} - // Allowed field names for UserSystemActionLinked. Mirrors the // PL/pgSQL CASE arms exactly — anything else falls through to the // three "preserve existing" branches in the original projector, @@ -484,7 +424,7 @@ func UserSystemActionLinkedFromEvent(e store.PersistedEvent) (UserSystemActionLi if len(e.Data) == 0 { return UserSystemActionLinkedPayload{}, fmt.Errorf("projector: empty UserSystemActionLinked payload") } - var raw userSystemActionLinkedRaw + var raw payloads.UserSystemActionLinked if err := json.Unmarshal(e.Data, &raw); err != nil { return UserSystemActionLinkedPayload{}, fmt.Errorf("projector: invalid UserSystemActionLinked payload: %w", err) } @@ -510,10 +450,6 @@ type UserProvisioningSettingsUpdatedPayload struct { UserProvisioningEnabled *bool } -type userProvisioningSettingsUpdatedRaw struct { - UserProvisioningEnabled *bool `json:"user_provisioning_enabled,omitempty"` -} - // UserProvisioningSettingsUpdatedFromEvent decodes // UserProvisioningSettingsUpdated. Empty payload is valid (no-op // event that only bumps projection_version + updated_at). @@ -525,7 +461,7 @@ func UserProvisioningSettingsUpdatedFromEvent(e store.PersistedEvent) (UserProvi if len(e.Data) == 0 { return out, nil } - var raw userProvisioningSettingsUpdatedRaw + var raw payloads.UserProvisioningSettingsUpdated if err := json.Unmarshal(e.Data, &raw); err != nil { return UserProvisioningSettingsUpdatedPayload{}, fmt.Errorf("projector: invalid UserProvisioningSettingsUpdated payload: %w", err) } diff --git a/internal/store/store.go b/internal/store/store.go index 2297d1f6..3f34b94f 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -25,11 +25,18 @@ type Queries = generated.Queries type PersistedEvent = generated.Event // Event represents a domain event. +// +// Data accepts any JSON-marshalable value. Prefer a typed payload +// struct from internal/eventtypes/payloads — sharing one struct between +// the handler emit site and the projector decoder catches schema drift +// at compile time. The legacy map[string]any literal remains supported +// for not-yet-converted call sites; AppendEvent json.Marshals whatever +// is passed, so the wire format is unchanged. type Event struct { StreamType string StreamID string EventType string - Data map[string]any + Data any Metadata map[string]any ActorType string ActorID string