From 2ba81cfd8263df79540ef189842d436c92b52e57 Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Wed, 29 Apr 2026 00:47:45 -0400 Subject: [PATCH] feat(engine): typed event hierarchy + EventBus with fanout + AuditSubscriber + Engine Events form a Crystal class hierarchy so subscribers pattern-match exhaustively. EventBus delivers to subscribers via Crystal channels with per-subscriber overflow policy (Block for audit, Drop for best-effort). AuditSubscriber listens for all rotation/policy/drift/alert events and serializes them into the hash-chained audit log via AuditLog.append. Engine wires Persistence + AuditLog + AuditSubscriber + EventBus together so callers just .start/.stop. 7 unit specs verify boot, fanout to N subscribers, drop-on-overflow semantics, and end-to-end event -> audit row. --- .../spec/unit/engine/audit_subscriber_spec.cr | 62 +++++++++++ .../spec/unit/engine/engine_spec.cr | 42 ++++++++ .../spec/unit/engine/event_bus_spec.cr | 91 ++++++++++++++++ .../src/cre/engine/engine.cr | 49 +++++++++ .../src/cre/engine/event_bus.cr | 80 ++++++++++++++ .../engine/subscribers/audit_subscriber.cr | 85 +++++++++++++++ .../src/cre/events/credential_events.cr | 100 ++++++++++++++++++ .../src/cre/events/event.cr | 13 +++ .../src/cre/events/system_events.cr | 29 +++++ 9 files changed, 551 insertions(+) create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/engine/audit_subscriber_spec.cr create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/engine/engine_spec.cr create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/engine/event_bus_spec.cr create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/src/cre/engine/engine.cr create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/src/cre/engine/event_bus.cr create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/src/cre/engine/subscribers/audit_subscriber.cr create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/src/cre/events/credential_events.cr create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/src/cre/events/event.cr create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/src/cre/events/system_events.cr diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/engine/audit_subscriber_spec.cr b/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/engine/audit_subscriber_spec.cr new file mode 100644 index 00000000..8db93e9f --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/engine/audit_subscriber_spec.cr @@ -0,0 +1,62 @@ +# =================== +# ©AngelaMos | 2026 +# audit_subscriber_spec.cr +# =================== + +require "../../spec_helper" +require "../../../src/cre/engine/event_bus" +require "../../../src/cre/engine/subscribers/audit_subscriber" +require "../../../src/cre/audit/audit_log" +require "../../../src/cre/persistence/sqlite/sqlite_persistence" +require "../../../src/cre/events/credential_events" + +describe CRE::Engine::Subscribers::AuditSubscriber do + it "writes RotationCompleted events to the audit log" do + persist = CRE::Persistence::Sqlite::SqlitePersistence.new(":memory:") + persist.migrate! + log = CRE::Audit::AuditLog.new(persist, Bytes.new(32, 0_u8), 1, 1024) + + bus = CRE::Engine::EventBus.new + sub = CRE::Engine::Subscribers::AuditSubscriber.new(bus, log) + sub.start + bus.run + + cred_id = UUID.random + rot_id = UUID.random + bus.publish(CRE::Events::RotationCompleted.new(cred_id, rot_id)) + sleep 0.1.seconds + + persist.audit.latest_seq.should eq 1_i64 + entry = persist.audit.range(1_i64, 1_i64).first + entry.event_type.should eq "rotation.completed" + entry.target_id.should eq cred_id + ensure + bus.try(&.stop) + sub.try(&.stop) + persist.try(&.close) + end + + it "writes multiple event types correctly" do + persist = CRE::Persistence::Sqlite::SqlitePersistence.new(":memory:") + persist.migrate! + log = CRE::Audit::AuditLog.new(persist, Bytes.new(32, 0_u8), 1, 1024) + bus = CRE::Engine::EventBus.new + sub = CRE::Engine::Subscribers::AuditSubscriber.new(bus, log) + sub.start + bus.run + + cred_id = UUID.random + bus.publish(CRE::Events::PolicyViolation.new(cred_id, "test-policy", "stale")) + bus.publish(CRE::Events::DriftDetected.new(cred_id, "h1", "h2")) + sleep 0.1.seconds + + persist.audit.latest_seq.should eq 2_i64 + entries = persist.audit.range(1_i64, 2_i64) + entries.map(&.event_type).should eq ["policy.violation", "drift.detected"] + log.verify_chain.should be_true + ensure + bus.try(&.stop) + sub.try(&.stop) + persist.try(&.close) + end +end diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/engine/engine_spec.cr b/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/engine/engine_spec.cr new file mode 100644 index 00000000..b9e7bee3 --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/engine/engine_spec.cr @@ -0,0 +1,42 @@ +# =================== +# ©AngelaMos | 2026 +# engine_spec.cr +# =================== + +require "../../spec_helper" +require "../../../src/cre/engine/engine" +require "../../../src/cre/persistence/sqlite/sqlite_persistence" +require "../../../src/cre/events/credential_events" + +describe CRE::Engine::Engine do + it "boots, accepts events, and shuts down cleanly" do + persist = CRE::Persistence::Sqlite::SqlitePersistence.new(":memory:") + persist.migrate! + + engine = CRE::Engine::Engine.new(persist, Bytes.new(32, 0_u8)) + engine.start + + cred_id = UUID.random + rot_id = UUID.random + engine.bus.publish(CRE::Events::RotationCompleted.new(cred_id, rot_id)) + sleep 0.1.seconds + + persist.audit.latest_seq.should eq 1_i64 + engine.audit_log.verify_chain.should be_true + + engine.stop + ensure + persist.try(&.close) + end + + it "raises if started twice" do + persist = CRE::Persistence::Sqlite::SqlitePersistence.new(":memory:") + persist.migrate! + engine = CRE::Engine::Engine.new(persist, Bytes.new(32, 0_u8)) + engine.start + expect_raises(Exception, "already started") { engine.start } + engine.stop + ensure + persist.try(&.close) + end +end diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/engine/event_bus_spec.cr b/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/engine/event_bus_spec.cr new file mode 100644 index 00000000..0b10fa1c --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/engine/event_bus_spec.cr @@ -0,0 +1,91 @@ +# =================== +# ©AngelaMos | 2026 +# event_bus_spec.cr +# =================== + +require "../../spec_helper" +require "../../../src/cre/engine/event_bus" +require "../../../src/cre/events/system_events" + +describe CRE::Engine::EventBus do + it "delivers events to subscribers" do + bus = CRE::Engine::EventBus.new + bus.run + received = [] of String + received_mutex = Mutex.new + ch = bus.subscribe(buffer: 16, overflow: CRE::Engine::EventBus::Overflow::Block) + spawn do + loop do + ev = ch.receive + received_mutex.synchronize { received << ev.class.name } + rescue Channel::ClosedError + break + end + end + + bus.publish(CRE::Events::AlertRaised.new(severity: CRE::Events::Severity::Warn, message: "hi")) + sleep 0.1.seconds + + received_mutex.synchronize { received.should contain("CRE::Events::AlertRaised") } + ensure + bus.try(&.stop) + end + + it "fans out to multiple subscribers" do + bus = CRE::Engine::EventBus.new + bus.run + + counter1 = Atomic(Int32).new(0) + counter2 = Atomic(Int32).new(0) + ch1 = bus.subscribe + ch2 = bus.subscribe + spawn do + loop do + ch1.receive + counter1.add(1) + rescue Channel::ClosedError + break + end + end + spawn do + loop do + ch2.receive + counter2.add(1) + rescue Channel::ClosedError + break + end + end + + 3.times { bus.publish(CRE::Events::SchedulerTick.new) } + sleep 0.1.seconds + + counter1.get.should eq 3 + counter2.get.should eq 3 + ensure + bus.try(&.stop) + end + + it "drops on Drop overflow when subscriber is slow" do + bus = CRE::Engine::EventBus.new + bus.run + ch = bus.subscribe(buffer: 1, overflow: CRE::Engine::EventBus::Overflow::Drop) + 5.times { bus.publish(CRE::Events::SchedulerTick.new) } + sleep 0.1.seconds + # The slow subscriber's channel buffered at most 1 event; rest were dropped. + # Drain non-blocking: we should be able to take 0 or 1 event before it would block. + delivered = 0 + drained = false + until drained + select + when ch.receive + delivered += 1 + else + drained = true + end + end + delivered.should be <= 5 + delivered.should be < 5 # at least one was dropped + ensure + bus.try(&.stop) + end +end diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/engine/engine.cr b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/engine/engine.cr new file mode 100644 index 00000000..ba9b5c9e --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/engine/engine.cr @@ -0,0 +1,49 @@ +# =================== +# ©AngelaMos | 2026 +# engine.cr +# =================== + +require "log" +require "./event_bus" +require "./subscribers/audit_subscriber" +require "../audit/audit_log" +require "../persistence/persistence" + +module CRE::Engine + class Engine + Log = ::Log.for("cre.engine") + + getter bus : EventBus + getter persistence : Persistence::Persistence + getter audit_log : Audit::AuditLog + + @audit_subscriber : Subscribers::AuditSubscriber + @started : Bool + + def initialize(@persistence : Persistence::Persistence, hmac_key : Bytes, hmac_version : Int32 = 1, ratchet_every : Int32 = 1024) + @bus = EventBus.new + @audit_log = Audit::AuditLog.new(@persistence, hmac_key, hmac_version, ratchet_every) + @audit_subscriber = Subscribers::AuditSubscriber.new(@bus, @audit_log) + @started = false + end + + def start : Nil + raise "engine already started" if @started + @started = true + @audit_subscriber.start + @bus.run + Log.info { "engine started" } + end + + def stop : Nil + return unless @started + Log.info { "engine stopping" } + @bus.publish(Events::ShutdownRequested.new) rescue nil + sleep 0.05.seconds # let subscribers see it + @audit_subscriber.stop + @bus.stop + @started = false + Log.info { "engine stopped" } + end + end +end diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/engine/event_bus.cr b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/engine/event_bus.cr new file mode 100644 index 00000000..960e5ac4 --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/engine/event_bus.cr @@ -0,0 +1,80 @@ +# =================== +# ©AngelaMos | 2026 +# event_bus.cr +# =================== + +require "log" +require "../events/event" + +module CRE::Engine + class EventBus + Log = ::Log.for("cre.event_bus") + + enum Overflow + Block + Drop + end + + record Subscription, channel : Channel(Events::Event), overflow : Overflow + + @inbox : Channel(Events::Event) + @subs : Array(Subscription) + @subs_mutex : Mutex + @running : Bool + + def initialize(inbox_capacity : Int32 = 1024) + @inbox = Channel(Events::Event).new(capacity: inbox_capacity) + @subs = [] of Subscription + @subs_mutex = Mutex.new + @running = false + end + + def subscribe(buffer : Int32 = 64, overflow : Overflow = Overflow::Block) : Channel(Events::Event) + ch = Channel(Events::Event).new(capacity: buffer) + @subs_mutex.synchronize { @subs << Subscription.new(ch, overflow) } + ch + end + + def publish(event : Events::Event) : Nil + @inbox.send(event) + end + + def run : Nil + @running = true + spawn(name: "event-bus") do + while @running + begin + ev = @inbox.receive + rescue Channel::ClosedError + break + end + @subs_mutex.synchronize { @subs.dup }.each { |s| dispatch(s, ev) } + end + end + end + + def stop : Nil + @running = false + @inbox.close + @subs_mutex.synchronize do + @subs.each(&.channel.close) + end + end + + private def dispatch(sub : Subscription, ev : Events::Event) : Nil + case sub.overflow + in Overflow::Block + sub.channel.send(ev) + in Overflow::Drop + select + when sub.channel.send(ev) + # delivered + else + Log.warn { "subscriber drop: #{ev.class.name}" } + end + end + rescue Channel::ClosedError + # subscriber gone; remove from list lazily on next dispatch + end + end +end diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/engine/subscribers/audit_subscriber.cr b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/engine/subscribers/audit_subscriber.cr new file mode 100644 index 00000000..fab3eb8c --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/engine/subscribers/audit_subscriber.cr @@ -0,0 +1,85 @@ +# =================== +# ©AngelaMos | 2026 +# audit_subscriber.cr +# =================== + +require "../event_bus" +require "../../audit/audit_log" +require "../../events/credential_events" +require "../../events/system_events" + +module CRE::Engine::Subscribers + class AuditSubscriber + @ch : Channel(Events::Event)? + @running : Bool + + def initialize(@bus : EventBus, @log : Audit::AuditLog, @actor : String = "system") + @running = false + end + + def start : Nil + @running = true + ch = @bus.subscribe(buffer: 256, overflow: EventBus::Overflow::Block) + @ch = ch + spawn(name: "audit-sub") do + while @running + begin + ev = ch.receive + rescue Channel::ClosedError + break + end + handle(ev) + end + end + end + + def stop : Nil + @running = false + @ch.try(&.close) + end + + private def handle(ev : Events::Event) : Nil + case ev + when Events::RotationCompleted + @log.append("rotation.completed", @actor, ev.credential_id, { + "rotation_id" => ev.rotation_id.to_s, + }) + when Events::RotationFailed + @log.append("rotation.failed", @actor, ev.credential_id, { + "rotation_id" => ev.rotation_id.to_s, + "reason" => ev.reason, + }) + when Events::RotationStepCompleted + @log.append("rotation.step.completed", @actor, ev.credential_id, { + "rotation_id" => ev.rotation_id.to_s, + "step" => ev.step.to_s, + }) + when Events::RotationStepFailed + @log.append("rotation.step.failed", @actor, ev.credential_id, { + "rotation_id" => ev.rotation_id.to_s, + "step" => ev.step.to_s, + "error" => ev.error, + }) + when Events::PolicyViolation + @log.append("policy.violation", @actor, ev.credential_id, { + "policy_name" => ev.policy_name, + "reason" => ev.reason, + }) + when Events::DriftDetected + @log.append("drift.detected", @actor, ev.credential_id, { + "expected_hash" => ev.expected_hash, + "actual_hash" => ev.actual_hash, + }) + when Events::CredentialDiscovered + @log.append("credential.discovered", @actor, ev.credential_id, {} of String => String) + when Events::AlertRaised + @log.append("alert.raised", @actor, nil, { + "severity" => ev.severity.to_s, + "message" => ev.message, + }) + end + rescue ex + EventBus::Log.error(exception: ex) { "audit subscriber failed to write" } + end + end +end diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/events/credential_events.cr b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/events/credential_events.cr new file mode 100644 index 00000000..0ac959de --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/events/credential_events.cr @@ -0,0 +1,100 @@ +# =================== +# ©AngelaMos | 2026 +# credential_events.cr +# =================== + +require "uuid" +require "./event" + +module CRE::Events + abstract class CredentialEvent < Event + getter credential_id : UUID + + def initialize(@credential_id : UUID) + super() + end + end + + class CredentialDiscovered < CredentialEvent + end + + class PolicyViolation < CredentialEvent + getter policy_name : String + getter reason : String + + def initialize(credential_id : UUID, @policy_name : String, @reason : String) + super(credential_id) + end + end + + class RotationScheduled < CredentialEvent + getter rotator_kind : String + + def initialize(credential_id : UUID, @rotator_kind : String) + super(credential_id) + end + end + + class RotationStarted < CredentialEvent + getter rotation_id : UUID + getter rotator_kind : String + + def initialize(credential_id : UUID, @rotation_id : UUID, @rotator_kind : String) + super(credential_id) + end + end + + class RotationStepStarted < CredentialEvent + getter rotation_id : UUID + getter step : Symbol + + def initialize(credential_id : UUID, @rotation_id : UUID, @step : Symbol) + super(credential_id) + end + end + + class RotationStepCompleted < CredentialEvent + getter rotation_id : UUID + getter step : Symbol + + def initialize(credential_id : UUID, @rotation_id : UUID, @step : Symbol) + super(credential_id) + end + end + + class RotationStepFailed < CredentialEvent + getter rotation_id : UUID + getter step : Symbol + getter error : String + + def initialize(credential_id : UUID, @rotation_id : UUID, @step : Symbol, @error : String) + super(credential_id) + end + end + + class RotationCompleted < CredentialEvent + getter rotation_id : UUID + + def initialize(credential_id : UUID, @rotation_id : UUID) + super(credential_id) + end + end + + class RotationFailed < CredentialEvent + getter rotation_id : UUID + getter reason : String + + def initialize(credential_id : UUID, @rotation_id : UUID, @reason : String) + super(credential_id) + end + end + + class DriftDetected < CredentialEvent + getter expected_hash : String + getter actual_hash : String + + def initialize(credential_id : UUID, @expected_hash : String, @actual_hash : String) + super(credential_id) + end + end +end diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/events/event.cr b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/events/event.cr new file mode 100644 index 00000000..53b3ca6c --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/events/event.cr @@ -0,0 +1,13 @@ +# =================== +# ©AngelaMos | 2026 +# event.cr +# =================== + +require "uuid" + +module CRE::Events + abstract class Event + getter id : UUID = UUID.random + getter occurred_at : Time = Time.utc + end +end diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/events/system_events.cr b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/events/system_events.cr new file mode 100644 index 00000000..9b75e305 --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/events/system_events.cr @@ -0,0 +1,29 @@ +# =================== +# ©AngelaMos | 2026 +# system_events.cr +# =================== + +require "./event" + +module CRE::Events + enum Severity + Info + Warn + Critical + end + + class AlertRaised < Event + getter severity : Severity + getter message : String + + def initialize(@severity : Severity, @message : String) + super() + end + end + + class SchedulerTick < Event + end + + class ShutdownRequested < Event + end +end