Event System

Event System

Events are how modules communicate in MuxCore. They're the signals that travel through the fabric — when something happens, an event goes out, and any module that cares can react.

---

What Are Events?

Think of events like notifications on your phone. When a download finishes, the downloader doesn't need to know about media managers, transcoders, or notification senders. It just publishes a "download complete" event. Any module that subscribed to that event reacts accordingly.

In the *arr world, integrations are hardcoded: Sonarr knows it needs to tell qBittorrent to download, then tell Plex to refresh. In MuxCore, the downloader just announces "I'm done." Everything else figures itself out.

---

How It Works

Module A publishes "download.completed"
         │
         ▼
    ┌──────────┐
    │ Event Bus │
    └──────────┘
         │
    ┌────┴────┬──────────┐
    ▼         ▼          ▼
Module B   Module C   Module D
(import)  (transcode)(notify)

The publisher doesn't know who's listening. Subscribers don't know who published. The event bus connects them.

---

Event Structure

Every event has the same shape:

type Event struct {
    ID        string            // unique ID for this event
    Type      string            // what kind of event (e.g. "download.completed")
    Source    string            // which module sent it
    Payload   []byte            // the data (JSON or protobuf)
    Metadata  map[string]string // extra context
    Timestamp time.Time         // when it happened
    TraceID   string            // for tracing across services
}

---

Publishing Events

Simple Publish

payload, _ := json.Marshal(map[string]string{
    "download_id": "abc123",
    "title":       "Movie Title",
})

eventsv1.NewEventServiceClient(conn).Publish(ctx, &eventsv1.PublishRequest{
    Event: contracts.Event{
    Type:    "download.completed",
    Source:  "downloader-qbittorrent",
    Payload: payload,
})

Publishing with Domain Events

Domain event types and payloads are defined in contract repos:

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

payload, _ := json.Marshal(dl.DownloadCompletedPayload{
    DownloadID: id,
    Title:      title,
    FileSize:   size,
})

eventsv1.NewEventServiceClient(conn).Publish(ctx, &eventsv1.PublishRequest{
    Event: contracts.Event{
    Type:    dl.EventDownloadCompleted,  // "download.completed"
    Source:  "downloader-qbittorrent",
    Payload: payload,
})

---

Subscribing to Events

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

eventsv1.NewEventServiceClient(conn).Subscribe(ctx, &eventsv1.SubscribeRequest{EventType:  dl.EventDownloadCompleted,
    func(ctx context.Context, event contracts.Event) error {
        var payload dl.DownloadCompletedPayload
        if err := json.Unmarshal(event.Payload, &payload); err != nil {
            return err
        }
        // React to the download finishing
        // Import into library, start transcode, send notification...
        return nil
    })

Subscribe to "*" to receive every event.

---

System Events (Built Into Core)

Core defines a small set of infrastructure events. Everything else is defined in domain contract repos.

Event	When It Fires	Payload
module.registered	Module joins the fabric	{module_id, version}
module.unregistered	Module is removed	{module_id}
module.degraded	Module running with reduced capabilities	{module_id, error}
cluster.node.joined	Machine joins the cluster	{node_id, grpc_addr, http_addr}
cluster.node.left	Machine leaves the cluster	{node_id}
cluster.node.degraded	Machine running degraded	{node_id}
cluster.leader.changed	New cluster leader elected	{node_id}
storage.tier.transition	Object moved between storage tiers	(varies)

---

Domain Events

Domain events are defined in their respective contract repos. Here are the common ones:

Contract Repo	Example Events
contracts-downloader	download.started, download.completed, download.failed
contracts-media	library.item.added, library.item.removed
contracts-playback	playback.started, playback.stopped
contracts-transcoder	transcode.started, transcode.completed, transcode.failed
contracts-metadata	metadata.fetched
contracts-artwork	artwork.fetched
contracts-content	content.missing, content.fetched
contracts-discovery	discovery.updated
contracts-quality	quality.decision, format.matched
contracts-mediainfo	media.analyzed

---

Real Example: Movie Request Flow

Here's how a full movie request flows through events:

1. User clicks "Request Movie" in the UI
   → publishes "media.requested"

2. Workflow engine receives "media.requested"
   → starts the "movie-request" tapestry
   → Step 1: calls metadata provider directly (gRPC)
   → Step 2: calls indexer directly (gRPC)

3. Downloader receives task
   → publishes "download.started"
   → ... time passes ...
   → publishes "download.completed"

4. Media manager receives "download.completed"
   → imports the file
   → publishes "library.item.added"

5. Transcoder receives "library.item.added"
   → transcodes to additional qualities
   → publishes "transcode.completed"

6. Notification module receives "library.item.added"
   → sends Discord/Telegram/email notification

---

Event Bus Implementation

By default, MuxCore uses an in-memory event bus — fast, zero-config, sufficient for single-node setups. For multi-node deployments, a NATS module (eventbus-nats) is available that distributes events across machines.

The in-memory bus:

• Dispatches events to matching subscribers concurrently
• Bounds goroutine concurrency to CPU count × 2
• Auto-generates event IDs and timestamps
• Propagates trace IDs from context

---

gRPC Event Service (Multi-Node)

When running across multiple machines, events are relayed through gRPC. Each node runs an EventService with three RPCs:

Publish — Send an event to a remote node

Node A: EventService.Publish(event) → Node B: delivers to local subscribers

Subscribe — Stream events from a remote node

Node A: EventService.Subscribe(["download.*"]) → Node B: streams matching events

Request — Send a request and wait for reply

Node A: EventService.Request(event, timeout) → Node B: processes and returns single event

This is how events travel between machines when you're not using NATS.

---

gRPC Module Mesh

For direct module-to-module communication, MuxCore provides a gRPC mesh service:

Call — Invoke a method on another module

service ModuleMesh {
  rpc Call(CallRequest) returns (CallResponse);
  rpc StreamCall(stream CallRequest) returns (stream CallResponse);
}

All module-to-module calls route through the gRPC mesh. The mesh handles routing transparently.

Call Policy

When a CallPolicyProvider is registered, the mesh client enforces access control before every call:

ctx = contracts.WithCallerID(ctx, "my-module")
result, err := meshv1.NewModuleMeshClient(conn).Call(ctx, &meshv1.CallRequest{TargetModule:  "target-module", "Method", payload)

The call policy provider checks: "is my-module allowed to call Method on target-module?" If no policy provider is registered, all calls are allowed.

---

Event Schema Versioning

All event payloads carry a schema version (EventSchemaVersion = "v1"). This ensures that as event formats evolve, consumers can detect version mismatches and handle them appropriately.

---

Best Practices

Do this: Publish an event when state changes.

bus.Publish(ctx, Event{Type: "download.completed", Payload: payload})

Not this: Call another module directly for state changes.

// BAD — tight coupling
mediaManager.Import(download)

Do this: Subscribe with specific event types.

bus.Subscribe(ctx, "download.completed", handler)

Not this: Subscribe to "*" and filter manually. (Use "*" only for debugging.)

---

Dead Letter Handling

Critical event handlers can use a Dead Letter provider to prevent silent data loss:

dl, _ := discoveryv1.NewDiscoveryServiceClient(conn).FindByCapability(ctx, &discoveryv1.FindByCapabilityRequest{Capability: "deadletter")
eventsv1.NewEventServiceClient(conn).Subscribe(ctx, &eventsv1.SubscribeRequest{EventType:  "workflow.step.completed", func(ctx context.Context, event contracts.Event) error {
    if err := processEvent(event); err != nil {
        if len(dl) > 0 {
            dl[0].Module.(contracts.DeadLetterProvider).Store(ctx, event, "my-handler", err)
        }
        return err
    }
    return nil
})

Failed events can be inspected, replayed, or discarded later.

---

Event Store (Event Sourcing)

For modules that need event sourcing — rebuilding state by replaying all past events — MuxCore provides an EventStore contract:

store, _ := discoveryv1.NewDiscoveryServiceClient(conn).FindByCapability(ctx, &discoveryv1.FindByCapabilityRequest{Capability: "event.store")
// Append events atomically
seq, _ := store.Append(ctx, "user-abc123", events)
// Rebuild state by replaying from the beginning
events, _ := store.Read(ctx, "user-abc123", 1, 0)
// Subscribe to live events from the last seen sequence
ch, _ := store.Subscribe(ctx, "user-abc123", lastSeq+1)

The EventStore is distinct from the EventBus. The bus is for real-time pub/sub. The store is for state reconstruction via replay.

---

Next Steps

• Storage — how file storage works
• Contracts Reference — EventBus and EventStore interfaces in detail
• Writing Modules — using events in your own modules