Slice State View

Name: Slice State View
Author: dilgerma

dilgerma/embuilder-node

1 installs
4 repo stars
Updated April 6, 2026
dilgerma/embuilder-node

Builds a state-view slice from an event model as a PostgreSQL read-model projection, with projection code, tests, a query API route and a migration.

About

Creates event-sourced state-view slices that project events into queryable PostgreSQL tables using Emmett and Knex, generating the projection, Testcontainers tests, an Express query endpoint and a Flyway migration. A developer uses it to add a read model to an event-driven Node backend.

Builds event-model state-view slices as PostgreSQL read-model projections
Generates projection, tests, query route and a Flyway migration per slice

Slice State View by the numbers

1 all-time installs (skills.sh)
Ranked #3,836 of 4,347 Backend & APIs skills by installs in the Skillselion catalog
Data as of Jul 11, 2026 (Skillselion catalog sync)

npx skills add https://github.com/dilgerma/embuilder-node --skill slice-state-view

Add your badge

Show developers this skill is listed on Skillselion. Paste this into your README.

[![Listed on Skillselion](https://skillselion.com/badge/skills/dilgerma/embuilder-node/slice-state-view.svg)](https://skillselion.com/skills/dilgerma/embuilder-node/slice-state-view)

Installs	1
repo stars	★ 4
Last updated	April 6, 2026
Repository	dilgerma/embuilder-node ↗

What it does

Builds a state-view slice from an event model as a PostgreSQL read-model projection, with projection code, tests, a query API route and a migration.

Files

SKILL.mdMarkdownGitHub ↗

Overview

State View Slices are read model projections that build table-based views from events. They consume events and project them into queryable database tables using PostgreSQL.

If the processors-array in the slice json is not empty. Treat this as an AUTOMATION Slice. Load the skill for automation slice.

Critical Requirements

Implementation Steps

When creating a state-view slice, you MUST create the following files:

1. src/slices/{SliceName}/{SliceName}Projection.ts - Projection handler 2. src/slices/{SliceName}/{SliceName}.test.ts - Projection tests 3. src/slices/{SliceName}/routes.ts - Query API endpoint 4. supabase/migrations/V{N}__{tablename}.sql - Database migration file 5. **src/slices/{SliceName}/ui-prompt.md - prompt to build the UI

Projection Structure

Every projection file follows this pattern:

1. Imports

import {postgreSQLRawSQLProjection} from '@event-driven-io/emmett-postgresql';
import {sql} from '@event-driven-io/dumbo';
import knex, {Knex} from 'knex';
import {EventType} from '../../events/EventType';

2. Read Model Types

Define TypeScript types for the read model:

export type {Name}ReadModelItem = {
    field1?: type,
    field2?: type,
    // fields from projection
}

export type {Name}ReadModel = {
    data: {Name}ReadModelItem[],
}

export const tableName = 'table_name';

3. Knex Instance Helper

export const getKnexInstance = (connectionString: string): Knex => {
    return knex({
        client: 'pg',
        connection: connectionString,
    });
};

4. Projection Definition

export const {Name}Projection = postgreSQLRawSQLProjection<EventType>({
    canHandle: ["Event1", "Event2"],  // events this projection handles
    evolve: (event, context) => {
        const {type, data} = event;
        const db = getKnexInstance(context.connection.connectionString);

        switch (type) {
            case "Event1":
                return sql(db(tableName)
                    .withSchema('public')
                    .insert({
                        field1: data.field1,
                        field2: data.field2,
                    })
                    .onConflict('id_field')  // upsert on conflict
                    .merge({field1: data.field1, field2: data.field2})
                    .toQuery());

            case "Event2":
                return sql(db(tableName)
                    .withSchema('public')
                    .where('id_field', data.id)
                    .update({
                        field1: data.field1,
                    })
                    .toQuery());

            default:
                return [];
        }
    }
});

Key Patterns

Insert with Upsert (Merge on Conflict)

Use this pattern for events that create or update records:

return sql(db(tableName)
    .withSchema('public')
    .insert({ /* fields */ })
    .onConflict('id_field')
    .merge({ /* fields to update */ })
    .toQuery());

Update Pattern

Use this for events that only update existing records:

return sql(db(tableName)
    .withSchema('public')
    .where('id_field', data.id)
    .update({ /* fields */ })
    .toQuery());

Delete Pattern

Use this for events that remove records:

return sql(db(tableName)
    .withSchema('public')
    .where('id_field', data.id)
    .delete()
    .toQuery());

Testing

Every projection MUST have tests using PostgreSQLProjectionSpec with Testcontainers:

import {before, after, describe, it} from "node:test";
import {PostgreSQLProjectionAssert, PostgreSQLProjectionSpec} from "@event-driven-io/emmett-postgresql";
import {{Name}Projection} from "./{Name}Projection";
import {PostgreSqlContainer, StartedPostgreSqlContainer} from "@testcontainers/postgresql";
import {EventType} from "../../events/EventType"
import knex, {Knex} from 'knex';
import assert from 'assert';
import {runFlywayMigrations} from "../../common/testHelpers";

describe('{Name} Specification', () => {
    let postgres: StartedPostgreSqlContainer;
    let connectionString: string;
    let db: Knex;

    let given: PostgreSQLProjectionSpec<EventType>

    before(async () => {
        postgres = await new PostgreSqlContainer("postgres").start();
        connectionString = postgres.getConnectionUri();

        db = knex({
            client: 'pg',
            connection: connectionString,
        });

        await runFlywayMigrations(connectionString);

        given = PostgreSQLProjectionSpec.for({
            projection: {Name}Projection,
            connectionString,
        });
    });

    after(async () => {
        await db?.destroy();
        await postgres?.stop();
    });

    it('spec: {Name} - scenario', async () => {
        const assertReadModel: PostgreSQLProjectionAssert = async ({connectionString: connStr}) => {
            const queryDb = knex({
                client: 'pg',
                connection: connStr,
            });

            try {
                const result = await queryDb('table_name')
                    .withSchema('public')
                    .select('*');

                assert.strictEqual(result.length, 1);
                // add more assertions
            } finally {
                await queryDb.destroy();
            }
        };

        await given([{
            type: 'EventName',
            data: { /* event data */ },
            metadata: {streamName: 'stream-id'}
        }])
            .when([])  // additional events to process
            .then(assertReadModel);
    });
});

Query API Routes

Every read model exposes a GET endpoint to fetch data:

import {Request, Response, Router} from 'express';
import {{Name}ReadModel, tableName} from "./{Name}Projection";
import {WebApiSetup} from "@event-driven-io/emmett-expressjs";
import createClient from "../../supabase/api";
import {readmodel} from "../../core/readmodel";
import {requireUser} from "../../supabase/requireUser";

export const api =
    (
        // external dependencies
    ): WebApiSetup =>
        (router: Router): void => {
            router.get('/api/query/{name}-collection', async (req: Request, res: Response) => {
                try {
                    const principal = await requireUser(req, res, true);
                    if (principal.error) {
                        return;
                    }

                    const userId = principal.user.id;
                    const id = req.query._id?.toString();

                    const supabase = createClient()

                    const query: any = {};
                    delete query._id;

                    const data: {Name}ReadModel | {Name}ReadModel[] | null =
                        id ? await readmodel(tableName, supabase).findById<{Name}ReadModel>("id_field", id) :
                            await readmodel(tableName, supabase).findAll<{Name}ReadModel>(query)

                    // Serialize, handling bigint properly
                    const sanitized = JSON.parse(
                        JSON.stringify(data || [], (key, value) =>
                            typeof value === 'bigint' ? value.toString() : value
                        )
                    );

                    return res.status(200).json(sanitized);
                } catch (err) {
                    console.error(err);
                    return res.status(500).json({ok: false, error: 'Server error'});
                }
            });
        };

Database Migrations

Each read model requires a migration file in supabase/migrations/:

Naming Convention: V{N}__{tablename}.sql

V{N} - Version number (sequential: V1, V2, V3, etc.)
{tablename} - Lowercase table name matching the projection's tableName

Example: V8__locations.sql

-- Create {tablename} table
CREATE TABLE IF NOT EXISTS "public"."{tablename}"
(
    id_field TEXT PRIMARY KEY,
    field1 TEXT,
    field2 INTEGER,
    field3 TEXT,
    restaurant_id uuid NOT NULL
);

Migration Guidelines:

Use IF NOT EXISTS for idempotency
Define PRIMARY KEY on the ID field used in onConflict()
CRITICAL: ALWAYS include restaurant_id uuid NOT NULL column (required for multi-tenancy)
Use appropriate SQL types (TEXT, INTEGER, BOOLEAN, TIMESTAMP, etc.)
Keep column names in snake_case (PostgreSQL convention)
Place files in supabase/migrations/ directory
Migrations run automatically via Flyway in tests

File Structure

src/slices/{SliceName}/
├── {SliceName}Projection.ts    # Projection logic
├── {SliceName}.test.ts          # Tests
└── routes.ts                    # Query endpoint

supabase/migrations/
└── V{N}__{tablename}.sql        # Database schema

slice json

in each slice folder, generate a file .slice.json

{
    "id" : "<slice id>",
    "slice": "<slice title>",
    "context": "<contextx>",
    "link": "https://miro.com/app/board/<board-id>=/?moveToWidget=<slice id>"
}

References

See templates/Locations/ for simple single-event projection example
See templates/Tables/ for multi-event projection with updates
See templates/V8__locations.sql for migration example
See templates/V2__tables.sql for migration example

UI Prompt

to build the UI prompt, list the following facts:

to build the UI - use this table "<schema>.<table_name>"

Payload example:
<payload example as JSON>

this is the table definition:
<table definition as SQL DDL>

import {after, before, describe, it} from "node:test";
import {PostgreSQLProjectionAssert, PostgreSQLProjectionSpec} from "@event-driven-io/emmett-postgresql";
import {LocationsProjection} from "./LocationsProjection";
import {PostgreSqlContainer, StartedPostgreSqlContainer} from "@testcontainers/postgresql";
import {LocationAdded} from "../../events/LocationAdded"
import knex, {Knex} from "knex";
import assert from "node:assert";
import {runFlywayMigrations} from "../../common/testHelpers";

describe('Locations Specification', () => {
    let postgres: StartedPostgreSqlContainer;
    let connectionString: string;
    let db: Knex;

    let given: PostgreSQLProjectionSpec<LocationAdded>

    before(async () => {
        postgres = await new PostgreSqlContainer("postgres").start();
        connectionString = postgres.getConnectionUri();

        db = knex({
            client: 'pg',
            connection: connectionString,
        });

        await runFlywayMigrations(connectionString);

        given = PostgreSQLProjectionSpec.for({
            projection: LocationsProjection,
            connectionString,
        });
    });

    after(async () => {
        await db?.destroy();
        await postgres?.stop();
    });

    it('spec: Locations - scenario', async () => {
        const city = "New York"
        const housenumber = "123"
        const location_id = "7771b53c-b378-4f10-a6b7-16f837316960"
        const name = "Main Office"
        const street = "Broadway"
        const zipCode = "10001"

        const assertLocations: PostgreSQLProjectionAssert = async ({connectionString: connStr}) => {
            const queryDb = knex({
                client: 'pg',
                connection: connStr,
            });

            try {
                const result = await queryDb('locations')
                    .withSchema('public')
                    .select('*');

                assert.strictEqual(result.length, 1, 'Should have 1 location');
                assert.strictEqual(result[0].location_id, location_id);
                assert.strictEqual(result[0].name, name);
                assert.strictEqual(result[0].zip_code, zipCode);
                assert.strictEqual(result[0].city, city);
            } finally {
                await queryDb.destroy();
            }
        };

        await given([{
            type: 'LocationAdded',
            data: {
                city: city,
                housenumber: housenumber,
                name: name,
                street: street,
                zipCode: zipCode,
                location_id: location_id
            },
            metadata: {streamName: 'fbf165d3-5fe2-4a34-af24-f0ad64ca8412'}
        }])
            .when([])
            .then(assertLocations);
    });

});

import {postgreSQLRawSQLProjection} from '@event-driven-io/emmett-postgresql';
import {sql} from '@event-driven-io/dumbo';
import knex, {Knex} from 'knex';
import {LocationAdded} from '../../events/LocationAdded';

export type LocationsReadModelItem = {
    name?: string,
    zipCode?: string,
    city?: string,
    location_id?: string,
}

export type LocationsReadModel = {
    data: LocationsReadModelItem[],
}

export const tableName = 'locations';

export const getKnexInstance = (connectionString: string): Knex => {
    return knex({
        client: 'pg',
        connection: connectionString,
    });
};

export const LocationsProjection = postgreSQLRawSQLProjection<LocationAdded>({
    canHandle: ["LocationAdded"],
    evolve: (event, context) => {
        const {type, data} = event;
        const db = getKnexInstance(context.connection.connectionString);

        switch (type) {
            case "LocationAdded":
                return sql(db(tableName)
                    .withSchema('public')
                    .insert({
                        location_id: data.location_id,
                        name: data.name,
                        zip_code: data.zipCode,
                        city: data.city,
                    })
                    .onConflict('location_id')
                    .merge({name: data.name, zip_code: data.zipCode, city: data.city})
                    .toQuery());

            default:
                return [];
        }
    }
});

import {Request, Response, Router} from 'express';
import {LocationsReadModel, tableName} from "./LocationsProjection";
import {WebApiSetup} from "@event-driven-io/emmett-expressjs";
import createClient from "../../supabase/api";
import {readmodel} from "../../core/readmodel";
import {requireUser} from "../../supabase/requireUser";

export const api =
    (
        // external dependencies
    ): WebApiSetup =>
        (router: Router): void => {
            router.get('/api/query/locations-collection', async (req: Request, res: Response) => {
                try {
                    const principal = await requireUser(req, res, true);
                    if (principal.error) {
                        return;
                    }

                    const userId = principal.user.id;
                    const id = req.query._id?.toString();

                    const supabase = createClient()

                    const query: any = {...req.query, user_id: userId};
                    delete query._id;

                    const data: LocationsReadModel | LocationsReadModel[] | null =
                        id ? await readmodel(tableName, supabase).findById<LocationsReadModel>("location_id", id) :
                            await readmodel(tableName, supabase).findAll<LocationsReadModel>(query)

                    // Serialize, handling bigint properly
                    const sanitized = JSON.parse(
                        JSON.stringify(data || [], (key, value) =>
                            typeof value === 'bigint' ? value.toString() : value
                        )
                    );

                    return res.status(200).json(sanitized);
                } catch (err) {
                    console.error(err);
                    return res.status(500).json({ok: false, error: 'Server error'});
                }
            });

        };

State-View Slice Templates

This folder contains real working examples from the codebase to use as templates for read model projections.

Simple Example: Locations

Files:

Locations/LocationsProjection.ts.sample - Single-event projection
Locations/Locations.test.ts.sample - Projection test
Locations/routes.ts.sample - Query API endpoint
V8__locations.sql - Database migration

Use this when:

Single event type creates/updates the read model
Simple insert with upsert (merge on conflict)
Basic read model with a few fields
No complex transformations needed

Key features:

canHandle: ["LocationAdded"] - handles one event type
Insert with .onConflict().merge() for upsert behavior
Simple field mapping from event to database
Query by ID or list all

Complex Example: Tables

Files:

Tables/TablesProjection.ts.sample - Multi-event projection
Tables/Tables.test.ts.sample - Test with multiple events
Tables/routes.ts.sample - Query API endpoint
V2__tables.sql - Database migration

Use this when:

Multiple event types update the same read model
Different operations (insert, update, delete)
Need to handle event sequences
More complex state management

Key features:

canHandle: ['TableAdded', 'TableUpdated'] - handles multiple events
Different SQL operations per event type:
TableAdded - insert with upsert
TableUpdated - direct update by ID
Test uses given([...]).when([...]).then(...) pattern for event sequences
Shows how to verify multiple records and updates

Migration Files

V8__locations.sql - Simple table with TEXT fields V2__tables.sql - Table with mixed types (TEXT, INTEGER)

All tables need the restaurant_id

Migration Naming:

Format: V{N}__{tablename}.sql
Sequential numbers (V1, V2, V3, etc.)
Double underscore before table name
Lowercase table name

Migration Patterns:

-- Always use IF NOT EXISTS
CREATE TABLE IF NOT EXISTS "public"."table_name"
(
    -- Primary key matching the onConflict field
    id_field TEXT PRIMARY KEY,

    -- Use appropriate SQL types
    text_field TEXT,
    number_field INTEGER,
    bool_field BOOLEAN,
    date_field TIMESTAMP
);

Query Endpoint Patterns

Both examples show the standard query endpoint pattern:

GET /api/query/{name}-collection
Optional ?_id=xxx parameter for single record
Returns array or single object
Requires authentication via requireUser
Uses readmodel helper for database queries
Handles bigint serialization

Testing Patterns

Setup:

Use Testcontainers for PostgreSQL
Run Flyway migrations before tests
Create Knex instance for assertions

Assertion Pattern:

const assertReadModel: PostgreSQLProjectionAssert = async ({connectionString}) => {
    const queryDb = knex({ client: 'pg', connection: connectionString });
    try {
        const result = await queryDb('table_name').select('*');
        // assertions
    } finally {
        await queryDb.destroy();
    }
};

Event Flow:

given([...]) - initial events to set up state
when([...]) - events to process during test
then(assertReadModel) - verify final state

import {Request, Response, Router} from 'express';
import {TablesReadModel, tableName} from "./TablesProjection";
import {WebApiSetup} from "@event-driven-io/emmett-expressjs";
import createClient from "../../supabase/api";
import {readmodel} from "../../core/readmodel";
import {requireUser} from "../../supabase/requireUser";

export const api =
    (
        // external dependencies
    ): WebApiSetup =>
        (router: Router): void => {
            router.get('/api/query/tables-collection', async (req: Request, res: Response) => {
                try {
                    const principal = await requireUser(req, res, true);
                    if (principal.error) {
                        return;
                    }

                    const userId = principal.user.id;
                    const id = req.query._id?.toString();

                    const supabase = createClient()

                    const query: any = {...req.query, user_id: userId};
                    delete query._id;

                    const data: TablesReadModel | TablesReadModel[] | null =
                        id ? await readmodel(tableName, supabase).findById<TablesReadModel>("table_id", id) :
                            await readmodel(tableName, supabase).findAll<TablesReadModel>(query)

                    // Serialize, handling bigint properly
                    const sanitized = JSON.parse(
                        JSON.stringify(data || [], (key, value) =>
                            typeof value === 'bigint' ? value.toString() : value
                        )
                    );

                    return res.status(200).json(sanitized);
                } catch (err) {
                    console.error(err);
                    return res.status(500).json({ok: false, error: 'Server error'});
                }
            });

        };

import {before, after, describe, it} from "node:test";
import {PostgreSQLProjectionAssert, PostgreSQLProjectionSpec} from "@event-driven-io/emmett-postgresql";
import {TablesProjection} from "./TablesProjection";
import {PostgreSqlContainer, StartedPostgreSqlContainer} from "@testcontainers/postgresql";
import {TableAdded} from "../../events/TableAdded"
import {TableUpdated} from "../../events/TableUpdated"
import knex from 'knex';
import assert from 'assert';
import {runFlywayMigrations} from "../../common/testHelpers";

describe('Tables Specification', () => {
    let postgres: StartedPostgreSqlContainer;
    let connectionString: string

    let given: PostgreSQLProjectionSpec<TableAdded | TableUpdated>

    before(async () => {
        postgres = await new PostgreSqlContainer("postgres").start();
        connectionString = postgres.getConnectionUri();

        await runFlywayMigrations(connectionString)

        given = PostgreSQLProjectionSpec.for({
            projection: TablesProjection,
            connectionString,
        });
    });

    after(async () => {
        await postgres?.stop();
    });

    it('spec: Tables - add and update scenario', async () => {
        const tableId1 = "fbf165d3-5fe2-4a34-af24-f0ad64ca8412"
        const tableId2 = "3ccaf75c-7120-4e4d-9de9-c5ad822a62ec"

        const assertTables: PostgreSQLProjectionAssert = async ({connectionString: connStr}) => {
            const queryDb = knex({
                client: 'pg',
                connection: connStr,
            });

            try {
                const result = await queryDb('tables')
                    .withSchema('public')
                    .select('table_id', 'name', 'seats')
                    .orderBy('table_id');

                assert.strictEqual(result.length, 2, 'Should have 2 tables');

                const table1 = result.find(t => t.table_id === tableId1);
                const table2 = result.find(t => t.table_id === tableId2);

                assert(table1, 'Table 1 should exist');
                assert.strictEqual(table1.name, "Table 1 Updated");
                assert.strictEqual(table1.seats, 8);

                assert(table2, 'Table 2 should exist');
                assert.strictEqual(table2.name, "Table 2");
                assert.strictEqual(table2.seats, 6);
            } finally {
                await queryDb.destroy();
            }
        };

        await given([{
            type: 'TableAdded',
            data: {
                minPersons: 2,
                name: "Table 1",
                reservable: true,
                seats: 4,
                tableid: tableId1
            },
            metadata: {streamName: 'fbf165d3-5fe2-4a34-af24-f0ad64ca8412'}
        }])
            .when([
                {
                    type: 'TableAdded',
                    data: {
                        minPersons: 2,
                        name: "Table 2",
                        reservable: true,
                        seats: 6,
                        tableid: tableId2
                    },
                    metadata: {streamName: 'fbf165d3-5fe2-4a34-af24-f0ad64ca8412'}
                },
                {
                    type: 'TableUpdated',
                    data: {
                        minPersons: 2,
                        name: "Table 1 Updated",
                        reservable: true,
                        seats: 8,
                        tableid: tableId1
                    },
                    metadata: {streamName: 'fbf165d3-5fe2-4a34-af24-f0ad64ca8412'}
                }
            ])
            .then(assertTables);
    });

});

import {postgreSQLRawSQLProjection} from '@event-driven-io/emmett-postgresql';
import {sql} from '@event-driven-io/dumbo';
import knex, {Knex} from 'knex';
import {ContextEvents} from '../../events/ContextEvents';

export type TablesReadModelItem = {
    seats: number,
    name: string,
    tableId: string,
}

export type TablesReadModel = {
    data: TablesReadModelItem[],
}

export const tableName = 'tables';

export const getKnexInstance = (connectionString: string): Knex => {
    return knex({
        client: 'pg',
        connection: connectionString,
    });
};


export const TablesProjection = postgreSQLRawSQLProjection<ContextEvents>({
    canHandle: ['TableAdded', 'TableUpdated'],
    evolve: (event, context) => {
        const {type, data} = event;
        const db = getKnexInstance(context.connection.connectionString);

        switch (type) {
            case 'TableAdded':
                return sql(db(tableName)
                    .withSchema('public')
                    .insert({
                        table_id: data.tableid,
                        name: data.name,
                        seats: data.seats,
                    })
                    .onConflict('table_id')
                    .merge({name: data.name, seats: data.seats})
                    .toQuery());

            case 'TableUpdated':
                return sql(db(tableName)
                    .withSchema('public')
                    .where('table_id', data.tableid)
                    .update({
                        name: data.name,
                        seats: data.seats,
                    })
                    .toQuery());

            default:
                return [];
        }
    }
});