swift-ddd-kit

swift-ddd-kit is a Swift framework that brings Domain-Driven Design, Event Sourcing, and CQRS to Server-Side Swift. While the Swift backend ecosystem has grown significantly, the building blocks for production-grade DDD architecture — aggregate roots, event sourcing repositories, CQRS projectors, and event migration — have remained largely absent. swift-ddd-kit fills that gap.

📖 API reference: gradyzhuo.github.io/swift-ddd-kit/documentation — DocC for every module of the latest release, with a version menu for older minors and main.

Overview

This framework focuses on:

• Modeling business logic using Aggregate Roots as consistency boundaries
• Capturing every state change as an immutable Domain Event
• Replaying events to reconstruct state (Event Sourcing)
• Separating write and read models through CQRS Projectors
• Evolving event schemas safely with Migration utilities
• Reducing boilerplate via SPM build-tool plugins (YAML → Swift code generation)

Architecture

Write and Read are fully independent — they share no direct coupling. Both reach into KurrentDB separately: the Write side appends domain events; the Read side reads those same events to build query-optimized models.

 ┌─────────────────────── FRAMEWORKS & DRIVERS ───────────────────────────────┐
 │         KurrentDB                              PostgreSQL / Memory         │
 │        (Event Store)                            (Read Store)               │
 └──────────────┬───────────────────────────────────────────────┬─────────────┘
                │ ↑ appends           reads events ↓ ───────►   │ ↑ persists
 ┌──────────────┴──────────────────────┬────────────────────────┴─────────────┐
 │     WRITE SIDE (Command)            │     READ SIDE (Query)                │
 ├─────────────────────────────────────┼──────────────────────────────────────┤
 │ INTERFACE ADAPTERS                  │ INTERFACE ADAPTERS                   │
 │   Command Handler  (Controller)     │   Query Handler  (Presenter)         │
 │   KurrentStorageCoordinator(Gateway)│   KurrentStorageCoordinator(Gateway) │
 │                                     │   ReadModelStore  (Gateway)          │
 ├─────────────────────────────────────┼──────────────────────────────────────┤
 │ USE CASES                           │ USE CASES                            │
 │   Usecase                           │   EventSourcingProjector             │
 │   EventSourcingRepository           │   ├─ buildReadModel(input:)          │
 │   EventTypeMapper  (Adapter)        │   └─ apply(readModel:events:)        │
 │                                     │   StatefulProjector                  │
 │                                     │   EventTypeMapper  (Adapter)         │
 ├─────────────────────────────────────┼──────────────────────────────────────┤
 │ ENTITIES  (DDDCore)                 │ ENTITIES  (DDDCore)                  │
 │   AggregateRoot                     │   ReadModel                          │
 │   ├─ when(happened:)                │   └─ id (Codable)                    │
 │   ├─ apply(event:)                  │                                      │
 │   └─ ensureInvariant()              │                                      │
 │   DomainEvent                       │                                      │
 └─────────────────────────────────────┴──────────────────────────────────────┘

Dependency direction follows Clean Architecture: all layers depend inward toward Entities. Interface Adapters (Gateways) implement protocols defined in Use Cases/Entities — never the reverse.

Flow

Write Side (Command)

Command Handler
  │
  ├─ 1. repository.find(byId:)
  │       └─ fetches event stream from KurrentDB → replays into AggregateRoot
  │
  ├─ 2. aggregate.apply(event:)
  │       ├─ when(happened:)     mutates in-memory state
  │       ├─ ensureInvariant()   validates domain invariants
  │       └─ queues uncommitted events in AggregateRootMetadata
  │
  ├─ 3. repository.save(aggregateRoot:)
  │       └─ appends uncommitted events to KurrentDB (optimistic concurrency)
  │
  └─ 4. eventBus.postAllEvent()
          └─ publishes saved domain events to DomainEventBus subscribers

Read Side (Query)

Query Handler
  │
  └─ StatefulEventSourcingProjector.execute(input:)
        │
        ├─ 1. store.fetch(byId:)
        │       └─ loads cached ReadModel snapshot at revision N
        │
        ├─ 2. coordinator.fetchEvents(byId:, afterRevision: N)
        │       └─ retrieves only new events from KurrentDB since last snapshot
        │
        ├─ 3. apply(readModel:events:)
        │       └─ folds new events into the ReadModel
        │
        ├─ 4. store.save(readModel:, revision:)
        │       └─ persists updated snapshot to PostgreSQL or in-memory store
        │
        └─ CQRSProjectorOutput<ReadModel>

Design Principles

Separation of Concerns — write model (domain), read model (projection), and infrastructure (event store) are kept distinct and independently replaceable.

Event-First Thinking — state is never stored directly; it is always derived by replaying events. Events are the source of truth.

Explicit Domain Modeling — business logic lives in aggregate roots. Anemic models with external mutation are avoided by design.

Why Swift?

• Strong type system catches event schema mismatches at compile time
• Native async/await concurrency maps cleanly onto event stream consumption
• Swift 6 strict concurrency enables safe multi-actor architectures
• A growing server-side ecosystem (SwiftNIO, Hummingbird, gRPC) makes Swift viable for production backends

Status

This project is actively evolving. It is intended as:

• A production-capable foundation for Swift backend systems following DDD + Event Sourcing
• A reference implementation for teams exploring these patterns in Swift
• A contribution to the Swift Server ecosystem

Feedback and contributions are welcome. See open issues for planned work.

Requirements

• Swift 6.1+
• Ubuntu (Linux) / macOS 15+
• KurrentDB (for event persistence)

iOS 18+ is supported for DDDCore, EventSourcing and LocalEventStore; DDDKit and every KurrentDB, Postgres or Pulsar module are server-side.

Installation

Add the package to your Package.swift:

dependencies: [
    .package(url: "https://github.com/gradyzhuo/swift-ddd-kit.git", from: "1.4.0")
]

Then add DDDKit and KurrentSupport to your target:

.target(
    name: "MyTarget",
    dependencies: [
        .product(name: "DDDKit", package: "swift-ddd-kit"),
        .product(name: "KurrentSupport", package: "swift-ddd-kit"),
    ]
)

Project scaffolding CLI

Looking to scaffold a runnable starter project (aggregate + use cases + KurrentDB wiring)? That's now swift-pangu-cli (pangu project create), a standalone tool with no dependency on this package's library targets.

Core Concepts

1. Define Domain Events

Events are the source of truth. Every state change is captured as an immutable event.

// A creation event
struct OrderCreated: DomainEvent {
    var id: UUID = .init()
    var occurred: Date = .now
    var aggregateRootId: String
    var metadata: CustomMetadata? = nil
    let customerId: String
}

// A deletion event
struct OrderCancelled: DeletedEvent {
    var id: UUID = .init()
    var occurred: Date = .now
    let aggregateRootId: String
    var metadata: CustomMetadata? = nil

    init(id: UUID, aggregateRootId: String, occurred: Date) {
        self.id = id
        self.aggregateRootId = aggregateRootId
        self.occurred = occurred
    }
}

metadata is filled in on read from the store's metadata (see Event Metadata Pattern); leave it nil when you create an event. DeletedEvent requires init(id:aggregateRootId:occurred:) so markDelete() can create it.

2. Implement an Aggregate Root

The aggregate root is the consistency boundary. All state mutations go through apply(event:), which calls when(happened:) to update in-memory state.

final class Order: AggregateRoot {
    typealias DeletedEventType = OrderCancelled

    let id: String
    private(set) var customerId: String = ""
    var metadata: AggregateRootMetadata = .init()

    init(id: String, customerId: String) throws {
        self.id = id
        try apply(event: OrderCreated(aggregateRootId: id, customerId: customerId))
    }

    required init?(events: [any DomainEvent]) throws {
        guard let first = events.first as? OrderCreated else { return nil }
        self.id = first.aggregateRootId
        try apply(events: events)
    }

    func when(happened event: some DomainEvent) throws {
        switch event {
        case let e as OrderCreated:
            customerId = e.customerId
        default:
            break
        }
    }
}

3. Implement an Event Mapper

The mapper deserializes stored records back into typed domain events.

struct OrderEventMapper: EventTypeMapper {
    func mapping(eventData: any RecordedEventLike) throws -> (any DomainEvent)? {
        switch eventData.eventType {
        case "OrderCreated":   return try eventData.decode(to: OrderCreated.self)
        case "OrderCancelled": return try eventData.decode(to: OrderCancelled.self)
        default:               return nil
        }
    }
}

Records the mapper returns nil for are skipped. In every store that serializes events (KurrentStorageCoordinator, PersistentEventStore), when events the mapper does not know sit at a stream's tail, find returns a lower version than the stream's real tail, so save keeps throwing commitConflict and finding again does not help. This stops old code from writing after events it cannot read. Keep the mapping when you retire an event type.

4. Implement a Repository

Repositories handle persistence and retrieval through event replay.

final class OrderRepository: EventSourcingRepository {
    typealias AggregateRootType = Order
    typealias Store = KurrentStorageCoordinator<Order, CustomMetadata>

    let store: Store

    init(client: KurrentDBClient) {
        store = .init(client: client, eventMapper: OrderEventMapper())
    }
}

5. Save and Find

let client = KurrentDBClient(settings: .localhost())
let repository = OrderRepository(client: client)

// Create and save
let order = try Order(id: "order-001", customerId: "customer-42")
try await repository.save(aggregateRoot: order)

// Replay from event stream
let found = try await repository.find(byId: "order-001")

// Soft delete (marks as deleted, still retrievable with hiddingDeleted: false)
try await repository.delete(byId: "order-001")

// Hard delete (irreversible — the id can never be written again)
try await repository.purge(byId: "order-001")

purge makes the id permanently unwritable: a later append throws streamPurged (502), so never purge an id that will be reused. KurrentDB keeps the data until a scavenge; PersistentEventStore removes the payload from the stream file.

Event Metadata Pattern (Ambient Context)

Application-defined metadata flows from Usecase entry to KurrentDB via Swift's TaskLocal. Domain types (AggregateRoot, DomainEvent schemas) stay free of audit / request / tenant concerns.

// 1. Application defines its schema (or uses the bundled CustomMetadata)
struct AuditMetadata: EventMetadata {
    let operatorId: String
    let tenantId: String
}

// 2. Repository binds the schema via Store.Metadata
final class OrderRepository: EventSourcingRepository {
    typealias AggregateRootType = Order
    typealias Store = KurrentStorageCoordinator<OrderStreamNaming, AuditMetadata>
    let store: Store
    init(store: Store) { self.store = store }
}

// 3. Usecase sets the ambient at entry
struct PlaceOrderUsecase {
    let repository: OrderRepository
    func execute(input: Input) async throws -> Output {
        let meta = AuditMetadata(operatorId: input.operatorId, tenantId: input.tenantId)
        return try await EventMetadataContext<AuditMetadata>.withValue(meta) {
            let order = try Order(id: input.orderId, customerId: input.customerId)
            try await repository.save(aggregateRoot: order)
            return Output(orderId: order.id)
        }
    }
}

// 4. Read-side reads event.metadata directly (filled by the generated mapper)
func apply(readModel: inout OrderActivity, events: [any DomainEvent]) {
    for event in events {
        if let created = event as? OrderCreated, let meta = created.metadata {
            readModel.lastOperator = meta.operatorId
        }
    }
}

CustomMetadata (in KurrentSupport) is a minimal bundled schema carrying a single operatorId field — it is the default Metadata type that generator-produced events reference (typealias Metadata = CustomMetadata). Most applications replace it with their own EventMetadata-conforming struct (like AuditMetadata above).

Limits and gotchas:

• Task.detached does NOT inherit ambient context. Capture and re-apply if you spawn detached work across a metadata boundary.
• All events in one save share one metadata payload (intentional; per-event metadata variation usually signals an aggregate boundary issue).
• Store.Metadata and event.Metadata alignment is by convention, not compile- time. Runtime mismatch yields event.metadata = nil rather than a crash.
• Event-type resolution on read uses the KurrentDB-native eventType field (populated from DomainEvent.eventType at write time). The metadata payload carries no type discriminator.

CQRS — Projectors and Read Models

For the query side, implement EventSourcingProjector to fold events into a read-optimized model.

struct OrderSummary: ReadModel {
    let id: String
    var customerId: String
    var status: String
}

final class OrderProjector: EventSourcingProjector {
    typealias ReadModelType = OrderSummary
    typealias Input = OrderProjectorInput
    typealias Store = KurrentStorageCoordinator<Order, CustomMetadata>

    let store: Store

    init(client: KurrentDBClient) {
        store = .init(client: client, eventMapper: OrderEventMapper())
    }

    func buildReadModel(input: Input) throws -> OrderSummary? {
        OrderSummary(id: input.id, customerId: "", status: "unknown")
    }

    func apply(readModel: inout OrderSummary, events: [any DomainEvent]) throws {
        for event in events {
            switch event {
            case let e as OrderCreated:
                readModel.customerId = e.customerId
                readModel.status = "active"
            case is OrderCancelled:
                readModel.status = "cancelled"
            default:
                break
            }
        }
    }
}

Persistent Subscription Runner (KurrentSupport)

Replaces hand-rolled Task { for try await ... } projection handlers with a declarative runner.

import KurrentSupport
import EventSourcing

let runner = KurrentProjection.PersistentSubscriptionRunner(
    client: kdbClient,
    stream: "$ce-Order",
    groupName: "order-projection"
)
.register(orderProjectorStateful) { record in
    OrderProjectorInput(orderId: parseId(from: record))
}
.register(customerProjectorStateful) { record in
    CustomerProjectorInput(customerId: parseId(from: record))
}

try await runner.run()  // cancellation returns normally; on a thrown error, call run() again (restart loop with backoff) — ServiceGroup does not restart failed services.

• Configurable retry via RetryPolicy (default: MaxRetriesPolicy(max: 5)).
• Subscription failure throws out of run() — caller decides whether to restart.
• Returning nil from the extract closure skips that projector for the event.
• When subscribing to system streams ($ce-, $et-), create the persistent subscription with resolveLink = true so record.streamIdentifier.name references the original aggregate stream.
• See docs/superpowers/specs/2026-04-28-kurrent-projection-runner-design.md for the full design (including Phase 2: Postgres-shared-transaction box).

EventTypeFilter — pre-filter routing (optional)

When you have multiple projectors registered to the same subscription but each listens to a different subset of event types, attach an EventTypeFilter to short-circuit dispatch for unrelated event types — no extractInput, no fetch, no apply, no cursor advance.

runner
    .register(orderSummaryStateful,
              eventFilter: OrderSummaryEventFilter()) { record in        // generated from yaml
        OrderSummaryInput(id: parseId(from: record))
    }
    .register(orderRegistryStateful,
              eventFilter: OrderRegistryEventFilter()) { record in
        OrderRegistryInput(id: parseId(from: record))
    }

{ModelName}EventFilter structs are auto-generated by ModelGeneratorPlugin based on the events listed under each entry in projection-model.yaml. You can also implement EventTypeFilter yourself for custom rules:

struct OnlyTransientEvents: EventTypeFilter {
    func handles(eventType: String) -> Bool {
        eventType.hasPrefix("Transient")
    }
}

The eventFilter parameter is optional — omit it to dispatch every event through extractInput (the original Phase 1 default). See docs/superpowers/specs/2026-04-28-event-type-filter-design.md and samples/KurrentProjectionDemo/ (third projector OrderRegistry).

TransactionalSubscriptionRunner — atomic across projectors (Postgres only)

When all your read models live in the same Postgres instance and you want all-or-nothing commits per event, use KurrentProjection.TransactionalSubscriptionRunner instead of PersistentSubscriptionRunner. Every event runs all registered projectors inside a single shared PostgresClient.withTransaction block; on success the transaction commits before ack, on any failure the transaction rolls back and RetryPolicy decides nack action (same as Phase 1).

import KurrentSupport
import EventSourcing
import ReadModelPersistencePostgres
import PostgresSupport      // for the convenience init

let runner = KurrentProjection.TransactionalSubscriptionRunner(
    client: kdbClient,
    pgClient: pgClient,                            // ← convenience init
    stream: "$ce-Order",
    groupName: "order-projection"
)
.register(
    projector: orderSummaryProjector,
    storeFactory: { _ in PostgresTransactionalReadModelStore<OrderSummary>() }
) { record in
    OrderSummaryInput(id: parseId(from: record))
}
.register(
    projector: orderRegistryProjector,
    storeFactory: { _ in PostgresTransactionalReadModelStore<OrderRegistry>() },
    eventFilter: OrderRegistryEventFilter()
) { record in
    OrderRegistryInput(id: parseId(from: record))
}

try await runner.run()

Which runner to choose

	PersistentSubscriptionRunner	TransactionalSubscriptionRunner
Cross-projector consistency	Eventually consistent (each store commits independently; partial state visible during retry)	All-or-nothing per event (single tx commits or rolls back)
Backend constraint	Any ReadModelStore (in-memory, Postgres, custom)	Requires a TransactionProvider; common case is Postgres-only via PostgresTransactionProvider
Per-event overhead	One fetch + one save per registered projector	One transaction begin + N saves + one commit
Failure mode	Already-committed projectors stay committed; retry idempotent via stored cursor	Whole event rolled back; retry redoes everything from scratch
When to choose	Mixed-backend read models (PG + Redis), simple cases, no atomicity requirement	All read models in one PG; cross-projector consistency required

Both runners share RetryPolicy, EventTypeFilter, cancellation semantics, and RunnerStopped error. They differ only in commit semantics. The underlying core protocols (TransactionProvider, TransactionalReadModelStore) are abstract — future SQLite or other transactional backends ship as new provider/store implementations without touching the runner.

See the runnable example: samples/KurrentTransactionalProjectionDemo/. It includes a SIMULATE_FAILURE=once env knob that injects a one-shot projector failure to demonstrate observable rollback (no partial state).

Postgres read models

ReadModelPersistencePostgres stores read model snapshots as JSONB in a Postgres table, written by a KurrentDB persistent subscription runner. Snapshots default to a monotonic write mode (a stale or out-of-order write never rolls a snapshot back), the runner creates and recreates its subscription group itself, and KurrentProjection.rebuild replays history after any schema change. Query the table with your own SQL.

try await PostgresReadModelMigration.createTable(on: pgClient)
// creates read_model_snapshots_monotonic (the default .monotonic write mode)

let runner = KurrentProjection.TransactionalSubscriptionRunner(
    client: kurrentClient,
    pgClient: pgClient,
    stream: "$ce-Order",
    groupName: "order-summary"
)
.register(
    projector: OrderSummaryProjector(store: coordinator),
    storeFactory: { _ in PostgresTransactionalReadModelStore<OrderSummary>() },
    extractInput: { record in
        record.streamName.hasPrefix("Order-")
            ? OrderSummaryInput(id: String(record.streamName.dropFirst("Order-".count))) : nil
    }
)

• Getting Started: Postgres read models
• Snapshot write modes
• Schema evolution and rebuilding

Local Event Store (iOS and macOS apps)

LocalEventStore ships PersistentEventStore, an EventStore for apps: one JSON file per stream in a directory you choose, each file replaced atomically on every write. It needs no server. Depend on the DDDCore, EventSourcing and LocalEventStore products, which build for iOS 18+ and macOS 15+.

open the store once, then hand that instance to the repository and to every projector. The projector names the store type itself:

import DDDCore
import EventSourcing
import Foundation
import LocalEventStore

final class Order: AggregateRoot {
    typealias DeletedEventType = OrderCancelled
    static var categoryRule: StreamCategoryRule { .custom("orders") }

    let id: String
    private(set) var customerId: String = ""
    var metadata: AggregateRootMetadata = .init()

    init(id: String, customerId: String) throws {
        self.id = id
        try apply(event: OrderCreated(aggregateRootId: id, customerId: customerId))
    }

    required init?(events: [any DomainEvent]) throws {
        guard let first = events.first as? OrderCreated else { return nil }
        self.id = first.aggregateRootId
        try apply(events: events)
    }

    func when(happened event: some DomainEvent) throws {
        switch event {
        case let e as OrderCreated:
            customerId = e.customerId
        default:
            break
        }
    }
}

struct OrderProjectorInput: CQRSProjectorInput {
    let id: String
}

final class OrderRepository: EventSourcingRepository {
    typealias AggregateRootType = Order
    typealias Store = PersistentEventStore<Order, CustomMetadata>

    let store: Store

    init(store: Store) {
        self.store = store
    }
}

final class OrderProjector: EventSourcingProjector {
    typealias ReadModelType = OrderSummary
    typealias Input = OrderProjectorInput
    typealias Store = PersistentEventStore<Order, CustomMetadata>

    let store: Store

    init(store: Store) {
        self.store = store
    }

    // buildReadModel(input:) and apply(readModel:events:) as in the CQRS section
}

let directory = URL.applicationSupportDirectory
    .appending(path: "EventStore/Order", directoryHint: .isDirectory)
let store = try PersistentEventStore<Order, CustomMetadata>.open(
    directory: directory,
    eventMapper: OrderEventMapper()
)
let repository = OrderRepository(store: store)
let projector = OrderProjector(store: store)

try await repository.save(aggregateRoot: try Order(id: "order-001", customerId: "customer-42"))
let summary = try await projector.execute(input: OrderProjectorInput(id: "order-001"))

await store.close()

• One open instance per directory. Give each aggregate type (each StreamNaming/Metadata pair) its own directory. A second open of a directory that is already open throws storeUnavailable(.occupied), even with other generic arguments.

• Share the instance between the repository and the projectors, as above. On the Kurrent path, by contrast, each projector builds its own coordinator.

• Location. Put the directory under the app's own Application Support. Do not put it in:

  • an App Group or other shared container: iOS terminates a suspended app that holds a lock there
  • a cloud-synced folder
  • a network file system

• No fsync. After an OS crash or power loss the latest writes can be lost, and a stream file can come back empty or damaged. That stream alone throws storeCorrupted (504). purge clears it, but it discards that stream's events and leaves the id unwritable; it does not repair anything.

• Stream name length. The whole stream name (<category>-<id>) is at most 120 UTF-8 bytes, so ids get less than that. A longer name throws storeUnavailable(.streamNameTooLong).

• Deleting the store. Call close() before deleting the directory.

• Unmapped events at a stream's tail make save keep conflicting; see Implement an Event Mapper.

• File format. Files carry formatVersion 1. An app downgraded after a future format bump reads those streams as storeCorrupted.

• Default MainActor isolation. In a target built with .defaultIsolation(MainActor.self), mark the aggregate type nonisolated, and the projector and its read model too. Otherwise their conformances are main actor-isolated and the code does not compile: PersistentEventStore.open fails with "main actor-isolated conformance of 'Order' to 'EventStreamNaming' cannot be used in actor-isolated context" when the aggregate overrides categoryRule or category, and the repository fails with "main actor-isolated conformance of 'Order' to 'AggregateRoot' cannot satisfy conformance requirement for a 'Sendable' type parameter" even when it inherits them. Marking only the overrides () is enough for , not for a repository. Events, the mapper and the projector input conform to protocols and need nothing. does compile. Checked with Swift 6.4:

Event Migration

When event schemas evolve, MigrationUtility handles replaying old events through migration handlers without losing history.

struct MyMigration: Migration {
    typealias AggregateRootType = Order
    var eventMapper: any EventTypeMapper = LegacyOrderEventMapper()
    var migrationHandlers: [any MigrationHandler] = [
        OrderCreatedV1ToV2Handler()
    ]
}

Code Generation Plugins

swift-ddd-kit includes two SPM build-tool plugins that generate Swift boilerplate at build time.

DomainEventGeneratorPlugin

Generates typed event structs from event.yaml.

// Package.swift
.target(
    name: "MyTarget",
    plugins: [
        .plugin(name: "DomainEventGeneratorPlugin", package: "swift-ddd-kit")
    ]
)

event.yaml syntax:

OrderCreated:
  kind: createdEvent         # createdEvent | domainEvent | deletedEvent (default: domainEvent)
  aggregateRootId:
    alias: orderId           # optional alias for the aggregateRootId property
  properties:
    - name: customerId
      type: String
    - name: totalAmount
      type: Double

OrderCancelled:
  kind: deletedEvent
  aggregateRootId:
    alias: orderId

Also requires event-generator-config.yaml:

accessModifier: public       # internal | package | public
aggregateRootName: Order     # optional, customizes the generated AggregateRoot protocol name

ProjectionModelGeneratorPlugin

Generates ReadModel and EventTypeMapper boilerplate from projection-model.yaml.

// Package.swift
.target(
    name: "MyTarget",
    plugins: [
        .plugin(name: "ProjectionModelGeneratorPlugin", package: "swift-ddd-kit")
    ]
)

projection-model.yaml syntax:

OrderSummary:
  model: readModel
  createdEvent: OrderCreated
  deletedEvent: OrderCancelled
  events:
    - OrderItemAdded
    - OrderShipped

GenerateKurrentDBProjectionsPlugin

Generates KurrentDB server-side projection .js files from projection-model.yaml. These projections run inside KurrentDB and route events into per-entity derived streams that Swift projectors read from.

Unlike the build-tool plugins above, this is a Command Plugin — you run it on demand:

swift package --allow-writing-to-package-directory generate-kurrentdb-projections \
  path/to/projection-model.yaml \
  --output projections/

Or use the CLI directly:

swift run generate kurrentdb-projection \
  path/to/projection-model.yaml \
  --output projections/

YAML schema

Add category and idField to any readModel definition, and the plugin will generate a .js file for it. Definitions without category are skipped.

Field	Type	Description
category	String	KurrentDB aggregate category. Generates fromStreams(["$ce-{category}"]). Required for JS generation.
idField	String	Field in event.body used to route events to the per-entity stream. Required when any event uses the standard routing (plain string).

Each item in events / createdEvents can be:

• Plain string — standard routing via idField:

  events:
    - OrderCreated
  

• Mapping with | body — custom JS placed inside the generated function(state, event) wrapper:

  events:
    - OrderReassigned: |
        linkTo("OrderSummary-" + event.body.newOrderId, event);
  

Both forms can be mixed in the same list.

Example

# projection-model.yaml
OrderSummary:
  model: readModel
  category: Order
  idField: orderId
  createdEvents:
    - OrderCreated
  events:
    - OrderUpdated
    - OrderReassigned: |
        linkTo("OrderSummary-" + event.body.newOrderId, event);

Generated projections/OrderSummaryProjection.js:

fromStreams(["$ce-Order"])
.when({
    $init: function(){ return {} },
    OrderCreated: function(state, event) {
        if (event.isJson) {
            linkTo("OrderSummary-" + event.body["orderId"], event);
        }
    },
    OrderUpdated: function(state, event) {
        if (event.isJson) {
            linkTo("OrderSummary-" + event.body["orderId"], event);
        }
    },
    OrderReassigned: function(state, event) {
        if (event.isJson) {
            linkTo("OrderSummary-" + event.body.newOrderId, event);
        }
    },
});

Three-tier design

Tier	YAML	Output
Standard routing	category + idField + plain string events	Fully generated JS
Custom handler	Event entry with | body	Boilerplate generated, custom body embedded
Full custom	No YAML — hand-written .js	Not touched by generator

Tiers 1 and 2 can be mixed within a single definition. Hand-written .js files in projections/ coexist without conflict.

Template Variables

TemplateVariables is a small runtime product plus the VariablesGeneratorPlugin build-tool plugin. A variables.yaml declares placeholders with typed inputs; the plugin generates a Sendable protocol with one async throws -> String method per variable, plus a __value(of:inputs:) dispatch seam. Your read-model layer implements the protocol, and PlaceholderSubstitution fills %Placeholder% tokens in any template string with the resolved values.

A complete working example (the yaml, the generator config, a stub implementation and tests) lives in Sources/TemplateVariablesDemo / Tests/TemplateVariablesDemoTests. The demo is compiled and its tests run in CI, so any change that breaks the codegen chain fails the build.

variables.yaml

Each top-level key is a variable name; placeholder is the %token% matched in templates; inputs is an ordered list of name: String single-key maps that becomes the generated method's parameter list.

CustomerName:
  placeholder: CustomerName
  inputs:
    - customerId: String

InvoiceTotal:
  placeholder: InvoiceTotal
  inputs:
    - customerId: String
    - invoiceId: String

VariablesGeneratorPlugin turns this into a protocol (name from variables-generator-config.yaml) that your read-model layer implements:

public protocol DemoLetterVariablesProtocol: Sendable {
    func customerName(customerId: String) async throws -> String
    func invoiceTotal(customerId: String, invoiceId: String) async throws -> String
}

Input names must be valid Swift identifiers that are not Swift keywords, and inputs is reserved. The generated file (generated-variables.swift) also contains VariablesRuntimeError (missingInput(placeholder:input:), unknownPlaceholder) and an extension adding __value(of:inputs:), which resolves a placeholder by name from a [String: String] of inputs.

variables-generator-config.yaml

accessModifier: public
protocolName: DemoLetterVariablesProtocol

• accessModifier: the access level of the generated declarations.
• protocolName: the name of the generated protocol.

Wiring

// Package.swift
.target(
    name: "MyTarget",
    dependencies: [
        .product(name: "TemplateVariables", package: "swift-ddd-kit"),
    ],
    plugins: [
        .plugin(name: "VariablesGeneratorPlugin", package: "swift-ddd-kit"),
    ]
)

The plugin looks for variables.yaml and variables-generator-config.yaml among the target's source files.

Substituting values

PlaceholderSubstitution.substitute(_:values:) replaces every %Name% token (grammar %[A-Za-z0-9_]+%) in a single pass. Substituted values are never re-scanned, so a value that itself contains %Other% is left as-is. A placeholder with no value throws PlaceholderSubstitutionError.missingValue(placeholder:).

let variables = DemoLetterVariables()
let values = [
    "CustomerName": try await variables.__value(of: "CustomerName", inputs: ["customerId": "c-1"]),
    "InvoiceTotal": try await variables.__value(of: "InvoiceTotal", inputs: ["customerId": "c-1", "invoiceId": "i-9"]),
]
let letter = try PlaceholderSubstitution.substitute("%CustomerName% 您好，本期應付 %InvoiceTotal%。", values: values)
// "王小明 您好，本期應付 NT$1,200。"

Building further generators

TemplateVariablesDefinition exposes the variables.yaml model and parser (VariableDefinition, VariablesParser, VariablesParseError, IdentifierValidation), the VariablesProtocolGenerator and VariablesGeneratorConfiguration, so other generators can build on the same definition. The __value(of:inputs:) extension is a public contract: code generated by other generators calls it verbatim, so changing its signature or switch semantics is a breaking change.

Cross-Context Events (Pulsar)

Two modules move Published Language events between bounded contexts over Apache Pulsar, without either side depending on a Pulsar client library:

• ContextForwarder (produce side) — PulsarRESTPublisher posts a PublishedLanguageEvent to Pulsar's REST produce endpoint (/topics/persistent/{tenant}/{namespace}/{topic}). No Pulsar SDK, no binary protocol — just AsyncHTTPClient.
• ContextReceiver (consume side, portable) + ContextReceiverWebSocket (Pulsar WebSocket transport) — ContextReceiver drives a PulsarMessageSource into a PublishedLanguageHandler, handling flow control and settlement. WebSocketMessageSource is the concrete transport, talking to Pulsar's WebSocket consumer endpoint (/ws/v2/consumer/...).

The payload asymmetry

Pulsar's REST produce endpoint accepts a raw JSON string as the message payload; its WebSocket consumer endpoint hands that same payload back base64-encoded. PulsarRESTPublisher.produceBody sends the raw string; ConsumerFrame.decodedEvent() base64-decodes before parsing. Get either side wrong and every message fails silently — this is exactly the kind of defect that only surfaces against a live broker, which is why Tests/ContextReceiverIntegrationTests/LivePulsarTests.swift exists: it posts through the real REST endpoint and asserts the event survives the trip through a real WebSocket consumer, unlike every other test in this plan, which either builds the REST body by hand or feeds hand-built base64 into the decoder.

Why pullMode is mandatory

ConsumerEndpoint.url always sets pullMode=true and does not expose it as a setting. In push mode, Pulsar streams messages as fast as the socket accepts them, with no host-side backpressure; in pull mode, the client must explicitly grant permits (ContextReceiver.FlowSettings.initialPermits/permitRefillThreshold) for the broker to deliver more. Pull mode is what makes the flow-control contract in ContextReceiver meaningful at all — without it, a slow handler has no way to signal "stop sending".

At-least-once delivery, host-side dedup

Pulsar redelivers on ack timeout or nack, so a PublishedLanguageHandler may see the same eventId more than once (ReceivedRecord.isRedelivery / redeliveryCount surface this for logging). The framework does not deduplicate on the consumer's behalf — hosts are expected to derive deterministic downstream aggregate/entity ids from the upstream eventId so that reapplying the same event is naturally idempotent, rather than tracking a separate dedup table.

Reading custom metadata in a rule

ForwardedRecord.metadata carries the KurrentDB customMetadata bytes the writing context attached (the JSON encoding of its EventMetadata struct — see the pluggable-metadata design). The kit does not know that schema; a rule decodes it with its own type:

struct Operator: Decodable { let operatorId: String }

ForwardingRule(eventTypes: ["CollaboratorAdded"]) { record in
    let body = try record.decodeBody(CollaboratorAdded.self)
    let actor = try record.decodeMetadata(Operator.self)   // nil when the write carried no metadata
    …
}

Semantics mirror decodeBody: absent metadata is nil, never an error (events written before a context started attaching metadata stay forwardable); bytes that are present but do not decode as the requested type are ForwardingError.permanent and park the record, because redelivery cannot repair them. A rule that wants leniency declares optional fields on its own type. This is deliberately stricter than aggregate loading: the generated EventMapper decodes metadata with try? and leaves event.metadata nil on failure, whereas the forwarder parks the record. The practical consequence is that a new rule declaring a required field which older events encoded under a different shape will park every such event on replay, so declare the field optional on the rule's type when old events may lack it.

Dead letter mapping

Pulsar's WebSocket API has no "send to DLQ now" command. ContextReceiver maps a permanent failure (an undecodable payload, or a handler classifying its own error as non-retryable) to ReceiveDisposition.dropToDeadLetter, which the transport implements as a negative-acknowledge. The message only actually parks in the dead letter topic once maxRedeliverCount is exhausted — both maxRedeliverCount and deadLetterTopic must be set on ConsumerEndpoint.Settings, or a message the host has explicitly given up on will redeliver forever instead of parking. DeadLetterMonitor polls the DLQ's backlog via the admin API so a growing park pile can page someone.

Platforms

ContextReceiverWebSocket builds on Linux and macOS 15+ (swift-websocket 1.5.x compiles on the macOS 26 SDK too). Its offline tests run everywhere; only the live-broker suite needs a running Pulsar.

The WebSocket transport is single-use — supervision is the host's job

WebSocketMessageSource wraps exactly one socket session: one instance == one connection to Pulsar's WebSocket consumer endpoint, and run() may be called exactly once. Call it a second time on the same instance and it throws ReceiveError.transportUnavailable immediately, rather than reconnecting or silently reusing state.

This is deliberate, not a missing feature. A transparent internal reconnect would open a brand-new broker-side consumer session, and a new session's permit budget starts at zero — but ContextReceiver.run() grants FlowSettings.initialPermits exactly once, before it ever enters its receive loop. If the transport reconnected behind that loop's back, the loop would keep await-ing frames from a session nobody ever granted permits to, and the consumer would silently stall forever — no error, no crash, just no more messages. That failure mode is worse than a thrown error, because nothing tells the host it happened.

So: retrying the same instance is the trap. run() on an already-run WebSocketMessageSource will throw, and there is no way to revive a finished frame stream — frames() always returns the same AsyncThrowingStream, and once it has completed, a fresh socket underneath it has nowhere to deliver into. Recovery means discard the instance and construct a fresh one, then restart ContextReceiver alongside it so initialPermits gets re-granted against the new broker session. Supervising that lifecycle — catching the throw, backing off, rebuilding both objects — is the host's responsibility; neither WebSocketMessageSource nor ContextReceiver do it for you.

This is the same shape as ForwarderGroup.runWithRestart on the produce side, with one crucial difference: runWithRestart retries body() — the same ContextForwarder instance — because a forwarder's underlying subscription survives a clean stream end. The receive side cannot do that; the socket itself is single-use, so each retry attempt must build new WebSocketMessageSource and ContextReceiver instances, not re-invoke the old ones. Copying ForwarderGroup's shape verbatim — retrying a captured receiver.run() closure — reproduces exactly the permit-starvation stall this contract exists to prevent.

func runReceiverWithRestart(
    endpoint: ConsumerEndpoint,
    handler: some PublishedLanguageHandler,
    restartDelay: Duration = .seconds(5),
    logger: Logger
) async {
    while !Task.isCancelled {
        // A fresh socket + a fresh receiver every attempt — never reuse
        // either instance across a restart. This is what re-grants
        // `initialPermits` against the new broker-side consumer session.
        let source = WebSocketMessageSource(
            endpoint: endpoint,
            authorizationHeader: { "Bearer \(try await fetchAccessToken())" },
            logger: logger
        )
        let receiver = ContextReceiver(source: source, handler: handler, logger: logger)

        do {
            try await withThrowingTaskGroup(of: Void.self) { group in
                group.addTask { try await source.run() }
                group.addTask { try await receiver.run() }
                try await group.next()
                group.cancelAll()
            }
            if Task.isCancelled { return }
            logger.warning("receiver stream ended — restarting in \(restartDelay)")
        } catch {
            if Task.isCancelled { return }
            logger.error("receiver stopped: \(error) — restarting in \(restartDelay)")
        }
        try? await Task.sleep(for: restartDelay)
    }
}

Running the live integration suite

Tests/ContextReceiverIntegrationTests/LivePulsarTests.swift is gated on the PULSAR_HTTP_URL environment variable so a plain swift test never needs a broker. To run it locally:

docker compose -f docker-compose.pulsar.yml up -d
until curl -sf http://localhost:8080/admin/v2/brokers/health >/dev/null; do sleep 2; done
docker run --rm --network host -v "$PWD":/src -w /src \
  -e PULSAR_HTTP_URL=http://localhost:8080 \
  swift:6.2-noble \
  bash -c 'swift test --filter ContextReceiverIntegrationTests'

This suite needs a live broker, so it deliberately stays out of CI. .github/workflows/swift-build-testing.yml runs the offline suites, builds the DDDCore, EventSourcing and LocalEventStore products for iOS (build-ios), and runs KurrentLiveContractTests against kurrentplatform/kurrentdb:26.1.3 (kurrent-live-contract); the Pulsar broker is not part of it.

Modules

Module	Purpose
DDDKit	Umbrella import
DDDCore	Core protocols: Entity, AggregateRoot, DomainEvent, DomainEventBus
EventSourcing	Abstract patterns: EventStore, EventSourcingRepository, EventSourcingProjector, InMemoryEventStore, RecordedEventLike, EventTypeMapper, CustomMetadata
LocalEventStore	PersistentEventStore: a file-backed EventStore for apps, one JSON file per stream
KurrentSupport	KurrentDB adapter: KurrentStorageCoordinator
EventBus	In-memory event bus for local event distribution
MigrationUtility	Event schema migration framework
ReadModelPersistence	ReadModelStore protocol + in-memory store for read model snapshots
ReadModelPersistencePostgres	PostgreSQL + JSONB backed ReadModelStore (optional dependency)
TestUtility	Test helpers: TestBundle, stream cleanup utilities
PublishedLanguage	PublishedLanguageEvent — the wire envelope shared by both sides of cross-context Pulsar events
ContextForwarder	Produce side: PulsarRESTPublisher, ForwarderGroup — posts events over Pulsar's REST endpoint, no Pulsar SDK
ContextReceiver	Consume side (portable): ContextReceiver runner, PulsarMessageSource transport seam, DeadLetterMonitor
ContextReceiverWebSocket	WebSocketMessageSource, the concrete Pulsar WebSocket consumer transport
TemplateVariables	Runtime substitution engine: PlaceholderSubstitution

iOS 18+ is supported for DDDCore, EventSourcing and LocalEventStore; DDDKit and every KurrentDB, Postgres or Pulsar module are server-side.

License

MIT