Event System and Automation
The event system provides pub/sub for reacting to content changes, triggering
automation rules, and delivering webhooks. In a deployment the bus runs on
Redis Streams (BOWRAIN_EVENT_BACKEND=redis), shared by the server and the
worker; the in-process ChannelEventBus below backs single-process
development and tests.
EventBus
The ChannelEventBus is a channel-based pub/sub implementation with per-subscriber goroutines:
bus := event.NewChannelEventBus()
// Subscribe to specific event types
sub := bus.Subscribe(platev.EventBlockUpdated, func(e platev.Event) {
fmt.Printf("Block %s updated in project %s\n", e.Data["block_id"], e.ProjectID)
})
// Subscribe to all events
allSub := bus.SubscribeAll(func(e platev.Event) {
fmt.Printf("Event: %s\n", e.Type)
})
// Unsubscribe
bus.Unsubscribe(sub)
The event types are declared in bowrain/core/event (imported as platev
above); the bus, the emitting store decorator, the automation engine and
webhook delivery live in bowrain/event.
Event Types
| Event | Emitted when |
|---|---|
block.created | A block is stored for the first time |
block.updated | A block is updated |
block.deleted | A block is deleted |
project.created | A project is created |
project.updated | A project is updated |
project.deleted | A project is deleted |
version.created | A version snapshot is created |
collection.created / collection.updated / collection.deleted | A collection changes |
item.created / item.deleted | An item is added or removed |
connector.pull.completed | A pull from a connector completes |
connector.push.completed | A push completes |
connector.sync.completed | A connector sync completes |
push.automations.completed | Every automation for a push has completed |
convergence.run.completed | A run finishes |
flow.started | A flow begins execution |
flow.completed / flow.failed | Declared; no execution path emits them, so no automation trigger fires on a flow finishing |
extraction.completed | Term extraction completes |
quality.gate.pass / quality.gate.fail | A language's ship-gate result changes |
source.review.completed | A source review task is completed |
review.completed | A project's review queue is emptied |
review.decided | One block's target is approved, rejected, or un-reviewed |
review.bulk_approved | An approve-passing pass promotes a language's passing targets |
voice.check.started / voice.check.completed | A voice check runs |
voice.drift / voice.corrected / voice.profile.updated | The voice loop moves |
stream.created / stream.merged / stream.deleted / stream.locked / stream.unlocked / stream.tagged | A stream changes |
member.*, role.template.*, invite.*, token.*, auth.*, session.grant.created, authz.denied | Membership and access |
rollback.performed, platform_config.changed | Administration |
agent.* | The in-product agent (dark by default) |
The canonical list is the EventType constants in
bowrain/core/event/event.go. The automation editor offers only the types an
execution path emits.
EventEmittingStore
The EventEmittingStore decorator wraps a ContentStore and emits events on all mutations:
cs, err := sqlitestore.NewSQLiteStore("working-copy.db")
if err != nil {
log.Fatal(err)
}
bus := event.NewChannelEventBus()
emittingStore := event.NewEventEmittingStore(cs, bus)
Automation Rules
The automation engine evaluates rules triggered by events. It takes the bus and
an ActionExecutor, which runs a rule's actions (run_flow, notify, and
the rest) when a rule matches:
engine := event.NewAutomationEngine(bus, executor)
engine.AddRule(event.AutomationRule{
Name: "auto-draft-on-push",
EventType: platev.EventPushCompleted,
Conditions: []event.Condition{
{Field: "project_id", Operator: "equals", Value: "proj-1"},
},
// Actions are executed by the ActionExecutor.
})
engine.Start(ctx)
Loop Prevention
Automation chains are tracked via CausationID. If a chain exceeds the maximum depth (default 5), it is automatically broken to prevent infinite loops.
Quality gate events
The server publishes quality.gate.fail and quality.gate.pass from the ship
gate (bowrain/server/shipgate_events.go). Each time it derives a project's
ship states, on a dashboard or ship-feed read and when a convergence run ends,
it compares each language's gates (store.ShipGateResults) with the failures
it has already announced, which the content store keeps in ship_gate_failures:
- An unmet gate with no announced failure, or whose
not_checkedvalue changed, publishesquality.gate.failand is recorded. - An announced failure whose gate is met again publishes
quality.gate.passand is cleared. - Any other result publishes nothing. A gate the derivation did not evaluate, such as the checks below full coverage, opens and clears nothing.
The gates are translated, checks, terms, stale and rejected. Each
event carries the project in ProjectID and this data:
| Key | Value |
|---|---|
gate_name | The gate |
locale | The language |
stream | The stream the content is on |
actual | Translated blocks for translated, and blocks at fault for every other gate |
required | Total blocks for translated, and 0 for every other gate |
not_checked | true when the gate is unmet because the language has no content or a check has no result |
ship_state | The language's derived ship state |
workspace_id, workspace_slug | The project's workspace |
A failure notification carries the group key
quality-gate:<project>:<stream>:<gate>:<locale> (event.QualityGateGroupKey),
and the pass for the same gate marks it read. The events block no operation.
Webhooks
Webhook delivery with HMAC-SHA256 signing and retry is a library primitive in
bowrain/event; no workspace-facing surface configures outbound webhooks.
wh := event.WebhookDelivery{
URL: "https://example.com/webhook",
Secret: "shared-secret",
}
err := wh.Deliver(ctx, eventData)
Signature verification on the receiving end:
valid := event.VerifySignature(payload, signature, secret)