opinionated-machine

Very opinionated DI framework for fastify, built on top of awilix

Table of Contents

• Basic usage
  • Managing global public dependencies across modules
  • Avoiding circular dependencies in typed cradle parameters
• Defining controllers
• Putting it all together
• Resolver Functions
  • Basic Resolvers
    • asSingletonClass
    • asSingletonFunction
    • asClassWithConfig
  • Domain Layer Resolvers
    • asServiceClass
    • asUseCaseClass
    • asRepositoryClass
    • asControllerClass
    • asSSEControllerClass
    • asDualModeControllerClass
  • Message Queue Resolvers
    • asMessageQueueHandlerClass
  • Background Job Resolvers
    • asEnqueuedJobWorkerClass
    • asPgBossProcessorClass
    • asPeriodicJobClass
    • asJobQueueClass
    • asEnqueuedJobQueueManagerFunction
• Server-Sent Events (SSE)
  • Prerequisites
  • Defining SSE Contracts
  • Creating SSE Controllers
  • Type-Safe SSE Handlers with buildHandler
  • SSE Controllers Without Dependencies
  • Registering SSE Controllers
  • Registering SSE Routes
  • Broadcasting Events
  • Controller-Level Hooks
  • Route-Level Options
  • Graceful Shutdown
  • Error Handling
  • Long-lived Connections vs Request-Response Streaming
  • SSE Parsing Utilities
    • parseSSEEvents
    • parseSSEBuffer
    • ParsedSSEEvent Type
  • Testing SSE Controllers
  • SSESessionSpy API
  • Session Monitoring
  • SSE Rooms
    • Enabling Rooms
    • Session Room Operations
    • Broadcasting to Rooms
    • Room Broadcaster (Decoupled Broadcasting)
    • Room Name Helpers
    • Room Query Methods
    • Auto-Leave on Disconnect
    • Multi-Node Deployments with Redis
  • SSE Test Utilities
    • Quick Reference
    • Inject vs HTTP Comparison
    • SSEHttpClient
    • SSEInjectClient
    • Contract-Aware Inject Helpers
• Dual-Mode Controllers (SSE + Sync)
  • Overview
  • Defining Dual-Mode Contracts
  • Response Headers (Sync Mode)
  • Status-Specific Response Schemas (responseBodySchemasByStatusCode)
  • Implementing Dual-Mode Controllers
  • Registering Dual-Mode Controllers
  • Accept Header Routing
  • Testing Dual-Mode Controllers

Basic usage

Define a module, or several modules, that will be used for resolving dependency graphs, using awilix:

import { AbstractModule, type InferModuleDependencies, asSingletonClass, asMessageQueueHandlerClass, asEnqueuedJobWorkerClass, asJobQueueClass, asControllerClass } from 'opinionated-machine'

export class MyModule extends AbstractModule {
    resolveDependencies(
        diOptions: DependencyInjectionOptions,
    ) {
        return {
            service: asSingletonClass(Service),

            // by default init and disposal methods from `message-queue-toolkit` consumers
            // will be assumed. If different values are necessary, pass second config object
            // and specify "asyncInit" and "asyncDispose" fields
            messageQueueConsumer: asMessageQueueHandlerClass(MessageQueueConsumer, {
                queueName: MessageQueueConsumer.QUEUE_ID,
                diOptions,
            }),

            // by default init and disposal methods from `background-jobs-commons` job workers
            // will be assumed. If different values are necessary, pass second config object
            // and specify "asyncInit" and "asyncDispose" fields
            jobWorker: asEnqueuedJobWorkerClass(JobWorker, {
                queueName: JobWorker.QUEUE_ID,
                diOptions,
            }),

            // by default disposal methods from `background-jobs-commons` job queue manager
            // will be assumed. If different values are necessary, specify "asyncDispose" fields
            // in the second config object
            queueManager: asJobQueueClass(
                QueueManager,
                {
                    diOptions,
                },
                {
                    asyncInit: (manager) => manager.start(resolveJobQueuesEnabled(options)),
                },
            ),
        }
    }

    // controllers will be automatically registered on fastify app
    // both REST and SSE controllers go here - SSE controllers are auto-detected
    resolveControllers(diOptions: DependencyInjectionOptions) {
        return {
            controller: asControllerClass(MyController),
        }
    }
}

// Dependencies are inferred from the return type of resolveDependencies()
export type ModuleDependencies = InferModuleDependencies<MyModule>

The InferModuleDependencies utility type extracts the dependency types from the resolvers returned by resolveDependencies(), so you don't need to maintain a separate type manually.

When a module is used as a secondary module, only resolvers marked as public (asServiceClass, asUseCaseClass, asJobQueueClass, asEnqueuedJobQueueManagerFunction) are exposed. Use InferPublicModuleDependencies to infer only the public dependencies (private ones are omitted entirely):

// Inferred as { service: Service } — private resolvers are omitted
export type MyModulePublicDependencies = InferPublicModuleDependencies<MyModule>

Managing global public dependencies across modules

When your application has multiple secondary modules, you need a single type that combines all their public dependencies. The library exports an empty PublicDependencies interface that each module can augment via TypeScript's module augmentation. Each module file adds its own public deps to this shared interface using declare module. The augmentations are project-wide — they apply everywhere as long as the augmenting file is part of your TypeScript compilation (included in tsconfig.json), with no explicit import chain required.

Start with a CommonModule that provides shared infrastructure dependencies (logger, config, etc.), then add domain modules that each augment the same interface independently.

// CommonModule.ts — shared infrastructure
import { AbstractModule, type InferPublicModuleDependencies } from 'opinionated-machine'

export class CommonModule extends AbstractModule {
  resolveDependencies(diOptions: DependencyInjectionOptions) {
    return {
      config: asSingletonFunction((): Config => loadConfig()),     // private — omitted
      logger: asServiceClass(Logger),                      // public
      eventEmitter: asServiceClass(AppEventEmitter),       // public
    }
  }
}

declare module 'opinionated-machine' {
  interface PublicDependencies extends InferPublicModuleDependencies<CommonModule> {}
}

// UsersModule.ts — no need to import CommonModule's type
import { AbstractModule, type InferPublicModuleDependencies } from 'opinionated-machine'

export class UsersModule extends AbstractModule {
  resolveDependencies(diOptions: DependencyInjectionOptions) {
    return {
      userService: asServiceClass(UserService),         // public
      userRepository: asRepositoryClass(UserRepository), // private — omitted
    }
  }
}

declare module 'opinionated-machine' {
  interface PublicDependencies extends InferPublicModuleDependencies<UsersModule> {}
}

// BillingModule.ts — independent, no chain
import { AbstractModule, type InferPublicModuleDependencies } from 'opinionated-machine'

export class BillingModule extends AbstractModule {
  resolveDependencies(diOptions: DependencyInjectionOptions) {
    return {
      billingService: asServiceClass(BillingService),       // public
      paymentGateway: asRepositoryClass(PaymentGateway),     // private — omitted
    }
  }
}

declare module 'opinionated-machine' {
  interface PublicDependencies extends InferPublicModuleDependencies<BillingModule> {}
}

Importing PublicDependencies from anywhere gives you the full accumulated type: { logger: Logger; eventEmitter: AppEventEmitter; userService: UserService; billingService: BillingService }. Private dependencies (config, userRepository, paymentGateway) are omitted automatically. No explicit import chain between modules is needed — each module augments the interface independently.

Typing constructor dependencies within a module

Classes within a module can access both the module's own dependencies (including private ones like repositories) and all public dependencies from other modules. Combine InferModuleDependencies with PublicDependencies to get the full cradle type available at runtime:

// UsersModule.ts
import {
  AbstractModule,
  type InferModuleDependencies,
  type InferPublicModuleDependencies,
  type PublicDependencies,
} from 'opinionated-machine'

// Module's own deps (public + private) merged with all public deps from other modules
type UsersModuleInjectables = InferModuleDependencies<UsersModule> & PublicDependencies

export class UserService {
  private readonly repository: UserRepository
  private readonly logger: Logger  // from CommonModule's public deps

  constructor(dependencies: UsersModuleInjectables) {
    this.repository = dependencies.userRepository  // own private dep — accessible
    this.logger = dependencies.logger              // public dep from another module — accessible
    // dependencies.billingRepository              // private dep from another module — type error
  }
}

class UserRepository {}

export class UsersModule extends AbstractModule {
  resolveDependencies(diOptions: DependencyInjectionOptions) {
    return {
      userService: asServiceClass(UserService),
      userRepository: asRepositoryClass(UserRepository),
    }
  }
}

declare module 'opinionated-machine' {
  interface PublicDependencies extends InferPublicModuleDependencies<UsersModule> {}
}

This gives each class access to exactly what the DI container provides at runtime: the module's own registered dependencies plus all public dependencies from secondary modules. Private dependencies from other modules are excluded at the type level, matching the runtime behavior.

Constructing the combined dependency type for DIContext

Use PublicDependencies when building the full dependency type:

import type { PublicDependencies } from 'opinionated-machine'

type Dependencies = InferModuleDependencies<PrimaryModule> & PublicDependencies

Avoiding circular dependencies in typed cradle parameters

Because InferModuleDependencies is inferred from the module's own resolveDependencies() return type, classes and functions that reference it inside the same module could create a circular type dependency. The library handles this automatically for class-based resolvers. For function-based resolvers, use the indexed access pattern described below.

Class-based resolvers (recommended — works automatically)

All class-based resolver functions (asSingletonClass, asServiceClass, asRepositoryClass, etc.) use a ClassValue<T> type internally, which infers the instance type from the class's prototype property rather than its constructor signature. This means classes can freely reference InferModuleDependencies in their constructors without causing circular type dependencies:

import { AbstractModule, type InferModuleDependencies, asServiceClass, asSingletonClass } from 'opinionated-machine'

export class MyService {
  // Constructor references ModuleDependencies — no circular dependency!
  constructor({ myHelper }: ModuleDependencies) {
    // myHelper is fully typed as MyHelper
  }
}

export class MyHelper {
  process() {}
}

export class MyModule extends AbstractModule {
  resolveDependencies(diOptions: DependencyInjectionOptions) {
    return {
      myService: asServiceClass(MyService),   // ClassValue<T> breaks the cycle
      myHelper: asSingletonClass(MyHelper),
    }
  }
}

export type ModuleDependencies = InferModuleDependencies<MyModule>

Prefer class-based resolvers wherever possible — they provide full type safety with no any fallback and no extra annotations needed.

Function-based resolvers (asSingletonFunction)

Function-based resolvers (asSingletonFunction) cannot use the ClassValue<T> trick because functions don't have a prototype property that separates return type from parameter types. Use indexed access on InferModuleDependencies to type individual dependencies, and always provide an explicit return type annotation on the factory function:

import { S3Client } from '@aws-sdk/client-s3'

// Inside resolveDependencies():
config: asSingletonClass(Config),
logger: asServiceClass(Logger),

s3Client: asSingletonFunction(
  ({ config, logger }: {
    config: ModuleDependencies['config']
    logger: ModuleDependencies['logger']
  }): S3Client => {
    return new S3Client({
      region: config.awsRegion,
      credentials: { accessKeyId: config.awsAccessKey, secretAccessKey: config.awsSecretKey },
      logger,
    })
  },
),

// ...

// At the bottom of the file:
export type ModuleDependencies = InferModuleDependencies<MyModule>

Indexed access types (ModuleDependencies['config']) are resolved lazily by TypeScript — it looks up individual properties without computing the entire ModuleDependencies type, avoiding the cycle. Each dependency stays in sync with the module's resolvers automatically.

For cross-module dependencies, use InferPublicModuleDependencies:

type CommonDeps = InferPublicModuleDependencies<CommonModule>

redis: asSingletonFunction(
  ({ config }: { config: CommonDeps['config'] }): Redis => {
    return new Redis({ host: config.redis.host, port: config.redis.port })
  },
),

The explicit return type is critical. Without it, TypeScript attempts to infer the return type from the function body, which requires resolving the parameter types, which triggers the circular reference:

// BREAKS — no explicit return type, TypeScript infers it from the body,
// requiring config's type to be resolved, triggering the cycle:
s3Client: asSingletonFunction(
  ({ config }: { config: ModuleDependencies['config'] }) => {
    return new S3Client({ region: config.awsRegion })
  },
),

Note: Pick<ModuleDependencies, 'a' | 'b'> does not work — Pick requires keyof ModuleDependencies, which forces TypeScript to resolve the entire type and triggers the circular reference. Each property must be accessed individually via indexed access.

Alternative: concrete parameter types

You can use concrete types instead of indexed access when the return type is dynamic or difficult to spell out explicitly. Because concrete types don't reference InferModuleDependencies, there is no circularity, so TypeScript can infer the return type for you:

// Return type inferred automatically — Config is a concrete type that doesn't
// reference InferModuleDependencies, so there's no circular reference.
redisConfig: asSingletonFunction(
  ({ config }: { config: Config }) => {
    return config.getRedisConfig()
  },
),

The trade-off is that parameter types won't auto-sync if the module's resolver changes — but you'll still get a type error at the resolver level if the types diverge.

Fallback: class wrapper

If the adapter needs many dependencies and the inline syntax becomes too verbose, wrap the adaptation logic in a class and use asSingletonClass instead. The constructor can reference ModuleDependencies directly since ClassValue<T> breaks the cycle automatically — no return type annotation needed:

import { S3Client } from '@aws-sdk/client-s3'

// Full adapter — adds domain-specific methods:
class S3StorageAdapter {
  private readonly client: S3Client

  constructor({ config, logger }: ModuleDependencies) {
    this.client = new S3Client({
      region: config.awsRegion,
      credentials: { accessKeyId: config.awsAccessKey, secretAccessKey: config.awsSecretKey },
      logger,
    })
  }

  async upload(bucket: string, key: string, body: Buffer): Promise<string> {
    await this.client.send(new PutObjectCommand({ Bucket: bucket, Key: key, Body: body }))
    return `https://${bucket}.s3.amazonaws.com/${key}`
  }
}

// In resolveDependencies():
s3StorageAdapter: asSingletonClass(S3StorageAdapter),

// Thin wrapper — just bridges the constructor signature:
class S3ClientProvider {
  readonly client: S3Client

  constructor({ config, logger }: ModuleDependencies) {
    this.client = new S3Client({
      region: config.awsRegion,
      credentials: { accessKeyId: config.awsAccessKey, secretAccessKey: config.awsSecretKey },
      logger,
    })
  }
}

// In resolveDependencies():
s3ClientProvider: asSingletonClass(S3ClientProvider),

// Consumers access the original instance directly:
// this.s3ClientProvider.client.send(new PutObjectCommand({ ... }))

This is more heavyweight than a function resolver but provides full type safety with no explicit return type needed, and scales cleanly to any number of dependencies.

You can also use the explicit generic pattern if you prefer (e.g. for isolatedDeclarations mode):

export type ModuleDependencies = {
    service: Service
    messageQueueConsumer: MessageQueueConsumer
    jobWorker: JobWorker
    queueManager: QueueManager
}

export class MyModule extends AbstractModule<ModuleDependencies, ExternalDependencies> {
    resolveDependencies(
        diOptions: DependencyInjectionOptions,
        _externalDependencies: ExternalDependencies,
    ): MandatoryNameAndRegistrationPair<ModuleDependencies> {
        return { /* ... */ }
    }
}

Defining controllers

Controllers require using fastify-api-contracts and allow to define application routes.

import { buildFastifyRoute } from '@lokalise/fastify-api-contracts'
import { buildRestContract } from '@lokalise/api-contracts'
import { z } from 'zod/v4'
import { AbstractController } from 'opinionated-machine'

const BODY_SCHEMA = z.object({})
const PATH_PARAMS_SCHEMA = z.object({
  userId: z.string(),
})

const contract = buildRestContract({
  method: 'delete',
  successResponseBodySchema: BODY_SCHEMA,
  requestPathParamsSchema: PATH_PARAMS_SCHEMA,
  pathResolver: (pathParams) => `/users/${pathParams.userId}`,
})

export class MyController extends AbstractController<typeof MyController.contracts> {
  public static contracts = { deleteItem: contract } as const
  private readonly service: Service

  constructor({ service }: ModuleDependencies) {
      super()
      this.service = testService
  }

    private deleteItem = buildFastifyRoute(
        TestController.contracts.deleteItem,
        async (req, reply) => {
            req.log.info(req.params.userId)
            this.service.execute()
            await reply.status(204).send()
        },
    )

    public buildRoutes() {
        return {
            deleteItem: this.deleteItem,
        }
    }
}

Putting it all together

Typical usage with a fastify app looks like this:

import { serializerCompiler, validatorCompiler } from 'fastify-type-provider-zod'
import { createContainer } from 'awilix'
import { fastify } from 'fastify'
import { DIContext } from 'opinionated-machine'

const module = new MyModule()
const container = createContainer({
    injectionMode: 'PROXY',
})

type AppConfig = {
    DATABASE_URL: string
    // ...
    // everything related to app configuration
}

type ExternalDependencies = {
    logger: Logger // most likely you would like to reuse logger instance from fastify app
}

const context = new DIContext<ModuleDependencies, AppConfig, ExternalDependencies>(container, {
    messageQueueConsumersEnabled: [MessageQueueConsumer.QUEUE_ID],
    jobQueuesEnabled: false,
    jobWorkersEnabled: false,
    periodicJobsEnabled: false,
})

context.registerDependencies({
    modules: [module],
    dependencyOverrides: {}, // dependency overrides if necessary, usually for testing purposes
    configOverrides: {}, // config overrides if necessary, will be merged with value inside existing config
    configDependencyId?: string // what is the dependency id in the graph for the config entity. Only used for config overrides. Default value is `config`
}, 
    // external dependencies that are instantiated outside of DI
    {
    logger: app.logger
})

const app = fastify()
app.setValidatorCompiler(validatorCompiler)
app.setSerializerCompiler(serializerCompiler)

app.after(() => {
    context.registerRoutes(app)
})
await app.ready()

Resolver Functions

The library provides a set of resolver functions that wrap awilix's asClass and asFunction with sensible defaults for different types of dependencies. All resolvers create singletons by default.

Basic Resolvers

asSingletonClass(Type, opts?)

Basic singleton class resolver. Use for general-purpose dependencies that don't fit other categories.

service: asSingletonClass(MyService)

asSingletonFunction(fn, opts?)

Basic singleton function resolver. Use when you need to resolve a dependency using a factory function.

config: asSingletonFunction((): Config => loadConfig())

asClassWithConfig(Type, config, opts?)

Register a class with an additional config parameter passed to the constructor. Uses asFunction wrapper internally to pass the config as a second parameter. Requires PROXY injection mode.

myService: asClassWithConfig(MyService, { enableFeature: true })

The class constructor receives dependencies as the first parameter and config as the second:

class MyService {
  constructor(deps: Dependencies, config: { enableFeature: boolean }) {
    // ...
  }
}

Domain Layer Resolvers

asServiceClass(Type, opts?)

For service classes. Marks the dependency as public (exposed when module is used as secondary).

userService: asServiceClass(UserService)

asUseCaseClass(Type, opts?)

For use case classes. Marks the dependency as public.

createUserUseCase: asUseCaseClass(CreateUserUseCase)

asRepositoryClass(Type, opts?)

For repository classes. Marks the dependency as private (not exposed when module is secondary).

userRepository: asRepositoryClass(UserRepository)

asControllerClass(Type, opts?)

For REST controller classes. Marks the dependency as private. Use in resolveControllers().

userController: asControllerClass(UserController)

asSSEControllerClass(Type, sseOptions?, opts?)

For SSE controller classes. Marks the dependency as private with isSSEController: true for auto-detection. Automatically configures closeAllConnections as the async dispose method for graceful shutdown. When sseOptions.diOptions.isTestMode is true, enables the connection spy for testing. Use in resolveControllers() alongside REST controllers.

// In resolveControllers()
resolveControllers(diOptions: DependencyInjectionOptions) {
  return {
    userController: asControllerClass(UserController),
    notificationsSSEController: asSSEControllerClass(NotificationsSSEController, { diOptions }),
  }
}

asDualModeControllerClass(Type, sseOptions?, opts?)

For dual-mode controller classes that handle both SSE and JSON responses on the same route. Marks the dependency as private with isDualModeController: true for auto-detection. Inherits all SSE controller features including connection management and graceful shutdown. When sseOptions.diOptions.isTestMode is true, enables the connection spy for testing SSE mode.

// In resolveControllers()
resolveControllers(diOptions: DependencyInjectionOptions) {
  return {
    userController: asControllerClass(UserController),
    chatController: asDualModeControllerClass(ChatDualModeController, { diOptions }),
  }
}

Message Queue Resolvers

asMessageQueueHandlerClass(Type, mqOptions, opts?)

For message queue consumers following message-queue-toolkit conventions. Automatically handles start/close lifecycle and respects messageQueueConsumersEnabled option.

messageQueueConsumer: asMessageQueueHandlerClass(MessageQueueConsumer, {
    queueName: MessageQueueConsumer.QUEUE_ID,
    diOptions,
})

Background Job Resolvers

asEnqueuedJobWorkerClass(Type, workerOptions, opts?)

For enqueued job workers following background-jobs-common conventions. Automatically handles start/dispose lifecycle and respects enqueuedJobWorkersEnabled option.

jobWorker: asEnqueuedJobWorkerClass(JobWorker, {
    queueName: JobWorker.QUEUE_ID,
    diOptions,
})

asPgBossProcessorClass(Type, processorOptions, opts?)

For pg-boss job processor classes. Similar to asEnqueuedJobWorkerClass but uses start/stop lifecycle methods and initializes after pgBoss (priority 20).

enrichUserPresenceJob: asPgBossProcessorClass(EnrichUserPresenceJob, {
    queueName: EnrichUserPresenceJob.QUEUE_ID,
    diOptions,
})

asPeriodicJobClass(Type, workerOptions, opts?)

For periodic job classes following background-jobs-common conventions. Uses eager injection via register method and respects periodicJobsEnabled option.

cleanupJob: asPeriodicJobClass(CleanupJob, {
    jobName: CleanupJob.JOB_NAME,
    diOptions,
})

asJobQueueClass(Type, queueOptions, opts?)

For job queue classes. Marks the dependency as public. Respects jobQueuesEnabled option.

queueManager: asJobQueueClass(QueueManager, {
    diOptions,
})

asEnqueuedJobQueueManagerFunction(fn, diOptions, opts?)

For job queue manager factory functions. Automatically calls start() with resolved enabled queues during initialization.

jobQueueManager: asEnqueuedJobQueueManagerFunction(
    createJobQueueManager,
    diOptions,
)

Server-Sent Events (SSE)

The library provides first-class support for Server-Sent Events using @fastify/sse. SSE enables real-time, unidirectional streaming from server to client - perfect for notifications, live updates, and streaming responses (like AI chat completions).

Prerequisites

Register the @fastify/sse plugin before using SSE controllers:

import FastifySSEPlugin from '@fastify/sse'

const app = fastify()
await app.register(FastifySSEPlugin)

Defining SSE Contracts

Use buildSseContract from @lokalise/api-contracts to define SSE routes. The method field determines the HTTP method. Paths are defined using pathResolver, a type-safe function that receives typed params and returns the URL path:

import { z } from 'zod'
import { buildSseContract } from '@lokalise/api-contracts'

// GET-based SSE stream with path params
export const channelStreamContract = buildSseContract({
  method: 'get',
  pathResolver: (params) => `/api/channels/${params.channelId}/stream`,
  requestPathParamsSchema: z.object({ channelId: z.string() }),
  requestQuerySchema: z.object({}),
  requestHeaderSchema: z.object({}),
  serverSentEventSchemas: {
    message: z.object({ content: z.string() }),
  },
})

// GET-based SSE stream without path params
export const notificationsContract = buildSseContract({
  method: 'get',
  pathResolver: () => '/api/notifications/stream',
  requestPathParamsSchema: z.object({}),
  requestQuerySchema: z.object({ userId: z.string().optional() }),
  requestHeaderSchema: z.object({}),
  serverSentEventSchemas: {
    notification: z.object({
      id: z.string(),
      message: z.string(),
    }),
  },
})

// POST-based SSE stream (e.g., AI chat completions)
export const chatCompletionContract = buildSseContract({
  method: 'post',
  pathResolver: () => '/api/chat/completions',
  requestPathParamsSchema: z.object({}),
  requestQuerySchema: z.object({}),
  requestHeaderSchema: z.object({}),
  requestBodySchema: z.object({
    message: z.string(),
    stream: z.literal(true),
  }),
  serverSentEventSchemas: {
    chunk: z.object({ content: z.string() }),
    done: z.object({ totalTokens: z.number() }),
  },
})

For reusable event schema definitions, you can use the SSEEventSchemas type (requires TypeScript 4.9+ for satisfies):

import { z } from 'zod'
import type { SSEEventSchemas } from 'opinionated-machine'

// Define reusable event schemas for multiple contracts
const streamingEvents = {
  chunk: z.object({ content: z.string() }),
  done: z.object({ totalTokens: z.number() }),
  error: z.object({ code: z.number(), message: z.string() }),
} satisfies SSEEventSchemas

Creating SSE Controllers

SSE controllers extend AbstractSSEController and must implement a two-parameter constructor. Use buildHandler for automatic type inference of request parameters:

import {
  AbstractSSEController,
  buildHandler,
  type SSEControllerConfig,
  type SSESession
} from 'opinionated-machine'

type Contracts = {
  notificationsStream: typeof notificationsContract
}

type Dependencies = {
  notificationService: NotificationService
}

export class NotificationsSSEController extends AbstractSSEController<Contracts> {
  public static contracts = {
    notificationsStream: notificationsContract,
  } as const

  private readonly notificationService: NotificationService

  // Required: two-parameter constructor (deps object, optional SSE config)
  constructor(deps: Dependencies, sseConfig?: SSEControllerConfig) {
    super(deps, sseConfig)
    this.notificationService = deps.notificationService
  }

  public buildSSERoutes() {
    return {
      notificationsStream: this.handleStream,
    }
  }

  // Handler with automatic type inference from contract
  // sse.start(mode) returns a session with type-safe event sending
  // Options (onConnect, onClose) are passed as the third parameter to buildHandler
  private handleStream = buildHandler(notificationsContract, {
    sse: async (request, sse) => {
      // request.query is typed from contract: { userId?: string }
      const userId = request.query.userId ?? 'anonymous'

      // Start streaming with 'keepAlive' mode - stays open for external events
      // Sends HTTP 200 + SSE headers immediately
      const session = sse.start('keepAlive', { context: { userId } })

      // For external triggers (subscriptions, timers, message queues), use sendEventInternal.
      // session.send is only available within this handler's scope - external callbacks
      // like subscription handlers execute later, outside this function, so they can't access session.
      // sendEventInternal is a controller method, so it's accessible from any callback.
      // It provides autocomplete for all event names defined in the controller's contracts.
      this.notificationService.subscribe(userId, async (notification) => {
        await this.sendEventInternal(session.id, {
          event: 'notification',
          data: notification,
        })
      })

      // For direct sending within the handler, use the session's send method.
      // It provides stricter per-route typing (only events from this specific contract).
      await session.send('notification', { id: 'welcome', message: 'Connected!' })

      // 'keepAlive' mode: handler returns, but connection stays open for subscription events
      // Connection closes when client disconnects or server calls closeConnection()
    },
  }, {
    onConnect: (session) => console.log('Client connected:', session.id),
    onClose: (session, reason) => {
      const userId = session.context?.userId as string
      this.notificationService.unsubscribe(userId)
      console.log(`Client disconnected (${reason}):`, session.id)
    },
  })
}

Type-Safe SSE Handlers with buildHandler

For automatic type inference of request parameters (similar to buildFastifyRoute for regular controllers), use buildHandler:

import {
  AbstractSSEController,
  buildHandler,
  type SSEControllerConfig,
  type SSESession
} from 'opinionated-machine'

class ChatSSEController extends AbstractSSEController<Contracts> {
  public static contracts = {
    chatCompletion: chatCompletionContract,
  } as const

  constructor(deps: Dependencies, sseConfig?: SSEControllerConfig) {
    super(deps, sseConfig)
  }

  // Handler with automatic type inference from contract
  // sse.start(mode) returns session with fully typed send()
  private handleChatCompletion = buildHandler(chatCompletionContract, {
    sse: async (request, sse) => {
      // request.body is typed as { message: string; stream: true }
      // request.query, request.params, request.headers all typed from contract
      const words = request.body.message.split(' ')

      // Start streaming with 'autoClose' mode - closes after handler completes
      // Sends HTTP 200 + SSE headers immediately
      const session = sse.start('autoClose')

      for (const word of words) {
        // session.send() provides compile-time type checking for event names and data
        await session.send('chunk', { content: word })
      }

      // 'autoClose' mode: connection closes automatically when handler returns
    },
  })

  public buildSSERoutes() {
    return {
      chatCompletion: this.handleChatCompletion,
    }
  }
}

You can also use InferSSERequest<Contract> for manual type annotation when needed:

import { type InferSSERequest, type SSEContext, type SSESession } from 'opinionated-machine'

private handleStream = async (
  request: InferSSERequest<typeof chatCompletionContract>,
  sse: SSEContext<typeof chatCompletionContract['serverSentEventSchemas']>,
) => {
  // request.body, request.params, etc. all typed from contract
  const session = sse.start('autoClose')
  // session.send() is typed based on contract serverSentEventSchemas
  await session.send('chunk', { content: 'hello' })
  // 'autoClose' mode: connection closes when handler returns
}

SSE Controllers Without Dependencies

For controllers without dependencies, still provide the two-parameter constructor:

export class SimpleSSEController extends AbstractSSEController<Contracts> {
  constructor(deps: object, sseConfig?: SSEControllerConfig) {
    super(deps, sseConfig)
  }

  // ... implementation
}

Registering SSE Controllers

Use asSSEControllerClass in your module's resolveControllers method alongside REST controllers. SSE controllers are automatically detected via the isSSEController flag and registered in the DI container:

import { AbstractModule, type InferModuleDependencies, asControllerClass, asSSEControllerClass, asServiceClass, type DependencyInjectionOptions } from 'opinionated-machine'

export class NotificationsModule extends AbstractModule {
  resolveDependencies() {
    return {
      notificationService: asServiceClass(NotificationService),
    }
  }

  resolveControllers(diOptions: DependencyInjectionOptions) {
    return {
      // REST controller
      usersController: asControllerClass(UsersController),
      // SSE controller (automatically detected and registered for SSE routes)
      notificationsSSEController: asSSEControllerClass(NotificationsSSEController, { diOptions }),
    }
  }
}

export type NotificationsModuleDependencies = InferModuleDependencies<NotificationsModule>

Registering SSE Routes

Call registerSSERoutes after registering the @fastify/sse plugin:

const app = fastify()
app.setValidatorCompiler(validatorCompiler)
app.setSerializerCompiler(serializerCompiler)

// Register @fastify/sse plugin first
await app.register(FastifySSEPlugin)

// Then register SSE routes
context.registerSSERoutes(app)

// Optionally with global preHandler for authentication
context.registerSSERoutes(app, {
  preHandler: async (request, reply) => {
    if (!request.headers.authorization) {
      reply.code(401).send({ error: 'Unauthorized' })
    }
  },
})

await app.ready()

Broadcasting Events

Send events to multiple connections using broadcast() or broadcastIf():

// Broadcast to ALL connected clients
await this.broadcast({
  event: 'system',
  data: { message: 'Server maintenance in 5 minutes' },
})

// Broadcast to sessions matching a predicate
await this.broadcastIf(
  { event: 'channel-update', data: { channelId: '123', newMessage: msg } },
  (session) => session.context.channelId === '123',
)

Both methods return the number of clients the message was successfully sent to.

Controller-Level Hooks

Override these optional methods on your controller for global session handling:

class MySSEController extends AbstractSSEController<Contracts> {
  // Called AFTER session is registered (for all routes)
  protected onConnectionEstablished(session: SSESession): void {
    this.metrics.incrementConnections()
  }

  // Called BEFORE session is unregistered (for all routes)
  protected onConnectionClosed(session: SSESession): void {
    this.metrics.decrementConnections()
  }
}

Route-Level Options

Each route can have its own preHandler, lifecycle hooks, and logger. Pass these as the third parameter to buildHandler:

public buildSSERoutes() {
  return {
    adminStream: this.handleAdminStream,
  }
}

private handleAdminStream = buildHandler(adminStreamContract, {
  sse: async (request, sse) => {
    const session = sse.start('keepAlive')
    // ... handler logic
  },
}, {
  // Route-specific authentication
  preHandler: (request, reply) => {
    if (!request.user?.isAdmin) {
      reply.code(403).send({ error: 'Forbidden' })
    }
  },
  onConnect: (session) => console.log('Admin connected'),
  onClose: (session, reason) => console.log(`Admin disconnected (${reason})`),
  // Handle client reconnection with Last-Event-ID
  onReconnect: async (session, lastEventId) => {
    // Return events to replay, or handle manually
    return this.getEventsSince(lastEventId)
  },
  // Optional: logger for error handling (requires @lokalise/node-core)
  logger: this.logger,
})

Available route options:

| Option | Description | | -------- | ------------- | | preHandler | Authentication/authorization hook that runs before SSE session | | onConnect | Called after client connects (SSE handshake complete) | | onClose | Called when session closes (client disconnect, network failure, or server close). Receives (session, reason) where reason is 'server' or 'client' | | onReconnect | Handle Last-Event-ID reconnection, return events to replay | | logger | Optional SSELogger for error handling (compatible with pino and @lokalise/node-core). If not provided, errors in lifecycle hooks are silently ignored | | serializer | Custom serializer for SSE data (e.g., for custom JSON encoding) | | heartbeatInterval | Interval in ms for heartbeat keep-alive messages | | contractMetadataToRouteMapper | Maps contract metadata to Fastify route options (see below) |

onClose reason parameter:

• 'server': Server explicitly closed the session (via closeConnection() or autoClose mode)
• 'client': Client closed the session (EventSource.close(), navigation, network failure)

options: {
  onConnect: (session) => console.log('Client connected'),
  onClose: (session, reason) => {
    console.log(`Session closed (${reason}):`, session.id)
    // reason is 'server' or 'client'
  },
  serializer: (data) => JSON.stringify(data, null, 2), // Pretty-print JSON
  heartbeatInterval: 30000, // Send heartbeat every 30 seconds
}

contractMetadataToRouteMapper

Allows attaching cross-cutting behavior (auth, rate limiting, tracing, etc.) to a route based on metadata defined in the contract.

The return value is merged into Fastify's RouteOptions as a base. The mapper can return any of: config, bodyLimit, onRequest, preParsing, preValidation, preHandler, preSerialization, onSend, onResponse, onError, onTimeout, onRequestAbort.

// In the contract definition
const adminStreamContract = buildSseContract({
  method: 'get',
  pathResolver: () => '/api/admin/stream',
  // ...schemas...
  metadata: { requiresAuth: true, rateLimit: 100 },
})

// In the controller — driven by metadata, not duplicated per-route
private handleAdminStream = buildHandler(adminStreamContract, {
  sse: async (request, sse) => {
    const session = sse.start('keepAlive')
    // ...
  },
}, {
  contractMetadataToRouteMapper: (metadata) => ({
    config: { rateLimit: metadata.rateLimit },
    onRequest: metadata.requiresAuth ? authHook : undefined,
  }),
})

This is the same API as contractMetadataToRouteMapper in @lokalise/fastify-api-contracts, making it straightforward to share a single mapper function across REST, SSE, and dual-mode routes.

SSE Session Methods

The session object returned by sse.start(mode) provides several useful methods:

private handleStream = buildHandler(streamContract, {
  sse: async (request, sse) => {
    const session = sse.start('autoClose')

    // Check if session is still active
    if (session.isConnected()) {
      await session.send('status', { connected: true })
    }

    // Get raw writable stream for advanced use cases (e.g., pipeline)
    const stream = session.getStream()

    // Stream messages from an async iterable with automatic validation
    async function* generateMessages() {
      yield { event: 'message' as const, data: { text: 'Hello' } }
      yield { event: 'message' as const, data: { text: 'World' } }
    }
    await session.sendStream(generateMessages())

    // 'autoClose' mode: connection closes when handler returns
  },
})

| Method | Description | | -------- | ------------- | | send(event, data, options?) | Send a typed event (validates against contract schema) | | isConnected() | Check if the session is still active | | getStream() | Get the underlying WritableStream for advanced use cases | | sendStream(messages) | Stream messages from an AsyncIterable with validation |

Graceful Shutdown

SSE controllers automatically close all connections during application shutdown. This is configured by asSSEControllerClass which sets closeAllConnections as the async dispose method with priority 5 (early in shutdown sequence).

Error Handling

When sendEvent() fails (e.g., client disconnected), it:

• Returns false to indicate failure
• Automatically removes the dead connection from tracking
• Prevents further send attempts to that connection

const sent = await this.sendEvent(connectionId, { event: 'update', data })
if (!sent) {
  // Connection was closed or failed - already removed from tracking
  this.cleanup(connectionId)
}

Lifecycle hook errors (onConnect, onReconnect, onClose):

• All lifecycle hooks are wrapped in try/catch to prevent crashes
• If a logger is provided in route options, errors are logged with context
• If no logger is provided, errors are silently ignored
• The session lifecycle continues even if a hook throws

// Provide a logger to capture lifecycle errors
public buildSSERoutes() {
  return {
    stream: this.handleStream,
  }
}

private handleStream = buildHandler(streamContract, {
  sse: async (request, sse) => {
    const session = sse.start('autoClose')
    // ... handler logic
  },
}, {
  logger: this.logger, // pino-compatible logger
  onConnect: (session) => { /* may throw */ },
  onClose: (session, reason) => { /* may throw */ },
})

Long-lived Connections vs Request-Response Streaming

SSE session lifetime is determined by the mode passed to sse.start(mode):

// sse.start('autoClose') - close connection when handler returns (request-response pattern)
// sse.start('keepAlive') - keep connection open for external events (subscription pattern)
// sse.respond(code, body) - send HTTP response before streaming (early return)

Long-lived sessions (notifications, live updates):

• Handler starts streaming with sse.start('keepAlive')
• Session stays open indefinitely after handler returns
• Events are sent later via callbacks using sendEventInternal()
• Client closes session when done (e.g., eventSource.close() or navigating away)
• Server cleans up via onConnectionClosed() hook

private handleStream = buildHandler(streamContract, {
  sse: async (request, sse) => {
    // Start streaming with 'keepAlive' mode - stays open for external events
    const session = sse.start('keepAlive')

    // Set up subscription - events sent via callback AFTER handler returns
    this.service.subscribe(session.id, (data) => {
      this.sendEventInternal(session.id, { event: 'update', data })
    })
    // 'keepAlive' mode: handler returns, but connection stays open
  },
})

// Clean up when client disconnects
protected onConnectionClosed(session: SSESession): void {
  this.service.unsubscribe(session.id)
}

Request-response streaming (AI completions):

• Handler starts streaming with sse.start('autoClose')
• Use session.send() for type-safe event sending within the handler
• Session automatically closes when handler returns

private handleChatCompletion = buildHandler(chatCompletionContract, {
  sse: async (request, sse) => {
    // Start streaming with 'autoClose' mode - closes when handler returns
    const session = sse.start('autoClose')

    const words = request.body.message.split(' ')
    for (const word of words) {
      await session.send('chunk', { content: word })
    }
    await session.send('done', { totalTokens: words.length })

    // 'autoClose' mode: connection closes automatically when handler returns
  },
})

Error handling before streaming:

Use sse.respond(code, body) to return an HTTP response before streaming starts. This is useful for any early return: validation errors, not found, redirects, etc.

private handleStream = buildHandler(streamContract, {
  sse: async (request, sse) => {
    // Early return BEFORE starting stream - can return any HTTP response
    const entity = await this.service.find(request.params.id)
    if (!entity) {
      return sse.respond(404, { error: 'Entity not found' })
    }

    // Validation passed - start streaming with autoClose mode
    const session = sse.start('autoClose')
    await session.send('data', entity)
    // Connection closes automatically when handler returns
  },
})

### SSE Parsing Utilities

The library provides production-ready utilities for parsing SSE (Server-Sent Events) streams:

| Function | Use Case |
|----------|----------|
| `parseSSEEvents` | **Testing & complete responses** - when you have the full response body |
| `parseSSEBuffer` | **Production streaming** - when data arrives incrementally in chunks |

#### parseSSEEvents

Parse a complete SSE response body into an array of events.

**When to use:** Testing with Fastify's `inject()`, or when the full response is available (e.g., request-response style SSE like OpenAI completions):

```ts
import { parseSSEEvents, type ParsedSSEEvent } from 'opinionated-machine'

const responseBody = `event: notification
data: {"id":"1","message":"Hello"}

event: notification
data: {"id":"2","message":"World"}

`

const events: ParsedSSEEvent[] = parseSSEEvents(responseBody)
// Result:
// [
//   { event: 'notification', data: '{"id":"1","message":"Hello"}' },
//   { event: 'notification', data: '{"id":"2","message":"World"}' }
// ]

// Access parsed data
const notifications = events.map(e => JSON.parse(e.data))

parseSSEBuffer

Parse a streaming SSE buffer, handling incomplete events at chunk boundaries.

When to use: Production clients consuming real-time SSE streams (notifications, live feeds, chat) where events arrive incrementally:

import { parseSSEBuffer, type ParseSSEBufferResult } from 'opinionated-machine'

let buffer = ''

// As chunks arrive from a stream...
for await (const chunk of stream) {
  buffer += chunk
  const result: ParseSSEBufferResult = parseSSEBuffer(buffer)

  // Process complete events
  for (const event of result.events) {
    console.log('Received:', event.event, event.data)
  }

  // Keep incomplete data for next chunk
  buffer = result.remaining
}

Production example with fetch:

const response = await fetch(url)
const reader = response.body!.getReader()
const decoder = new TextDecoder()
let buffer = ''

while (true) {
  const { done, value } = await reader.read()
  if (done) break

  buffer += decoder.decode(value, { stream: true })
  const { events, remaining } = parseSSEBuffer(buffer)
  buffer = remaining

  for (const event of events) {
    console.log('Received:', event.event, JSON.parse(event.data))
  }
}

ParsedSSEEvent Type

Both functions return events with this structure:

type ParsedSSEEvent = {
  id?: string      // Event ID (from "id:" field)
  event?: string   // Event type (from "event:" field)
  data: string     // Event data (from "data:" field, always present)
  retry?: number   // Reconnection interval (from "retry:" field)
}

Testing SSE Controllers

The test client depends on the session mode:

| Session Mode | Test Client | Why | |-------------|-------------|--------| | autoClose | SSEInjectClient or injectSSE/injectPayloadSSE | Handler completes and closes connection; all events available at once | | keepAlive | SSEHttpClient | Connection stays open; events arrive incrementally via server push |

Enable the connection spy by passing isTestMode: true in diOptions (required for awaitServerConnection).

Testing keepAlive SSE (long-lived connections)

Use SSEHttpClient against your running app. The key pattern:

1. Connect with awaitServerConnection to eliminate the race condition
2. Call collectEvents() before pushing events (they arrive asynchronously)
3. Push events from the server via sendEventInternal() or broadcastToRoom()
4. Await the collected events
5. Always call client.close() to release the connection

import { SSEHttpClient, SSETestServer } from 'opinionated-machine'

describe('NotificationsSSEController', () => {
  let app: AppInstance
  let server: SSETestServer
  let controller: NotificationsSSEController

  beforeAll(async () => {
    app = await getApp({ /* your test config */ })
    controller = app.diContainer.resolve('notificationsSSEController')

    // SSETestServer.start() starts your app on a random port and provides baseUrl
    server = await SSETestServer.start(app)
  })

  afterAll(async () => {
    await server.close()
  })

  it('receives notifications over keepAlive SSE', async () => {
    // 1. Connect with awaitServerConnection to eliminate race condition
    const { client, serverConnection } = await SSEHttpClient.connect(
      server.baseUrl,
      '/api/notifications/stream',
      {
        query: { userId: 'test-user' },
        awaitServerConnection: { controller },
      },
    )

    expect(client.response.ok).toBe(true)

    // 2. Start collecting events BEFORE pushing (they arrive asynchronously)
    const eventsPromise = client.collectEvents(2)

    // 3. Push events from server
    await controller.sendEventInternal(serverConnection.id, {
      event: 'notification',
      data: { id: '1', message: 'Hello!' },
    })
    await controller.sendEventInternal(serverConnection.id, {
      event: 'notification',
      data: { id: '2', message: 'World!' },
    })

    // 4. Await collected events
    const events = await eventsPromise

    expect(events).toHaveLength(2)
    expect(JSON.parse(events[0].data)).toEqual({ id: '1', message: 'Hello!' })
    expect(JSON.parse(events[1].data)).toEqual({ id: '2', message: 'World!' })

    // 5. Clean up
    client.close()
  })
})

Testing autoClose SSE (request-response streaming)

Use SSEInjectClient or the contract-aware injectSSE/injectPayloadSSE helpers. No real HTTP server needed - all events are available immediately after the handler completes:

import { SSEInjectClient } from 'opinionated-machine'

it('streams chat completions', async () => {
  const client = new SSEInjectClient(app) // works without app.listen()

  const conn = await client.connectWithBody(
    '/api/chat/completions',
    { message: 'Hello world' },
  )

  expect(conn.getStatusCode()).toBe(200)
  const events = conn.getReceivedEvents()
  const chunks = events.filter((e) => e.event === 'chunk')
  expect(chunks.length).toBeGreaterThan(0)
})

SSESessionSpy API

The connectionSpy is available when isTestMode: true is passed to asSSEControllerClass:

// Wait for a session to be established (with timeout)
const session = await controller.connectionSpy.waitForConnection({ timeout: 5000 })

// Wait for a session matching a predicate (useful for multiple sessions)
const session = await controller.connectionSpy.waitForConnection({
  timeout: 5000,
  predicate: (s) => s.request.url.includes('/api/notifications'),
})

// Check if a specific session is active
const isConnected = controller.connectionSpy.isConnected(sessionId)

// Wait for a specific session to disconnect
await controller.connectionSpy.waitForDisconnection(sessionId, { timeout: 5000 })

// Get all session events (connect/disconnect history)
const events = controller.connectionSpy.getEvents()

// Clear event history and claimed sessions between tests
controller.connectionSpy.clear()

Note: waitForConnection tracks "claimed" sessions internally. Each call returns a unique unclaimed session, allowing sequential waits for the same URL path without returning the same session twice. This is used internally by SSEHttpClient.connect() with awaitServerConnection.

Session Monitoring

Controllers have access to utility methods for monitoring sessions:

// Get count of active sessions
const count = this.getConnectionCount()

// Get all active sessions (for iteration/inspection)
const sessions = this.getConnections()

// Check if session spy is enabled (useful for conditional logic)
if (this.hasConnectionSpy()) {
  // ...
}

SSE Rooms

SSE Rooms provide Socket.IO-style room functionality for grouping connections and broadcasting messages to specific groups. Common use cases include:

• Multi-tenant systems - Broadcast announcements to all users within an organization or team
• Live dashboards - Multiple users viewing the same dashboard join a room to receive real-time metric updates
• Stock tickers - Users subscribe to specific symbols; each symbol is a room receiving price updates
• Sports/game scores - Users following specific matches join those rooms for live score updates

Enabling Rooms

Room infrastructure is registered at the module level via resolveDependencies(). Controllers opt in with rooms: true, which resolves sseRoomBroadcaster from the DI cradle:

import { asValue } from 'awilix'
import {
  AbstractModule,
  AbstractSSEController,
  asSingletonClass,
  asSSEControllerClass,
  SSERoomBroadcaster,
  SSERoomManager,
} from 'opinionated-machine'

class DashboardModule extends AbstractModule {
  resolveDependencies() {
    return {
      // Required: room infrastructure — registered once, shared across controllers
      sseRoomManager: asValue(new SSERoomManager()),
      sseRoomBroadcaster: asSingletonClass(SSERoomBroadcaster), // expects 'sseRoomManager' in cradle — name must match exactly
    }
  }

  resolveControllers(diOptions: DependencyInjectionOptions) {
    return {
      // rooms: true → resolves 'sseRoomBroadcaster' from DI cradle
      dashboardController: asSSEControllerClass(DashboardSSEController, {
        diOptions,
        rooms: true,
      }),
    }
  }
}

Required DI registrations for rooms: Any module using rooms: true must have both sseRoomManager and sseRoomBroadcaster registered in the DI container before the controller is resolved. SSERoomBroadcaster expects sseRoomManager in its constructor cradle.

Session Room Operations

When rooms are enabled, each SSE session has access to room operations via session.rooms:

private handleDashboardStream = buildHandler(dashboardStreamContract, {
  sse: async (request, sse) => {
    const session = sse.start('keepAlive')

    // Join one or more rooms
    session.rooms.join(`dashboard:${request.params.dashboardId}`)
    session.rooms.join(['org:acme', 'plan:enterprise']) // Multiple rooms

    // Leave rooms
    session.rooms.leave('plan:enterprise')
  },
})

Broadcasting to Rooms

Use broadcastToRoom() from your controller to send type-safe messages to all connections in a room. Event names and data are validated against your contract schemas at compile time:

class DashboardSSEController extends AbstractSSEController<typeof contracts> {
  // Send metrics update to everyone viewing the dashboard
  // Event name and data are type-checked against contract's sseEvents
  async broadcastMetricsUpdate(dashboardId: string, metrics: DashboardMetrics) {
    const count = await this.broadcastToRoom(
      `dashboard:${dashboardId}`,
      'metricsUpdate', // Must be a valid event name from contracts
      metrics,         // Must match the schema for 'metricsUpdate'
    )
    console.log(`Metrics sent to ${count} viewers`)
  }

  // Broadcast to a room
  async broadcastChange(dashboardId: string, change: DashboardChange) {
    await this.broadcastToRoom(
      `dashboard:${dashboardId}`,
      'change',
      change,
    )
  }

  // Broadcast to multiple rooms (connections in any room receive it, de-duplicated)
  async announceFeature(feature: string) {
    const count = await this.broadcastToRoom(
      ['premium', 'beta-testers'],
      'featureFlag',
      { flag: feature, enabled: true },
    )
    // Each connection receives the message only once, even if in multiple rooms
  }

  // Local-only broadcast (skip Redis propagation in multi-node setups)
  async localAnnouncement(room: string, message: string) {
    await this.broadcastToRoom(room, 'announcement', { message }, { local: true })
  }
}

Room Broadcaster (Decoupled Broadcasting)

The broadcastToRoom() method on the controller is protected, which means domain services (use cases, event handlers, message queue consumers) can't call it directly. The SSERoomBroadcaster solves this — it's a shared, non-generic service registered in DI that domain services receive directly:

import { defineEvent, type SSERoomBroadcaster } from 'opinionated-machine'
import { z } from 'zod'

// 1. Define type-safe events with schemas
const metricsUpdateEvent = defineEvent(
  'metricsUpdate',
  z.object({ cpu: z.number(), memory: z.number() }),
)

// 2. Register domain service in resolveDependencies()
class DashboardModule extends AbstractModule {
  resolveDependencies() {
    return {
      sseRoomManager: asValue(new SSERoomManager()),
      sseRoomBroadcaster: asSingletonClass(SSERoomBroadcaster), // expects 'sseRoomManager' in cradle — name must match exactly
      metricsService: asSingletonClass(MetricsService),
    }
  }

  resolveControllers(diOptions: DependencyInjectionOptions) {
    return {
      dashboardController: asSSEControllerClass(DashboardSSEController, {
        diOptions,
        rooms: true,
      }),
    }
  }
}

// 3. Inject broadcaster into domain services — no generic needed
class MetricsService {
  private broadcaster: SSERoomBroadcaster

  constructor(deps: { sseRoomBroadcaster: SSERoomBroadcaster }) {
    this.broadcaster = deps.sseRoomBroadcaster
  }

  async onMetricsUpdate(dashboardId: string, metrics: { cpu: number; memory: number }) {
    // Type-safe: data is validated against the event's schema at compile time
    await this.broadcaster.broadcastToRoom(
      `dashboard:${dashboardId}`,
      metricsUpdateEvent,
      metrics,
    )
  }
}

The broadcaster provides broadcastToRoom() (with defineEvent()-based type safety), broadcastMessage() (raw SSEMessage), plus room query methods (getConnectionsInRoom, getConnectionCountInRoom). Multiple controllers register their sendEvent with the same broadcaster — the first to recognize a connection handles delivery.

Room Name Helpers

Room names are plain strings (like Socket.IO), but defineRoom() adds type-safe resolvers that ensure consistent naming across controllers and domain services:

import { defineRoom } from 'opinionated-machine'

// Define typed room name resolvers
const dashboardRoom = defineRoom<{ dashboardId: string }>(
  ({ dashboardId }) => `dashboard:${dashboardId}`,
)

const projectChannelRoom = defineRoom<{ projectId: string; channelId: string }>(
  ({ projectId, channelId }) => `project:${projectId}:channel:${channelId}`,
)

// In controller handler — params are type-checked
session.rooms.join(dashboardRoom({ dashboardId: request.params.dashboardId }))

// In domain service — same resolver, same type safety
await broadcaster.broadcastToRoom(
  dashboardRoom({ dashboardId }),
  'metricsUpdate',
  metrics,
)

defineRoom() is a zero-overhead identity wrapper — it simply returns the function you pass in, typed as RoomNameResolver<TParams>. The value is purely at compile time: typos in room name patterns become type errors, and refactoring a room's naming scheme only requires changing one place.

Room Query Methods

Controllers have access to room query methods:

class DashboardSSEController extends AbstractSSEController<typeof contracts> {
  // Get all connection IDs in a room
  getDashboardViewers(dashboardId: string): string[] {
    return this.getConnectionsInRoom(`dashboard:${dashboardId}`)
  }

  // Get count of connections in a room
  getDashboardViewerCount(dashboardId: string): number {
    return this.getConnectionCountInRoom(`dashboard:${dashboardId}`)
  }

  // Get all rooms a specific connection is in
  getConnectionRooms(connectionId: string): string[] {
    return this.getRooms(connectionId)
  }

  // Manually join/leave rooms from controller (useful for admin operations)
  moveToRoom(connectionId: string, fromRoom: string, toRoom: string) {
    this.leaveRoom(connectionId, fromRoom)
    this.joinRoom(connectionId, toRoom)
  }
}

Auto-Leave on Disconnect

When a connection closes (client disconnect or server close), it automatically leaves all rooms. No manual cleanup is required.

Multi-Node Deployments with Redis

For multi-node deployments where connections are distributed across servers, pass a Redis adapter to SSERoomManager when registering room infrastructure:

import { RedisAdapter } from '@opinionated-machine/sse-rooms-redis'

class InfraModule extends AbstractModule {
  resolveDependencies() {
    return {
      sseRoomManager: asSingletonFunction(({ redis }: { redis: Redis }): SSERoomManager =>
        new SSERoomManager({
          adapter: new RedisAdapter({
            pubClient: redis,
            subClient: redis.duplicate(),
            channelPrefix: 'myapp:sse:room:', // Optional, default: 'sse:room:'
          }),
        }),
      ),
      sseRoomBroadcaster: asSingletonClass(SSERoomBroadcaster), // expects 'sseRoomManager' in cradle — name must match exactly
    }
  }
}

class DashboardModule extends AbstractModule {
  resolveControllers(diOptions: DependencyInjectionOptions) {
    return {
      dashboardController: asSSEControllerClass(DashboardSSEController, {
        diOptions,
        rooms: true,
      }),
    }
  }
}

The Redis adapter uses Pub/Sub for cross-node message propagation. When you call broadcastToRoom(), the message is published to R