npm package: @mment/cqrs-es
Description: A serverless-first CQRS+ES framework for Next.js 15 applications with git-like branching capabilities.
Latest version: 0.1.2

# @mment/cqrs-es

A serverless-first CQRS+ES framework for Next.js 15 applications with git-like branching capabilities.

## Features

- **Event Sourcing with Git-Like Branching**: Fork, work on branches, and squash merge with three-way conflict detection
- **Multi-Tenant Isolation**: Database-per-tenant architecture with connection pooling
- **Zero Redundancy (DRY)**: Single source of truth using Zod schemas with auto-generated TypeScript types
- **Serverless-First**: Optimized for Vercel and other serverless platforms
- **Type-Safe**: Strict TypeScript with no `any` types allowed
- **Role-Based Access Control**: Platform and tenant-level permissions

## Installation

```bash
npm install @mment/cqrs-es
```

### Peer Dependencies

```bash
npm install next@^15.5.0 zod@^3.x postgres@^3.x
```

## Quick Start

### Environment Setup

Create a `.env.local` file:

```bash
DATABASE_URL=postgresql://user:password@localhost:5432/platform_db
TENANT_DATABASE_URL_TEMPLATE=postgresql://user:password@localhost:5432/tenant_{tenantId}
EVENT_STORE_MAX_CONNECTIONS=10
DEFAULT_BRANCH=main
```

## Complete Example: Todo Application

Follow these steps to build a working Todo application with CQRS+ES.

### Step 1: Define Your Commands

Create `app/domain/todo/commands.ts`:

```typescript
// app/domain/todo/commands.ts
import { z } from "zod";
import { command } from "@mment/cqrs-es/schemas/command-builder";
import { CoreRole, TenantRole } from "@mment/cqrs-es/schemas";

// Commands with roles and validation
export const createTodoCommand = command("CreateTodo", "Todo")
  .withData({
    title: z.string().min(1).max(200),
    description: z.string().max(1000).optional(),
  })
  .requireRoles(CoreRole.User, TenantRole.Member);

export const completeTodoCommand = command("CompleteTodo", "Todo").withData({}).requireRoles(CoreRole.User, TenantRole.Member);

export const uncompleteTodoCommand = command("UncompleteTodo", "Todo").withData({}).requireRoles(CoreRole.User, TenantRole.Member);

// Type exports
export type CreateTodoCommand = z.infer<typeof createTodoCommand.dataSchema>;
export type CompleteTodoCommand = z.infer<typeof completeTodoCommand.dataSchema>;
export type UncompleteTodoCommand = z.infer<typeof uncompleteTodoCommand.dataSchema>;
```

### Step 2: Define Your Aggregate

Create `app/domain/todo/aggregate.ts`:

```typescript
// app/domain/todo/aggregate.ts
import { defineAggregate } from "@mment/cqrs-es/aggregates/aggregate-definition";
import type { DomainEvent } from "@mment/cqrs-es/schemas";

export type TodoState = {
  id: string;
  title: string;
  description?: string;
  completed: boolean;
  createdBy: string;
  completedBy?: string;
  completedAt?: number;
  createdAt: number;
};

export const TodoAggregate = defineAggregate<TodoState>(
  "Todo",
  {
    id: "",
    title: "",
    description: undefined,
    completed: false,
    createdBy: "",
    completedBy: undefined,
    completedAt: undefined,
    createdAt: Date.now(),
  },
  {
    TodoCreated: (state, event) => ({
      ...state,
      id: event.aggregateId,
      title: event.data.title as string,
      description: event.data.description as string | undefined,
      createdBy: event.metadata.userId || "",
      createdAt: event.timestamp,
    }),

    TodoCompleted: (state, event) => ({
      ...state,
      completed: true,
      completedBy: event.metadata.userId,
      completedAt: event.timestamp,
    }),

    TodoUncompleted: (state, event) => ({
      ...state,
      completed: false,
      completedBy: undefined,
      completedAt: undefined,
    }),
  },
);
```

### Step 3: Create Command Handlers

Create `app/domain/todo/handlers.ts`:

```typescript
// app/domain/todo/handlers.ts
import type { CommandHandler, CommandContext } from "@mment/cqrs-es/commands/command-processor";
import type { Command, DomainEvent } from "@mment/cqrs-es/schemas";
import { createTodoCommand, completeTodoCommand, uncompleteTodoCommand } from "./commands";
import { TodoAggregate } from "./aggregate";

export const createTodoHandler: CommandHandler = {
  metadata: createTodoCommand,
  execute: async (command: Command, context: CommandContext) => {
    const aggregate = await context.aggregateLoader.load(command.aggregate.id, TodoAggregate);

    // Business rule: Todo must not exist
    if (aggregate.id !== "") {
      return {
        success: false,
        aggregateId: command.aggregate.id,
        error: "Todo already exists",
      };
    }

    const event: DomainEvent = {
      aggregateId: command.aggregate.id,
      aggregateType: "Todo",
      type: "TodoCreated",
      data: command.data,
      timestamp: Date.now(),
      metadata: {
        commandId: command.command.id,
        userId: command.context.userId,
        tenantId: command.context.tenantId,
      },
    };

    return {
      success: true,
      aggregateId: command.aggregate.id,
      events: [event],
    };
  },
};

export const completeTodoHandler: CommandHandler = {
  metadata: completeTodoCommand,
  execute: async (command: Command, context: CommandContext) => {
    const aggregate = await context.aggregateLoader.load(command.aggregate.id, TodoAggregate);

    // Business rules
    if (aggregate.id === "") {
      return {
        success: false,
        aggregateId: command.aggregate.id,
        error: "Todo not found",
      };
    }

    if (aggregate.completed) {
      return {
        success: false,
        aggregateId: command.aggregate.id,
        error: "Todo is already completed",
      };
    }

    const event: DomainEvent = {
      aggregateId: command.aggregate.id,
      aggregateType: "Todo",
      type: "TodoCompleted",
      data: {},
      timestamp: Date.now(),
      metadata: {
        commandId: command.command.id,
        userId: command.context.userId,
        tenantId: command.context.tenantId,
      },
    };

    return {
      success: true,
      aggregateId: command.aggregate.id,
      events: [event],
    };
  },
};

export const uncompleteTodoHandler: CommandHandler = {
  metadata: uncompleteTodoCommand,
  execute: async (command: Command, context: CommandContext) => {
    const aggregate = await context.aggregateLoader.load(command.aggregate.id, TodoAggregate);

    // Business rules
    if (aggregate.id === "") {
      return {
        success: false,
        aggregateId: command.aggregate.id,
        error: "Todo not found",
      };
    }

    if (!aggregate.completed) {
      return {
        success: false,
        aggregateId: command.aggregate.id,
        error: "Todo is not completed",
      };
    }

    const event: DomainEvent = {
      aggregateId: command.aggregate.id,
      aggregateType: "Todo",
      type: "TodoUncompleted",
      data: {},
      timestamp: Date.now(),
      metadata: {
        commandId: command.command.id,
        userId: command.context.userId,
        tenantId: command.context.tenantId,
      },
    };

    return {
      success: true,
      aggregateId: command.aggregate.id,
      events: [event],
    };
  },
};

export const allTodoHandlers = [createTodoHandler, completeTodoHandler, uncompleteTodoHandler];
```

### Step 4: Define Your Read Model

Create `app/domain/todo/read-model.ts`:

```typescript
// app/domain/todo/read-model.ts
import { z } from "zod";
import { readModel } from "@mment/cqrs-es/schemas/read-model-builder";

export const TodoReadModel = readModel("Todo", "todos")
  .withSchema({
    id: z.string(),
    title: z.string(),
    description: z.string().optional(),
    completed: z.boolean(),
    createdBy: z.string(),
    createdAt: z.date(),
    completedBy: z.string().optional(),
    completedAt: z.date().optional(),
  })
  .withIndexes([
    { name: "idx_todos_completed", columns: ["completed"] },
    { name: "idx_todos_created_at", columns: ["createdAt"] },
  ]);

// Type export
export type Todo = z.infer<typeof TodoReadModel.schema>;
```

### Step 5: Set Up Projections (Read Model Updates)

Create `app/domain/todo/projections.ts`:

```typescript
// app/domain/todo/projections.ts
import { ProjectionManager } from "@mment/cqrs-es/projections/projection-manager";
import type { EventStore } from "@mment/cqrs-es/event-store/event-store";
import type { Kysely } from "kysely";
import type { DB } from "@mment/cqrs-es/database/types";

export function setupTodoProjections(eventStore: EventStore, db: Kysely<DB>): ProjectionManager {
  const projectionManager = new ProjectionManager(eventStore, db);

  // Simple inline handlers using .on()
  projectionManager.on("TodoCreated", async (event, context) => {
    await context.db
      .insertInto("todos")
      .values({
        id: event.aggregateId,
        title: event.data.title as string,
        description: event.data.description as string | undefined,
        completed: false,
        createdBy: event.metadata.userId || "",
        createdAt: new Date(event.timestamp),
      })
      .execute();
  });

  projectionManager.on("TodoCompleted", async (event, context) => {
    await context.db
      .updateTable("todos")
      .set({
        completed: true,
        completedBy: event.metadata.userId,
        completedAt: new Date(event.timestamp),
      })
      .where("id", "=", event.aggregateId)
      .execute();
  });

  projectionManager.on("TodoUncompleted", async (event, context) => {
    await context.db
      .updateTable("todos")
      .set({
        completed: false,
        completedBy: undefined,
        completedAt: undefined,
      })
      .where("id", "=", event.aggregateId)
      .execute();
  });

  return projectionManager;
}
```

### Step 6: Register Table Creation Hook

Create `app/domain/todo/setup.ts` to automatically create tables when tenants are created:

```typescript
// app/domain/todo/setup.ts
import { registerTenantInitializationHook } from "@mment/cqrs-es/tenant";
import { generateTableSchema } from "@mment/cqrs-es/schemas/read-model-builder";
import { sql } from "kysely";
import { TodoReadModel } from "./read-model";

// This hook runs automatically when a new tenant database is created
registerTenantInitializationHook("todo-tables", async (db, tenantId) => {
  // Generate SQL from the read model defined in Step 4
  const tableSQL = generateTableSchema(TodoReadModel);

  // Execute the SQL to create the table
  await sql.raw(tableSQL).execute(db);

  console.log(`Todo tables created for tenant ${tenantId}`);
});
```

Then import this setup file in your app initialization (e.g., `app/layout.tsx` or `instrumentation.ts`):

```typescript
// app/instrumentation.ts (or app/layout.tsx)
import "@/domain/todo/setup"; // Registers the hook on import

export function register() {
  // Framework will now auto-create todo tables for new tenants
}
```

### Step 7: Create Command Processor and Register Handlers

Create `app/platform/command-processor.ts` to set up the command processor:

```typescript
// app/platform/command-processor.ts
import { CommandProcessor } from "@mment/cqrs-es/commands/command-processor";
import { createPlatformEventStore } from "@mment/cqrs-es/factories";
import { allTodoHandlers } from "@/domain/todo/handlers";

// Create event store
const eventStore = createPlatformEventStore();

// Create command processor
export const processor = new CommandProcessor(eventStore);

// Register all todo handlers
allTodoHandlers.forEach((handler) => processor.register(handler));
```

### Step 8: Create Your API Route

Create `app/api/todos/route.ts`:

```typescript
// app/api/todos/route.ts
import { createCommandRoute } from "@mment/cqrs-es/commands/route-handler";
import { createTenantContext } from "@mment/cqrs-es/factories";
import { processor } from "@/platform/command-processor";
import { setupTodoProjections } from "@/domain/todo/projections";

// Command handler (POST) with SSE streaming
export const { POST } = createCommandRoute({
  processor,
  getAuth: async (req) => {
    // TODO: Get from your auth system
    return {
      userId: "user-123",
      platformRole: "user",
      tenantRole: "member",
    };
  },
  getTenantId: async (req) => {
    // TODO: Extract from subdomain or header
    return "tenant-1";
  },
});

// Query handler (GET)
export async function GET(request: Request): Promise<Response> {
  const url = new URL(request.url);
  const completed = url.searchParams.get("completed");

  // Get tenant database connection
  const tenantContext = await createTenantContext("tenant-1"); // TODO: Get from request
  const { db } = tenantContext;

  // Set up projections if needed
  const projectionManager = setupTodoProjections(tenantContext.eventStore, db);

  const query = db.selectFrom("todos").selectAll();

  if (completed !== null) {
    query.where("completed", "=", completed === "true");
  }

  const todos = await query.orderBy("createdAt", "desc").execute();

  return Response.json(todos);
}
```

### Step 9: Start Using Your API

**Create a todo:**

```bash
curl -X POST http://localhost:3000/api/todos \
  -H "Content-Type: application/json" \
  -d '{
    "type": "CreateTodo",
    "aggregateId": "todo-1",
    "data": {
      "title": "Learn CQRS+ES",
      "description": "Build a todo app with event sourcing"
    }
  }'
```

**Complete a todo:**

```bash
curl -X POST http://localhost:3000/api/todos \
  -H "Content-Type: application/json" \
  -d '{
    "type": "CompleteTodo",
    "aggregateId": "todo-1",
    "data": {}
  }'
```

**Query todos:**

```bash
# Get all todos
curl http://localhost:3000/api/todos

# Get only completed todos
curl http://localhost:3000/api/todos?completed=true
```

### What Just Happened?

1. ✅ **Commands defined** (Step 1) - Zod validation and role requirements
2. ✅ **Aggregate defined** (Step 2) - Business logic and event handlers
3. ✅ **Handlers created** (Step 3) - Enforce rules and generate events
4. ✅ **Read model defined** (Step 4) - Single source of truth with indexes
5. ✅ **Projections set up** (Step 5) - Automatically update read model
6. ✅ **Table creation automated** (Step 6) - Hook generates tables from schema
7. ✅ **Processor registered** (Step 7) - All handlers wired to event store
8. ✅ **Routes created** (Step 8) - Commands via POST, queries via GET
9. ✅ **API tested** (Step 9) - Working Todo app with SSE streaming

### Next Steps

- Add more commands (DeleteTodo, UpdateTodo)
- Add branching for draft editing
- Implement authentication and tenant context
- Add more projections for different views

## Architecture

### Directory Structure

```
src/
├── aggregates/    # Aggregate loader
├── branches/      # Branch operations
├── commands/      # Command handling
├── database/      # Database
├── event-store/   # Event store implementation
├── projections/   # Projection system
├── queries/       # Query utilities
└── schemas/       # Schema builders
├── tenant/        # Tenant-specific utilities
└── utils/         # TypeScript types
```

### Core Concepts

#### Event Sourcing

Events are append-only in the event store. All state changes are captured as events.

#### Git-Like Branching

- Create branches from any point in time
- Work independently on branches
- Squash merge with automatic conflict detection

#### Multi-Tenant Isolation

- Each tenant has its own database
- Connection pooling per tenant
- Tenant context resolved from request

## Development

### Build

```bash
npm run build
```

### Testing

```bash
npm test                 # Run all tests
npm run test:watch       # Watch mode
npm run test:coverage    # Generate coverage report
```

### Linting & Formatting

```bash
npm run lint            # Check ESLint rules
npm run lint:fix        # Auto-fix ESLint issues
npm run format          # Check Prettier formatting
npm run format:fix      # Auto-format with Prettier
```

## Code Standards

### TypeScript

- Strict mode enabled
- No `any` types allowed - use `unknown` with type guards
- Explicit return types required (except React components)

### Naming Conventions

- **Files/folders**: kebab-case (`event-store.ts`)
- **Types**: PascalCase (`DomainEvent`)
- **Functions**: camelCase (`processCommand`)
- **Constants**: UPPER_SNAKE_CASE (`DEFAULT_BRANCH`)

### Testing

- Target: 90%+ code coverage
- Vitest with global test functions
- Integration tests for full command → event → projection flow

## Role-Based Access Control

### Platform Roles

- `user` - Standard platform user
- `admin` - Platform administrator

### Tenant Roles

- `owner` - Tenant owner (full access)
- `admin` - Tenant admin
- `member` - Tenant member

## Contributing

We use a greenfield approach for database schema:

1. Modify `src/event-store/schema.sql`
2. Run database reset (no migrations in development)
3. Always start fresh