refactor(prisma-adapter): separate event/snapshot/view tables (#81)

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: tonyfromundefined

.changeset/dark-experts-allow.md

@@ -0,0 +1,5 @@
+---
+"ventyd": patch
+---
+
+fix: add snapshot feature in prisma adapter

src/adapter/prisma.ts

@@ -5,7 +5,7 @@ import type {
   InferStateFromSchema,
 } from "../types";
 
-type BaseEventTypeRow = Omit<
+type BaseEventRow = Omit<
   BaseEventType,
   "eventCreatedAt" | "body" | "version"
 > & {
@@ -14,15 +14,30 @@ type BaseEventTypeRow = Omit<
   version?: number;
 };
 
+type BaseSnapshotRow = {
+  entityId: string;
+  entityName: string;
+  state: unknown;
+  version: number;
+};
+
 export type VentydAdapterOptions<
   $$Schema,
-  $$PrismaEventRow extends BaseEventTypeRow,
+  $$PrismaEventRow extends BaseEventRow,
   $$PrismaEventRowInput,
+  $$PrismaSnapshotRow extends BaseSnapshotRow,
   $$PrismaSnapshotRowInput,
+  $$PrismaViewRowInput,
 > = {
   prisma: {
     $transaction: (commands: unknown[]) => Promise<unknown[]>;
   };
+  /**
+   * Save a snapshot every N events. When the latest event version is a
+   * multiple of this value, a snapshot is written inside the same transaction
+   * as the events and view update.
+   */
+  snapshotEvery?: number;
   tables: {
     event: {
       findMany(args?: {
@@ -35,30 +50,46 @@ export type VentydAdapterOptions<
       createMany(args?: { data: $$PrismaEventRowInput[] }): Promise<unknown>;
     };
     snapshot: {
+      findFirst(args: {
+        where: { entityId: string; entityName: string };
+      }): Promise<$$PrismaSnapshotRow | null>;
       upsert(args: {
-        where: { id?: string };
+        where: { entityId: string };
         update: $$PrismaSnapshotRowInput;
         create: $$PrismaSnapshotRowInput;
       }): Promise<unknown>;
     };
+    view: {
+      upsert(args: {
+        where: { entityId: string };
+        update: $$PrismaViewRowInput;
+        create: $$PrismaViewRowInput;
+      }): Promise<unknown>;
+    };
   };
-  entityToRow(args: {
+  entityToViewRow(args: {
     entityId: string;
+    entityName: string;
     state: InferStateFromSchema<$$Schema>;
-  }): $$PrismaSnapshotRowInput;
+    version: number;
+  }): $$PrismaViewRowInput;
 };
 
 export function prismaAdapter<
   $$Schema,
-  $$PrismaEventRow extends BaseEventTypeRow,
+  $$PrismaEventRow extends BaseEventRow,
   $$PrismaEventRowInput,
+  $$PrismaSnapshotRow extends BaseSnapshotRow,
   $$PrismaSnapshotRowInput,
+  $$PrismaViewRowInput,
 >(
   options: VentydAdapterOptions<
     $$Schema,
     $$PrismaEventRow,
     $$PrismaEventRowInput,
-    $$PrismaSnapshotRowInput
+    $$PrismaSnapshotRow,
+    $$PrismaSnapshotRowInput,
+    $$PrismaViewRowInput
   >,
 ): Adapter<$$Schema> {
   function eventToRow(_event: InferEventFromSchema<$$Schema>) {
@@ -105,22 +136,64 @@ export function prismaAdapter<
 
       return rows.map(rowToEvent);
     },
-    async commitEvents({ events, entityId, state }) {
-      const snapshotRow = options.entityToRow({
+
+    async getSnapshot({ entityName, entityId }) {
+      const row = await options.tables.snapshot.findFirst({
+        where: { entityId, entityName },
+      });
+
+      if (!row) return null;
+
+      return {
+        state: row.state as InferStateFromSchema<$$Schema>,
+        version: row.version,
+      };
+    },
+
+    async commitEvents({ events, entityId, entityName, state }) {
+      const lastEvent = events[events.length - 1] as BaseEventType | undefined;
+      const lastVersion = lastEvent?.version ?? 0;
+
+      const viewRow = options.entityToViewRow({
         entityId,
+        entityName,
         state,
+        version: lastVersion,
       });
 
-      await options.prisma.$transaction([
+      const transactionCommands: unknown[] = [
         options.tables.event.createMany({
           data: events.map(eventToRow),
         }),
-        options.tables.snapshot.upsert({
-          where: { id: entityId },
-          update: snapshotRow,
-          create: snapshotRow,
+        options.tables.view.upsert({
+          where: { entityId },
+          update: viewRow,
+          create: viewRow,
         }),
-      ]);
+      ];
+
+      const shouldSnapshot =
+        options.snapshotEvery != null &&
+        lastVersion > 0 &&
+        lastVersion % options.snapshotEvery === 0;
+
+      if (shouldSnapshot) {
+        const snapshotRow = {
+          entityId,
+          entityName,
+          state,
+          version: lastVersion,
+        } as $$PrismaSnapshotRowInput;
+        transactionCommands.push(
+          options.tables.snapshot.upsert({
+            where: { entityId },
+            update: snapshotRow,
+            create: snapshotRow,
+          }),
+        );
+      }
+
+      await options.prisma.$transaction(transactionCommands);
     },
   };
 }