dev: Added new transaction methods

packages/data/src/adapter.test.ts

@@ -1,22 +1,34 @@
 import { test, expect, assert } from 'vitest';
-import { adapter, database, schema } from '@do-ob/data';
-import { Ambit, inputify } from '@do-ob/core';
-import { randomUUID } from 'node:crypto';
+import { adapter, Database, database, schema } from '@do-ob/data';
+import { inputify } from '@do-ob/core';
+import { seed } from './seed';
+
+async function prepareInput(db: Database) {
+  seed();
+  const [ subject ] = await db.insert(schema.entity).values({}).returning({ $id: schema.entity.$id });
+  const [ dispatch ] = await db.insert(schema.dispatch).values({
+    $subject: subject.$id,
+    $action: 'register',
+  }).returning({ $id: schema.dispatch.$id });
+
+  return inputify({
+    $dispatch: dispatch.$id,
+    $subject: subject.$id,
+  });
+}
 
 test('should insert a new entity into the database', async () => {
   const db = database();
   const dbAdapter = adapter(db);
 
-  const input = inputify({
-    $dispatch: randomUUID(),
-    $subject: randomUUID(),
-  });
+  const input = await prepareInput(db);
+
   const result = await dbAdapter.insert(input)(schema.locale, {
     name: 'my_locale',
     code: 'en-US'
   });
 
-  expect(result).toBeDefined();
+  assert(result);
   expect(result).toMatchObject({
     name: 'my_locale',
     code: 'en-US',
@@ -53,11 +65,8 @@ test('should update an entity in the database', async () => {
   const db = database();
   const dbAdapter = adapter(db);
 
-  const input = inputify({
-    $dispatch: randomUUID(),
-    $subject: randomUUID(),
-    ambit: Ambit.Global,
-  });
+  const input = await prepareInput(db);
+
   const inserted = await dbAdapter.insert(input)(schema.locale, {
     name: 'my_locale_to_update',
     code: 'en-US',
@@ -78,5 +87,13 @@ test('should update an entity in the database', async () => {
     content: 'this is the updated content'
   });
 
-});
+  const mutations = await db.query.mutate.findMany({
+    where: (table, { and, eq }) => and(
+      eq(table.$entity, result.$id),
+      eq(table.operation, 'update'),
+    ),
+  });
+
+  expect(mutations).toHaveLength(1);
 
+});

packages/data/src/adapter.ts

@@ -1,9 +1,10 @@
 import { adaptify, Ambit } from '@do-ob/core';
-import { entity, mutate } from '@do-ob/data/schema';
+import { schema } from '@do-ob/data/schema';
 import { Database } from '@do-ob/data/database';
 import { PgTableWithColumns, TableConfig } from 'drizzle-orm/pg-core';
 import { getTableName, eq, SQL, sql, and, getTableColumns } from 'drizzle-orm';
 import { RowList } from 'postgres';
+import { insert } from './transaction';
 
 /**
  * Builds an sql filter based on an ambit.
@@ -16,9 +17,9 @@ function scope(
     case Ambit.Global:
       return sql`true`;
     case Ambit.Owned:
-      return eq(entity.table.$owner, $subject);
+      return eq(schema.entity.$owner, $subject);
     case Ambit.Created:
-      return eq(entity.table.$creator, $subject);
+      return eq(schema.entity.$creator, $subject);
     case Ambit.Member:
       return sql`false`; // TODO: Implement member scope.
     case Ambit.None:
@@ -39,12 +40,16 @@ export function adapter<
     /**
      * Safely inserts a new entity into the database with authorization controls and audits.
      */
-    insert: ({ $dispatch, $subject }) => async <
+    insert: ({ $subject, $dispatch }) => async <
       C extends TableConfig,
     >(
       table: PgTableWithColumns<C>,
       value: Omit<PgTableWithColumns<C>['$inferInsert'], '$id'>,
     ) => {
+      if(!$subject) {
+        return;
+      }
+
       const db = await database;
 
       /**
@@ -55,52 +60,24 @@ export function adapter<
         throw new Error('Only self-declared entity tables, prefixed with "entity_", can be logically inserted.');
       }
 
-      /**
-       * Begin the database transaction.
-       */
-      return db.transaction(async (tx) => {
-        /**
-         * Create an entity record.
-         */
-        const [ entityRecord ] = await tx.insert(entity.table).values({
-          type: tableName.replace('entity_', ''),
-          $owner: $subject,
-          $creator: $subject,
-        }).returning();
-
-        /**
-         * Create the entity type record.
-         */
-        const [ typeRecord ] = await tx.insert(table).values({
-          ...value as PgTableWithColumns<C>['$inferInsert'],
-          $id: entityRecord.$id,
-        }).returning();
-
-        /**
-         * Create a mutation record of the insert.
-         */
-        tx.insert(mutate.table).values([
+      const [ result, entity ] = await db.transaction(
+        insert(
+          table,
+          value,
           {
-            $dispatch: $dispatch,
-            $entity: entityRecord.$id,
-            table: tableName,
-            operation: 'create',
-            mutation: typeRecord,
+            $owner: $subject,
+            $creator: $subject
           },
           {
             $dispatch: $dispatch,
-            $entity: entityRecord.$id,
-            table: getTableName(entity.table),
-            operation: 'create',
-            mutation: entityRecord,
           }
-        ]);
-
-        return {
-          ...typeRecord as PgTableWithColumns<C>['$inferInsert'],
-          entity: entityRecord,
-        };
-      });
+        ),
+      );
+
+      return {
+        ...result,
+        entity,
+      };
     },
 
     /**
@@ -140,10 +117,10 @@ export function adapter<
          */
         const updateChunks: SQL[] = [];
         updateChunks.push(tx.update(table).set(next as object).getSQL());
-        updateChunks.push(sql`from ${entity.table}`);
+        updateChunks.push(sql`from ${schema.entity}`);
         updateChunks.push(sql`where ${and(
           eq(table.$id, $id),
-          eq(table.$id, entity.table.$id),
+          eq(table.$id, schema.entity.$id),
           scope($subject, ambit),
         )}`);
         updateChunks.push(sql`returning ${
@@ -176,7 +153,7 @@ export function adapter<
         /**
          * Create the mutation record.
          */
-        tx.insert(mutate.table).values({
+        tx.insert(schema.mutate).values({
           $dispatch: $dispatch,
           $entity: $id,
           table: tableName,

packages/data/src/seed/action.ts

@@ -3,6 +3,11 @@ import { SchemaInsert } from '@do-ob/data/schema';
 
 export const records: Array<SchemaInsert['action']> = Object.keys(actionSchema).map((key) => {
   const definition = actionSchema[key as keyof typeof actionSchema];
+
+  if  (!definition) {
+    throw new Error(`No definition found for ${key}`);
+  }
+
   return {
     $id: definition.$id,
     definition: definition,

packages/data/src/transaction.ts

@@ -2,6 +2,8 @@ import type { Transaction } from './transaction/transaction.types.ts';
 import { database } from '@do-ob/data/database';
 
 export * from './transaction/entity.ts';
+export * from './transaction/insert.ts';
+export * from './transaction/audit.ts';
 
 export async function transact<R>(transaction: (tx: Transaction) => Promise<R>): Promise<R> {
   const db = await database();

packages/data/src/transaction/audit.ts

@@ -0,0 +1,32 @@
+import type { Transaction } from './transaction.types';
+import { getTableName, type TableConfig } from 'drizzle-orm';
+import type { PgTableWithColumns } from 'drizzle-orm/pg-core';
+import { schema } from '@do-ob/data/schema';
+
+export interface AuditMutationChanges<C extends TableConfig> {
+  type: 'create' | 'update' | 'delete';
+  table: PgTableWithColumns<C>,
+  value: PgTableWithColumns<C>['$inferSelect'] & { $id: string },
+}
+
+export function auditMutation<
+  C extends TableConfig,
+>(
+  $dispatch: string,
+  mutations: AuditMutationChanges<C>[],
+) {
+  return async (tx: Transaction) => {
+    /**
+     * Create a mutation record of the insert.
+     */
+    await tx.insert(schema.mutate).values(
+      mutations.map(mutation => ({
+        $dispatch,
+        $entity: mutation.value.$id,
+        table: getTableName(mutation.table),
+        operation: mutation.type,
+        mutation: mutation.value
+      }))
+    );
+  };
+}

packages/data/src/transaction/insert.ts

@@ -0,0 +1,66 @@
+import type { Transaction } from './transaction.types';
+import { getTableName, type TableConfig } from 'drizzle-orm';
+import type { PgTableWithColumns } from 'drizzle-orm/pg-core';
+import { schema } from '@do-ob/data/schema';
+
+export function insert<
+  C extends TableConfig,
+> (
+  table: PgTableWithColumns<C>,
+  value: Omit<PgTableWithColumns<C>['$inferInsert'], '$id'>,
+  meta: {
+    $owner?: string,
+    $creator?: string,
+  } = {},
+  audit?: {
+    $dispatch: string,
+  },
+) {
+  return async (tx: Transaction): Promise<[PgTableWithColumns<C>['$inferSelect'], typeof schema.entity.$inferSelect]> => {
+    const tableName = getTableName(table);
+
+    /**
+         * Create an entity record.
+         */
+    const [ entityRecord ] = await tx.insert(schema.entity).values({
+      type: tableName.replace('entity_', ''),
+      ...meta
+    }).returning();
+
+    /**
+     * Create the entity type record.
+     */
+    const [ typeRecord ] = await tx.insert(table).values({
+      ...value as PgTableWithColumns<C>['$inferInsert'],
+      $id: entityRecord.$id,
+    }).returning();
+
+    if (!typeRecord) {
+      tx.rollback();
+    }
+
+    if (audit) {
+      tx.insert(schema.mutate).values([
+        {
+          $dispatch: audit.$dispatch,
+          $entity: entityRecord.$id,
+          table: tableName,
+          operation: 'create',
+          mutation: typeRecord,
+        },
+        {
+          $dispatch: audit.$dispatch,
+          $entity: entityRecord.$id,
+          table: tableName,
+          operation: 'create',
+          mutation: entityRecord,
+        }
+      ]);
+    }
+
+    return [
+      typeRecord as PgTableWithColumns<C>['$inferSelect'],
+      entityRecord,
+    ];
+  };
+}