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.
This commit is contained in:
CarterPerez-dev
parent
fe5e9cd07b
commit
2ba81cfd82
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
Loading…
Reference in New Issue