@slck/mediator

v1.0.2

Published

a month ago

Cross-cutting mediator kernel — send/stream/publish, 11-middleware pipeline, outbox, inbox, saga, and event bus for Node.js

0High
0Medium
0Low

chandru415

mediator cqrs command query notification pub-sub middleware pipeline outbox inbox saga event-bus typescript node

@slck/mediator

Cross-cutting mediator kernel for Node.js — type-safe CQRS dispatch, 11-behaviour middleware pipeline, pub/sub notifications, streaming, outbox, inbox, saga orchestration, and integration event bus — all with structured observability baked in.

Quick Start

npm install
npm run build
node dist/example/main.js

Architecture Overview

graph TB
    classDef http fill:#6366f1,color:#fff,stroke:#4338ca
    classDef core fill:#0ea5e9,color:#fff,stroke:#0284c7
    classDef mw fill:#f59e0b,color:#000,stroke:#d97706
    classDef durable fill:#10b981,color:#fff,stroke:#059669
    classDef obs fill:#8b5cf6,color:#fff,stroke:#7c3aed
    classDef transport fill:#ef4444,color:#fff,stroke:#dc2626

    HTTP["🌐 HTTP / CLI / Tests"]:::http

    subgraph Mediator["⚡ Mediator Core"]
        SEND["send()  Command · Query"]:::core
        STREAM["stream()  StreamQuery"]:::core
        PUBLISH["publish()  Notification"]:::core
    end

    subgraph Pipeline["🔧 Middleware Pipeline"]
        direction LR
        L["Logging"]:::mw
        V["Validation"]:::mw
        A["Authorization"]:::mw
        M["Metrics"]:::mw
        RL["RateLimit"]:::mw
        BH["Bulkhead"]:::mw
        C["Caching"]:::mw
        R["Retry"]:::mw
        T["Timeout"]:::mw
        CB["CircuitBreaker"]:::mw
        TX["Transaction"]:::mw
        L-->V-->A-->M-->RL-->BH-->C-->R-->T-->CB-->TX
    end

    subgraph Durable["💾 Durable Infrastructure"]
        OB["Outbox"]:::durable
        IB["Inbox"]:::durable
        SG["Saga"]:::durable
    end

    subgraph Bus["📡 Integration Event Bus"]
        KAFKA["Kafka"]:::transport
        NATS["NATS"]:::transport
        RABBIT["RabbitMQ"]:::transport
        SQS["AWS SQS"]:::transport
        MEM["In-Memory"]:::transport
    end

    subgraph Obs["🔭 Observability"]
        LOG["Logger"]:::obs
        MET["Metrics"]:::obs
        TR["Tracer"]:::obs
    end

    HTTP --> SEND
    HTTP --> STREAM
    HTTP --> PUBLISH
    SEND --> Pipeline
    STREAM --> Pipeline
    Pipeline --> HANDLER["🎯 Handler"]:::core
    HANDLER --> Durable
    Durable --> Bus
    SEND -.-> Obs
    STREAM -.-> Obs
    PUBLISH -.-> Obs
    Pipeline -.-> Obs

Request → Handler Dispatch (1:1)

sequenceDiagram
    autonumber
    participant C as 🌐 Caller
    participant M as ⚡ Mediator
    participant P as 🔧 Pipeline
    participant H as 🎯 Handler
    participant N as 📣 Notification

    C->>M: send(CreateOrderCommand)
    activate M
    M->>P: Logging → Validation → Metrics → Retry → Timeout
    activate P
    P->>H: handle(request, context)
    activate H
    H-->>N: publish(OrderCreatedEvent)
    H-->>P: OrderDto
    deactivate H
    P-->>M: OrderDto
    deactivate P
    M-->>C: OrderDto ✅
    deactivate M

Middleware Pipeline — Russian Doll

graph LR
    classDef outer fill:#6366f1,color:#fff,stroke:none
    classDef mid fill:#0ea5e9,color:#fff,stroke:none
    classDef inner fill:#10b981,color:#fff,stroke:none
    classDef core fill:#f59e0b,color:#000,stroke:none

    REQ(["📨 Request"])
    LOG["① Logging"]:::outer
    VAL["② Validation"]:::outer
    AUTH["③ Authorization"]:::outer
    MET["④ Metrics"]:::mid
    RL["⑤ RateLimit"]:::mid
    BH["⑥ Bulkhead"]:::mid
    CACHE["⑦ Caching"]:::mid
    RETRY["⑧ Retry"]:::inner
    TO["⑨ Timeout"]:::inner
    CB["⑩ CircuitBreaker"]:::inner
    TX["⑪ Transaction"]:::inner
    HANDLER(["🎯 Handler"]):::core

    REQ-->LOG-->VAL-->AUTH-->MET-->RL-->BH-->CACHE-->RETRY-->TO-->CB-->TX-->HANDLER
    HANDLER-.->TX-.->CB-.->TO-.->RETRY-.->CACHE-.->BH-.->RL-.->MET-.->AUTH-.->VAL-.->LOG

Each layer calls next() to proceed or throws to short-circuit. Responses bubble back through the same stack.

Notification Fan-Out (1:N)

graph LR
    classDef event fill:#f59e0b,color:#000,stroke:#d97706
    classDef handler fill:#10b981,color:#fff,stroke:#059669
    classDef pub fill:#6366f1,color:#fff,stroke:#4338ca

    E(["📣 OrderCreatedEvent"]):::event
    PUB{{"🔀 Publisher"}}:::pub

    H1["📧 SendEmailHandler"]:::handler
    H2["📊 UpdateAnalyticsHandler"]:::handler
    H3["📦 TriggerFulfillmentHandler"]:::handler
    H4["🔖 AuditLogHandler"]:::handler

    E-->PUB
    PUB-->H1
    PUB-->H2
    PUB-->H3
    PUB-->H4

| Publisher | Behaviour | |---|---| | ForeachAwaitPublisher (default) | Sequential — isolated failure per handler | | TaskWhenAllPublisher | Parallel via Promise.allSettled |

Durable Infrastructure

flowchart TD
    classDef handler fill:#0ea5e9,color:#fff,stroke:#0284c7
    classDef store fill:#10b981,color:#fff,stroke:#059669
    classDef worker fill:#8b5cf6,color:#fff,stroke:#7c3aed
    classDef bus fill:#ef4444,color:#fff,stroke:#dc2626
    classDef dl fill:#f59e0b,color:#000,stroke:#d97706

    H(["🎯 Handler"]):::handler
    OW["📝 OutboxWriter\nwrite to store atomically"]:::store
    OS[("📦 OutboxStore\nInMemory · File · SQL · Mongo")]:::store
    WK["⚙️ OutboxWorker\nlease-based processing"]:::worker
    BUS["📡 Event Bus"]:::bus
    DL[("☠️ Dead-Letter Store")]:::dl
    IN[("📥 InboxStore\nidempotency keys")]:::store
    IC["🔒 InboxConsumer\ndedup on consume"]:::store

    H-->OW-->OS
    WK--"poll pending"-->OS
    WK--"publish"-->BUS
    WK--"mark processed"-->OS
    WK--"on max retries"-->DL
    BUS--"deliver"-->IC
    IC--"check key"-->IN
    IC--"mark processed"-->IN

Saga Orchestration

stateDiagram-v2
    [*] --> Running : start()
    Running --> Running : step N execute ✅
    Running --> Compensating : step N throws ❌
    Compensating --> Compensating : compensate step N-1 … 1
    Compensating --> Failed : all compensations done
    Running --> Completed : all steps done ✅

    note right of Compensating
        LIFO rollback —
        compensations run in
        reverse step order
    end note

Store Tiers

graph LR
    classDef mem fill:#6366f1,color:#fff,stroke:none
    classDef file fill:#0ea5e9,color:#fff,stroke:none
    classDef prod fill:#10b981,color:#fff,stroke:none

    subgraph Outbox
      O1["InMemoryOutboxStore\n⚡ dev / test"]:::mem
      O2["FileOutboxStore\n💾 single-process"]:::file
      O3["DurableOutboxStore\n🏭 SQL · Mongo"]:::prod
    end
    subgraph Inbox
      I1["InMemoryInboxStore"]:::mem
      I2["FileInboxStore"]:::file
      I3["DurableInboxStore"]:::prod
    end
    subgraph Saga
      S1["InMemorySagaStore"]:::mem
      S2["FileSagaStore"]:::file
      S3["DurableSagaStore"]:::prod
    end

    O1-->O2-->O3
    I1-->I2-->I3
    S1-->S2-->S3

Integration Event Bus — Transports

graph TB
    classDef bus fill:#6366f1,color:#fff,stroke:#4338ca
    classDef t fill:#ef4444,color:#fff,stroke:#dc2626

    BUS(["📡 IntegrationEventBus"]):::bus
    K["☕ Kafka\nKafkaJS"]:::t
    N["🌊 NATS"]:::t
    R["🐇 RabbitMQ\namqplib"]:::t
    S["☁️ AWS SQS"]:::t
    I["🧪 In-Memory\ndev / test"]:::t

    BUS-->K
    BUS-->N
    BUS-->R
    BUS-->S
    BUS-->I

All messages use a versioned JSON envelope:

{
  "specVersion": "1.0",
  "id": "uuid",
  "name": "orders.order-created",
  "version": 1,
  "schema": "orders.order-created/v1",
  "payload": { ... },
  "occurredAt": "2026-05-01T00:00:00.000Z",
  "idempotencyKey": "orders.order-created:order-123",
  "headers": { "aggregate-id": "order-123", "aggregate-type": "order" }
}

Observability

graph LR
    classDef iface fill:#8b5cf6,color:#fff,stroke:#7c3aed
    classDef impl fill:#6366f1,color:#fff,stroke:#4338ca
    classDef otel fill:#0ea5e9,color:#fff,stroke:#0284c7

    LI["Logger interface"]:::iface --> CL["ConsoleLogger\n(built-in)"]:::impl
    LI --> PL["PinoLogger\n(custom)"]:::otel
    LI --> WL["WinstonLogger\n(custom)"]:::otel

    MI["Metrics interface"]:::iface --> IM["InMemoryMetrics\n(built-in)"]:::impl
    MI --> PM["Prometheus adapter\n(custom)"]:::otel

    TI["Tracer interface"]:::iface --> IT["InMemoryTracer\n(built-in)"]:::impl
    TI --> OT["OtelTracer\n@opentelemetry/api"]:::otel

Standard metric names emitted on every dispatch:

| Metric | Type | Tags | |---|---|---| | request_success_total | counter | request | | request_error_total | counter | request | | request_duration_ms | histogram | request | | stream_success_total | counter | request | | event_publish_count | counter | notification | | middleware_request_duration_ms | histogram | request |

Project Structure

src/
├── index.ts               ← Public barrel — import everything from here
├── mediator/
│   ├── contracts.ts       ← RequestBase, NotificationBase, all interfaces
│   ├── mediator.ts        ← RuntimeMediator
│   ├── generated.ts       ← GeneratedMediator (for code-gen consumers)
│   ├── registry.ts        ← HandlerRegistry with notification cache
│   ├── dispatch.ts        ← Shared send/stream/publish base + DispatchFilter
│   ├── publishers.ts      ← ForeachAwaitPublisher, TaskWhenAllPublisher
│   ├── container.ts       ← ServiceCollection / ServiceProvider
│   └── errors.ts          ← All error types
├── middleware/             ← 11 pipeline behaviours
│   ├── logging.ts
│   ├── validation.ts
│   ├── authorization.ts
│   ├── metrics.ts
│   ├── rate-limit.ts
│   ├── bulkhead.ts
│   ├── caching.ts
│   ├── retry.ts
│   ├── timeout.ts
│   ├── circuit-breaker.ts
│   └── transaction.ts
├── outbox/outbox.ts       ← OutboxWriter, OutboxWorker, stores, dead-letter
├── inbox/inbox.ts         ← InboxConsumer, stores
├── saga/saga.ts           ← SagaManager, SagaTypeRegistry, stores
├── event-bus/eventbus.ts  ← IntegrationEventBus, transports, envelope
├── platform/
│   ├── observability.ts   ← Logger, Metrics, Tracer interfaces + built-ins
│   └── lifecycle.ts       ← WorkerHost, HostedService
├── adapters/              ← SQL, MongoDB, OpenTelemetry production adapters
├── features/              ← Runnable CQRS feature slices (orders, users)
└── example/               ← End-to-end demo (generated mediator + full stack)
tools/
└── generate-mediator.cjs  ← Build-time direct-dispatch code generator

Build & Generate

# Install dependencies
npm install

# Re-generate direct-dispatch mediator from src/features/*
npm run generate

# Compile TypeScript
npm run build

# Run end-to-end example
node dist/example/main.js

The code generator scans src/features/ for handler classes and emits src/example/generated-mediator.ts with a direct if (request instanceof X) dispatch table — no runtime reflection, zero allocation on the hot path.

Documentation

→ DEVELOPER_GUIDE.md — Full API reference with code examples for every feature.

Shipped inside the package. After install, open it at:
node_modules/@slck/mediator/DEVELOPER_GUIDE.md

Published

Vulnerabilities

Links

Maintainers

Keywords

Readme

@slck/mediator

Quick Start

Architecture Overview

Request → Handler Dispatch (1:1)

Middleware Pipeline — Russian Doll

Notification Fan-Out (1:N)

Durable Infrastructure

Saga Orchestration

Store Tiers

Integration Event Bus — Transports

Observability

Project Structure

Build & Generate

Documentation