How Multica's Event-Driven Architecture Decouples Handlers and Services

Multica uses a lightweight in-process pub/sub event bus to separate HTTP handlers and core services, allowing them to communicate through typed events without direct imports or tight coupling.

In the multica-ai/multica codebase, an event-driven architecture eliminates hard dependencies between the HTTP layer and side-effect logic. Instead of handlers directly invoking notification systems or logging infrastructure, they publish events to a central bus. This pattern ensures that core business logic remains focused on domain operations while cross-cutting concerns like real-time sync, activity logging, and subscriber management react independently to state changes.

The Event Bus as a Decoupling Layer

At the heart of this design sits the events.Bus struct defined in server/internal/events/bus.go. It maintains a map of event types to callback functions, acting as a message broker that keeps publishers and subscribers ignorant of each other.

Publish-Only APIs

Handlers and services interact with the bus exclusively through the Publish method. For example, the Handler type in server/internal/handler/handler.go provides a helper method that wraps the bus:

func (h *Handler) publish(eventType, workspaceID, actorType, actorID string, payload any) {
    h.Bus.Publish(events.Event{
        Type:        eventType,
        WorkspaceID: workspaceID,
        ActorType:   actorType,
        ActorID:     actorID,
        Payload:     payload,
    })
}

The publisher only knows the event type and payload structure. It never imports the concrete subscriber logic, satisfying the Single Responsibility Principle and keeping HTTP handlers thin.

Global Typed Subscriptions

Consumers register via bus.Subscribe(eventType, handler) or bus.SubscribeAll(handler), allowing listeners to reside in separate packages. For instance, server/cmd/server/subscriber_listeners.go registers callbacks that automatically subscribe users when issues are created:

bus.Subscribe(protocol.EventIssueCreated, func(e events.Event) {
    payload, _ := e.Payload.(map[string]any)
    issue := payload["issue"].(handler.IssueResponse)

    // creator becomes a subscriber
    addSubscriber(bus, queries, e.WorkspaceID, issue.ID,
        issue.CreatorType, issue.CreatorID, "creator")
})

Because the bus holds only function references, you can add, remove, or modify listeners without touching the publishing code in server/internal/handler/handler.go or server/internal/service/task.go.

Recoverable Synchronous Dispatch

While the bus operates synchronously in-process, it protects publishers from downstream failures. The dispatch loop executes each listener inside a recover block, ensuring that a panic in one subscriber does not propagate to others or crash the publisher. This guarantees that handlers never block on slow or faulty consumers, maintaining API responsiveness while still allowing immediate side effects.

Separating Domain Logic from Infrastructure

Services also publish events directly. In server/internal/service/task.go, the TaskService emits progress updates without knowing which UI clients will receive them:

func (s *TaskService) ReportProgress(ctx context.Context, taskID, workspaceID, summary string, step, total int) {
    s.Bus.Publish(events.Event{
        Type:        protocol.EventTaskProgress,
        WorkspaceID: workspaceID,
        ActorType:   "system",
        Payload: protocol.TaskProgressPayload{
            TaskID:  taskID,
            Summary: summary,
            Step:    step,
            Total:   total,
        },
    })
}

Real-time WebSocket pushes, activity logs, and notification emails all consume these events from independent packages. This separation means you can disable an entire subsystem (like audit logging) by simply unregistering its listeners, without modifying the task execution logic.

Summary

• Centralized Bus: events.Bus in server/internal/events/bus.go provides the Publish, Subscribe, and SubscribeAll API that decouples producers from consumers.
• Handler Isolation: HTTP handlers in server/internal/handler/handler.go emit events through a private publish helper, staying blind to subscriber implementations.
• Service Emission: Core services like TaskService emit domain events directly, keeping business logic free of infrastructure concerns.
• Safe Dispatch: The bus recovers from panics in individual listeners, preventing cascade failures.
• Modular Subscribers: Listeners in server/cmd/server/subscriber_listeners.go and similar files can be added or removed without changing publisher code.

Frequently Asked Questions

What is the role of the events.Bus in Multica?

The events.Bus acts as an in-process message broker that enables the event-driven architecture. It stores a registry of callback functions keyed by event type, allowing any package to subscribe to domain changes without importing the publisher's code. This centralizes communication between HTTP handlers, background services, and cross-cutting infrastructure like WebSocket hubs.

How does Multica ensure that one failing listener doesn't crash the system?

The bus iterates over matched handlers and executes each inside a recover block. If a subscriber panics, the bus catches the error, preventing propagation to other listeners or the publisher. This design guarantees that a bug in activity logging or notification code cannot bring down the API server.

Can subscribers be added without modifying the publishing code?

Yes. Subscribers register themselves at runtime using bus.Subscribe or bus.SubscribeAll. Because the bus maintains a map of callbacks, you can introduce new listeners (such as audit loggers or analytics collectors) in separate files like server/cmd/server/subscriber_listeners.go without touching the handler or service code that emits the events.

Where are event listeners registered in the Multica codebase?

Listener registration typically occurs in server/cmd/server/ files such as subscriber_listeners.go and activity_listeners.go. These files import the event bus instance and call subscription methods during application startup, wiring up domain reactions like auto-subscription and real-time UI updates independently of the core business logic.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}