Merge pull request #18 from wataruoguchi/refactor_primary_keys

fix(emmett-event-store-kysely): deprecate primary_keys

Author: wataruoguchi

docs/emmett-event-store-kysely.md

@@ -185,7 +185,6 @@ const registry = createSnapshotProjectionRegistry(
   ["CartCreated", "ItemAdded", "CartCheckedOut"],
   {
     tableName: "carts",
-    primaryKeys: ["tenant_id", "cart_id", "partition"],
     extractKeys: (event, partition) => ({
       tenant_id: event.data.eventMeta.tenantId,
       cart_id: event.data.eventMeta.cartId,
@@ -201,6 +200,16 @@ const registry = createSnapshotProjectionRegistry(
 );
 ```
 
+**Arguments:**
+
+- **First argument**: Array of event types to handle
+- **Second argument**: Configuration object
+  - `tableName`: Database table name for the projection
+  - `extractKeys`: Function that returns primary key values from the event (keys are inferred automatically)
+  - `evolve`: Your domain's evolve function (reuse from write model)
+  - `initialState`: Function that returns the initial aggregate state
+  - `mapToColumns` _(optional)_: Function to denormalize state fields into table columns for querying
+
 ### Process Events (On-Demand)
 
 For on-demand processing (tests, backfills, or scheduled jobs), use the projection runner:
@@ -339,7 +348,6 @@ const registry = createSnapshotProjectionRegistry(
   ["CartCreated", "ItemAdded"],
   {
     tableName: "carts",
-    primaryKeys: ["tenant_id", "cart_id"],
     extractKeys: (event, partition) => ({ /* ... */ }),
     evolve: domainEvolve,
     initialState: () => ({ /* ... */ }),

example/src/docs/PROJECTIONS_ARCHITECTURE.md

@@ -42,7 +42,6 @@ export function cartsSnapshotProjection() {
     ["CartCreated", "ItemAdded", "CartCheckedOut"],
     {
       tableName: "carts",
-      primaryKeys: ["tenant_id", "cart_id", "partition"],
       extractKeys: (event, partition) => ({
         tenant_id: event.data.eventMeta.tenantId,
         cart_id: event.data.eventMeta.cartId,

example/src/docs/README.md

@@ -53,7 +53,6 @@ const registry = createSnapshotProjectionRegistry(
   ["CartCreated", "ItemAdded", "CartCheckedOut"],
   {
     tableName: "carts",
-    primaryKeys: ["tenant_id", "cart_id", "partition"],
     extractKeys: (event, partition) => ({...}),
     evolve: domainEvolve,  // Same logic as write model
     initialState,

example/src/modules/cart/application/event-sourcing/cart.read-model.ts

@@ -69,7 +69,6 @@ export function cartsSnapshotProjection(): ProjectionRegistry<DatabaseExecutor>
     ],
     {
       tableName: "carts",
-      primaryKeys: ["tenant_id", "cart_id", "partition"],
       extractKeys: (
         event: ProjectionEvent<CartDomainEvent>,
         partition: string,

example/src/modules/generator/application/event-sourcing/generator.read-model.ts

@@ -60,7 +60,6 @@ export function generatorsSnapshotProjection(): ProjectionRegistry<DatabaseExecu
     GeneratorDomainEvent
   >(["GeneratorCreated", "GeneratorUpdated", "GeneratorDeleted"], {
     tableName: "generators",
-    primaryKeys: ["tenant_id", "generator_id", "partition"],
     extractKeys: (
       event: ProjectionEvent<GeneratorDomainEvent>,
       partition: string,

packages/emmett-event-store-kysely/.releaserc.json

@@ -1,4 +1,3 @@
 {
   "extends": ["semantic-release-monorepo", "../../.releaserc.json"]
 }
-

packages/emmett-event-store-kysely/README.md

@@ -81,7 +81,6 @@ const registry = createSnapshotProjectionRegistry(
   ["CartCreated", "ItemAdded", "CartCheckedOut"],
   {
     tableName: "carts",
-    primaryKeys: ["tenant_id", "cart_id", "partition"],
     extractKeys: (event, partition) => ({
       tenant_id: event.data.eventMeta.tenantId,
       cart_id: event.data.eventMeta.cartId,

packages/emmett-event-store-kysely/src/index.ts

@@ -10,6 +10,7 @@ export type {
   ProjectionReadStreamOptions,
 } from "./event-store/kysely-event-store.js";
 export { createProjectionRunner } from "./projections/runner.js";
+export type { ProjectEvents } from "./projections/runner.js";
 export {
   createSnapshotProjection,
   createSnapshotProjectionRegistry,

packages/emmett-event-store-kysely/src/projections/runner.ts

@@ -20,11 +20,17 @@ export type ProjectionRunnerDeps = {
   registry: ProjectionRegistry;
 };
 
+export type ProjectEvents = (
+  subscriptionId: string,
+  streamId: string,
+  opts?: { partition?: string; batchSize?: number },
+) => Promise<{ processed: number; currentStreamVersion: bigint }>;
+
 export function createProjectionRunner({
   db,
   readStream,
   registry,
-}: ProjectionRunnerDeps) {
+}: ProjectionRunnerDeps): { projectEvents: ProjectEvents } {
   type EventWithMetadata = Event & {
     metadata: {
       streamId: string;

packages/emmett-event-store-kysely/src/projections/snapshot-projection.ts

@@ -25,13 +25,17 @@ export type SnapshotProjectionConfig<
   tableName: TTable;
 
   /**
-   * The primary key columns that uniquely identify a row
+   * @deprecated The primary key columns are now automatically inferred from the keys returned by extractKeys.
+   * This field is optional and will be removed in a future version.
+   *
+   * If provided, it will be validated against the keys returned by extractKeys.
    * e.g., ['tenant_id', 'cart_id', 'partition']
    */
-  primaryKeys: string[];
+  primaryKeys?: string[];
 
   /**
-   * Extract primary key values from the event data
+   * Extract primary key values from the event data.
+   * The keys of the returned object will be used as the primary key columns for upsert operations.
    */
   extractKeys: (
     event: ProjectionEvent<E>,
@@ -78,7 +82,6 @@ export type SnapshotProjectionConfig<
  * ```typescript
  * const cartProjection = createSnapshotProjection({
  *   tableName: 'carts',
- *   primaryKeys: ['tenant_id', 'cart_id', 'partition'],
  *   extractKeys: (event, partition) => ({
  *     tenant_id: event.data.eventMeta.tenantId,
  *     cart_id: event.data.eventMeta.cartId,
@@ -103,21 +106,24 @@ export function createSnapshotProjection<
 >(
   config: SnapshotProjectionConfig<TState, TTable, E>,
 ): ProjectionHandler<DatabaseExecutor, E> {
-  const {
-    tableName,
-    primaryKeys,
-    extractKeys,
-    evolve,
-    initialState,
-    mapToColumns,
-  } = config;
+  const { tableName, extractKeys, evolve, initialState, mapToColumns } = config;
+
+  // Cache the inferred primary keys after the first call
+  let inferredPrimaryKeys: string[] | undefined;
 
   return async (
     { db, partition }: ProjectionContext<DatabaseExecutor>,
     event: ProjectionEvent<E>,
   ) => {
     const keys = extractKeys(event, partition);
 
+    // Infer primary keys from extractKeys on first call
+    if (!inferredPrimaryKeys) {
+      inferredPrimaryKeys = Object.keys(keys);
+    }
+
+    const primaryKeys = inferredPrimaryKeys;
+
     // Check if event is newer than what we've already processed
     // Note: Casting to `any` is necessary because Kysely cannot infer types for dynamic table names.
     // The table name is provided at runtime, so TypeScript cannot verify the table structure at compile time.
@@ -212,7 +218,6 @@ export function createSnapshotProjection<
  *   ['CartCreated', 'ItemAddedToCart', 'ItemRemovedFromCart'],
  *   {
  *     tableName: 'carts',
- *     primaryKeys: ['tenant_id', 'cart_id', 'partition'],
  *     extractKeys: (event, partition) => ({
  *       tenant_id: event.data.eventMeta.tenantId,
  *       cart_id: event.data.eventMeta.cartId,