Event System

Event System

Events (signals) are how modules communicate without knowing about each other. This page explains the event bus, event types, how to publish and subscribe, and module lifecycle event management.

---

How Events Work

Modules publish events on the event bus. Other modules subscribe to event types they care about. The publisher never knows who's listening:

Downloader publishes: "download.completed"
    ↓
Media Manager receives it → imports the file
Transcoder receives it → starts optimizing
Notification module receives it → sends a Discord message

---

Event Types

Event types use dotted naming: domain.action. Core defines infrastructure events. Domain contracts define domain events.

Core Events

Event Type	Payload	When
module.registered	ModuleRegisteredPayload	Module registers successfully
module.unregistered	ModuleUnregisteredPayload	Module unregisters
module.degraded	ModuleDegradedPayload	Module health check fails
node.joined	NodeInfo	New core joins the cluster
node.left	NodeID	Core leaves cluster (evicted or graceful)
node.degraded	NodeDegradedPayload	Node reports reduced capability
leader.changed	LeaderID	Cluster leader changes

Domain Events

Defined in their respective contracts-* repos:

• download.started, download.progress, download.completed, download.failed — contracts-downloader
• transcode.started, transcode.completed — contracts-transcoder
• media.imported, media.deleted — contracts-media
• workflow.step.started, workflow.step.completed, workflow.step.failed — contracts-workflow

---

Publishing Events

import "github.com/Muxcore-Media/contracts-downloader"

payload, _ := json.Marshal(dl.DownloadCompletedPayload{
    DownloadID: id,
    Title:      "The Terminator",
    FileSize:   72 * 1024 * 1024 * 1024,
})

bus.Publish(ctx, contracts.Event{
    Type:    dl.EventDownloadCompleted,
    Source:  "downloader-qbittorrent",
    Payload: payload,
})

Sidecar modules publish via gRPC:

events := eventsv1.NewEventServiceClient(conn)
events.Publish(ctx, &eventsv1.PublishRequest{
    Event: &eventsv1.Event{
        Type:    "download.completed",
        Source:  moduleID,
        Payload: jsonPayload,
    },
})

Publish Policy

Event publication is subject to capability-based enforcement. The built-in PublishPolicyProvider checks that the publishing module's declared capabilities include the event type being published. Set MUXCORE_STRICT_PUBLISH_POLICY=true to hard-deny capability mismatches.

---

Subscribing to Events

bus.Subscribe(ctx, dl.EventDownloadCompleted, func(ctx context.Context, event contracts.Event) error {
    var payload dl.DownloadCompletedPayload
    json.Unmarshal(event.Payload, &payload)
    // Process the download...
    return nil
})

Use "*" to subscribe to all events (useful for audit loggers and monitoring modules).

Sidecar modules subscribe via gRPC streaming:

events := eventsv1.NewEventServiceClient(conn)
stream, _ := events.Subscribe(ctx, &eventsv1.SubscribeRequest{
    EventTypes: []string{"download.completed", "transcode.completed"},
})
for {
    event, err := stream.Recv()
    // Handle event...
}

---

Per-Module Subscriptions (SubscribeModule / UnsubscribeAll)

Modules that subscribe to events during Start() and need clean teardown during Stop() should use SubscribeModule and UnsubscribeAll:

// In Start():
bus.SubscribeModule(ctx, m.id, "download.completed", m.handleDownloadCompleted)

// In Stop():
bus.UnsubscribeAll(ctx, m.id)  // removes all subscriptions tagged with this module ID

This is cleaner than tracking individual handler references. UnsubscribeAll removes every subscription tagged with the given module ID in a single call — no need to store and individually unsubscribe each handler.

UnsubscribeAll with an empty module ID is a no-op. It only removes subscriptions that were registered with SubscribeModule.

---

Request-Reply Pattern

For synchronous request-reply, use Request():

reply, err := bus.Request(ctx, contracts.Event{
    Type:    "metadata.lookup",
    Source:  "media-movies",
    Payload: payload,
}, 5*time.Second)

The event bus creates a temporary subscription for <type>.reply, publishes the request event, and waits for a reply. The first reply wins. Timeout or context cancellation returns an error.

---

Event Handler Guidelines

• Handlers run concurrently — the event bus dispatches each handler in its own goroutine. Don't rely on handler ordering.
• 30-second timeout — each handler gets a 30-second context timeout. Long-running work should be enqueued, not done in the handler.
• Return errors for logging — returned errors are logged but don't affect other subscribers.
• Don't block — if your handler does heavy work, spawn a goroutine.
• Trace propagation — trace IDs are propagated from publisher to subscriber context automatically.

---

Audit Events

When an audit logger is configured, the event bus records every Publish, Subscribe, and Unsubscribe call:

Action	Audited Fields
event.publish	source, event type, subscriber count, trace ID
event.subscribe	event type, handler pointer
event.unsubscribe	event type, handler pointer

Read operations (Get, List, Exists) are intentionally not audited at the bus level — they're hot paths that would overwhelm the audit trail.

---

Event Schema Versioning

The constant EventSchemaVersion = "v1" is defined in pkg/contracts/eventschema.go but is reserved — it is not currently consumed at runtime. It is defined for future typed-payload support. Current events use raw []byte payloads that callers marshal/unmarshal themselves.

---

Next Steps

• Module System — how modules discover each other
• Contracts Reference — EventBus and EventHandler interfaces
• Writing Modules — publish and subscribe from sidecar modules