refactor(ledger): one layout table on every store, and content read in either byte shape

packages/ledger/src/content/content-places.ts

@@ -29,7 +29,13 @@ const HeadRows = Schema.Array(
   Schema.Struct({ upload_id: Schema.String, bytes: Schema.Number, chunks: Schema.Number, media_type: Schema.String }),
 );
 
-const ChunkRows = Schema.Array(Schema.Struct({ data: Schema.Uint8Array }));
+const StoredBytes = Schema.Union([Schema.Uint8Array, Schema.instanceOf(ArrayBuffer)]);
+
+const ChunkRows = Schema.Array(Schema.Struct({ data: StoredBytes }));
+
+function viewOf(stored: Readonly<Uint8Array> | Readonly<ArrayBuffer>): Uint8Array {
+  return stored instanceof Uint8Array ? stored : new Uint8Array(stored);
+}
 
 export function matching(key: Key): SQL {
   return SQL.merge(
@@ -84,7 +90,8 @@ async function chunkOf({ execute, shelf }: Shelved, upload: Key, index: number):
   const { rows } = await execute.query(
     SQL`SELECT data FROM ${SQL.identifier(shelf.chunks)} WHERE ${matching(upload)} AND chunk = ${index}`,
   );
-  return Schema.decodeUnknownSync(ChunkRows)(rows).at(0)?.data;
+  const chunk = Schema.decodeUnknownSync(ChunkRows)(rows).at(0);
+  return chunk === undefined ? undefined : viewOf(chunk.data);
 }
 
 export function uploadOf(place: Key, uploadId: string): Key {

packages/ledger/src/content/content-store.test.ts

@@ -1,6 +1,6 @@
 import { SQL } from '@event-driven-io/dumbo';
 import { sqliteFormatter } from '@event-driven-io/dumbo/sqlite';
-import { mostContentChunkBytes, sha256Of, textToKeep } from '@on.auto/operations';
+import { mostContentChunkBytes, sha256Of, textOfContent, textToKeep } from '@on.auto/operations';
 import { Effect } from 'effect';
 import { describe, expect, it } from 'vitest';
 
@@ -23,12 +23,13 @@ interface Recording {
   readonly statements: readonly string[];
 }
 
-function recordingFinding(heads: readonly unknown[]): Recording {
+function recordingAnswering(rowsOf: (statement: string) => readonly unknown[]): Recording {
   const statements: string[] = [];
   const execute: StatementExecutor = {
     query: (sql) => {
-      statements.push(described(sql));
-      return Promise.resolve({ rows: heads });
+      const statement = described(sql);
+      statements.push(statement);
+      return Promise.resolve({ rows: rowsOf(statement) });
     },
     command: (sql) => {
       statements.push(described(sql));
@@ -48,12 +49,27 @@ function recordingFinding(heads: readonly unknown[]): Recording {
   };
 }
 
+function recordingFinding(heads: readonly unknown[]): Recording {
+  return recordingAnswering(() => heads);
+}
+
+function recordingHoldingOneChunk(data: unknown): Recording {
+  return recordingAnswering((statement) => (statement.startsWith('SELECT data') ? [{ data }] : [headOf('upload')]));
+}
+
 function kindsOf({ statements }: Recording): readonly string[] {
   return statements.map((statement) => statement.split(' ').slice(0, 3).join(' '));
 }
 
 const largerThanTwoChunks = 'x'.repeat(2 * mostContentChunkBytes + 1);
 
+const keptBytes = new TextEncoder().encode('kept');
+
+const chunkShapes: ReadonlyArray<readonly [string, unknown]> = [
+  ['a Uint8Array', keptBytes],
+  ['an ArrayBuffer', keptBytes.slice().buffer],
+];
+
 function headOf(uploadId: string) {
   return { upload_id: uploadId, bytes: 4, chunks: 1, media_type: 'text/plain' };
 }
@@ -131,6 +147,19 @@ describe('the reads and the removals of a recorded content', () => {
     expect([read, kindsOf(recording)]).toEqual([undefined, ['SELECT upload_id, CAST(bytes']]);
   });
 
+  it.each(chunkShapes)('read a content back from a chunk its store answers as %s', async (_shape, chunk) => {
+    const recording = recordingHoldingOneChunk(chunk);
+
+    const read = await Effect.runPromise(
+      Effect.flatMap(
+        Effect.flatMap(sqliteContentOn(recording.pool).content.get(alpha, sha256Of('kept')), Effect.fromNullishOr),
+        textOfContent,
+      ),
+    );
+
+    expect(read).toBe('kept');
+  });
+
   it('mark a run unstaged, then remove the heads and the chunks it staged in one transaction, the heads first', async () => {
     const recording = recordingFinding([]);
 

packages/ledger/src/layout/layout-dialects.ts

@@ -1,59 +1,13 @@
 import { SQL } from '@event-driven-io/dumbo';
-import { Schema } from 'effect';
-
-import type { StatementExecutor } from '../event-store.ts';
-
-export interface FoundLayout {
-  readonly layout: number | undefined;
-  readonly tables: number;
-}
 
 export interface LayoutDialect {
-  readonly found: (execute: StatementExecutor) => Promise<FoundLayout>;
-  readonly written: (layout: number) => readonly SQL[];
+  readonly tablesOfTheSchema: () => SQL;
 }
 
-const noLayout = 0;
-
-const NumberedRow = Schema.Tuple([Schema.Struct({ tables: Schema.Number, layout: Schema.Number })]);
-
 export const sqliteLayout: LayoutDialect = {
-  found: async (execute) => {
-    const { rows } = await execute.query(
-      SQL`SELECT (SELECT count(*) FROM sqlite_master WHERE type = 'table') AS tables, user_version AS layout FROM pragma_user_version`,
-    );
-    const [{ tables, layout }] = Schema.decodeUnknownSync(NumberedRow)(rows);
-    return { layout: layout === noLayout ? undefined : layout, tables };
-  },
-  written: (layout) => [SQL`PRAGMA user_version = ${SQL.plain(String(layout))}`],
+  tablesOfTheSchema: () => SQL`SELECT name FROM sqlite_master WHERE type = 'table'`,
 };
 
-const layoutTable = 'ledger_layout';
-
-const TablesRow = Schema.Tuple([Schema.Struct({ tables: Schema.Number, numbered: Schema.Boolean })]);
-
-const LayoutRow = Schema.Tuple([Schema.Struct({ layout: Schema.Number })]);
-
-async function layoutKeptIn(execute: StatementExecutor): Promise<number> {
-  const { rows } = await execute.query(SQL`SELECT layout FROM ${SQL.identifier(layoutTable)}`);
-  const [{ layout }] = Schema.decodeUnknownSync(LayoutRow)(rows);
-  return layout;
-}
-
 export const postgresqlLayout: LayoutDialect = {
-  found: async (execute) => {
-    const { rows } = await execute.query(
-      SQL`SELECT count(*)::int AS tables, count(*) FILTER (WHERE tablename = ${layoutTable}) > 0 AS numbered
-        FROM pg_tables WHERE schemaname = current_schema()`,
-    );
-    const [{ tables, numbered }] = Schema.decodeUnknownSync(TablesRow)(rows);
-    return { layout: numbered ? await layoutKeptIn(execute) : undefined, tables };
-  },
-  written: (layout) => [
-    SQL`CREATE TABLE ${SQL.identifier(layoutTable)} (
-      one_row BOOLEAN PRIMARY KEY DEFAULT TRUE CHECK (one_row),
-      layout INTEGER NOT NULL
-    )`,
-    SQL`INSERT INTO ${SQL.identifier(layoutTable)} (layout) VALUES (${layout})`,
-  ],
+  tablesOfTheSchema: () => SQL`SELECT tablename AS name FROM pg_tables WHERE schemaname = current_schema()`,
 };

packages/ledger/src/layout/layout-kept.ts

@@ -1,20 +1,57 @@
+import { SQL } from '@event-driven-io/dumbo';
+import { Schema } from 'effect';
+
+import { namesFound } from '../content/content-tables.ts';
 import type { StatementExecutor } from '../event-store.ts';
 import { inTurn } from '../projections/inline-projection.ts';
-import type { FoundLayout, LayoutDialect } from './layout-dialects.ts';
+import type { LayoutDialect } from './layout-dialects.ts';
 import { ledgerLayoutRefused } from './ledger-layout-refused.ts';
 import { ledgerLayout } from './ledger-layout.ts';
 
+export interface FoundLayout {
+  readonly layout: number | undefined;
+  readonly tables: number;
+}
+
+const layoutTable = 'ledger_layout';
+
+const LayoutRow = Schema.Tuple([Schema.Struct({ layout: Schema.Number })]);
+
+async function layoutKeptIn(execute: StatementExecutor): Promise<number> {
+  const { rows } = await execute.query(SQL`SELECT layout FROM ${SQL.identifier(layoutTable)}`);
+  const [{ layout }] = Schema.decodeUnknownSync(LayoutRow)(rows);
+  return layout;
+}
+
+export async function layoutFound(execute: StatementExecutor, dialect: LayoutDialect): Promise<FoundLayout> {
+  const tables = await namesFound(execute, dialect.tablesOfTheSchema());
+  return { layout: tables.includes(layoutTable) ? await layoutKeptIn(execute) : undefined, tables: tables.length };
+}
+
+export function layoutWritten(execute: StatementExecutor, layout: number): Promise<void> {
+  return inTurn(
+    [
+      SQL`CREATE TABLE ${SQL.identifier(layoutTable)} (
+        one_row BOOLEAN NOT NULL PRIMARY KEY DEFAULT TRUE CHECK (one_row),
+        layout INTEGER NOT NULL
+      )`,
+      SQL`INSERT INTO ${SQL.identifier(layoutTable)} (layout) VALUES (${layout})`,
+    ],
+    (statement) => execute.command(statement),
+  );
+}
+
 function isNew({ layout, tables }: FoundLayout): boolean {
   return layout === undefined && tables === 0;
 }
 
 export async function requireTheLayoutKept(execute: StatementExecutor, dialect: LayoutDialect): Promise<void> {
-  const found = await dialect.found(execute);
+  const found = await layoutFound(execute, dialect);
   if (found.layout === ledgerLayout) {
     return;
   }
   if (!isNew(found)) {
     throw ledgerLayoutRefused({ found: found.layout, kept: ledgerLayout });
   }
-  await inTurn(dialect.written(ledgerLayout), (statement) => execute.command(statement));
+  await layoutWritten(execute, ledgerLayout);
 }

packages/ledger/src/layout/ledger-layout-behaviour.ts

@@ -1,8 +1,7 @@
 import { describe, expect, it } from 'vitest';
 
-import { inTurn } from '../projections/inline-projection.ts';
 import { aLedger, type LedgerEntry } from '../testing/ledger-entry.ts';
-import type { FoundLayout } from './layout-dialects.ts';
+import { layoutFound, layoutWritten, type FoundLayout } from './layout-kept.ts';
 import { LedgerLayoutRefused } from './ledger-layout-refused.ts';
 import { ledgerLayout } from './ledger-layout.ts';
 
@@ -12,13 +11,11 @@ const whatToDo =
   'Nothing stored before launch is carried over, so delete the local database and start the server again';
 
 export function layoutIn(entry: LedgerEntry, database: string): Promise<FoundLayout> {
-  return entry.executedOn(database, (execute) => entry.layout.found(execute));
+  return entry.executedOn(database, (execute) => layoutFound(execute, entry.layout));
 }
 
 function stampedWith(entry: LedgerEntry, database: string, layout: number): Promise<void> {
-  return entry.executedOn(database, (execute) =>
-    inTurn(entry.layout.written(layout), (statement) => execute.command(statement)),
-  );
+  return entry.executedOn(database, (execute) => layoutWritten(execute, layout));
 }
 
 function refusalOf(entry: LedgerEntry, database: string): Promise<unknown> {

packages/ledger/src/layout/ledger-layout.test.ts

@@ -9,7 +9,7 @@ import { temporaryDatabase } from '../testing/temporary-database.ts';
 import { ledgerLayout } from './ledger-layout.ts';
 
 const pinned = {
-  layout: 1,
+  layout: 2,
   catalog: [
     {
       name: 'kept_secrets',
@@ -23,6 +23,10 @@ const pinned = {
       name: 'ledger_first_messages_by_kind',
       sql: "CREATE INDEX ledger_first_messages_by_kind ON emt_messages (substr(stream_id, 1, instr(stream_id, '/') + instr(substr(stream_id, instr(stream_id, '/') + 1), '/') + instr(substr(stream_id, instr(stream_id, '/') + instr(substr(stream_id, instr(stream_id, '/') + 1), '/') + 1), '/') + instr(substr(stream_id, instr(stream_id, '/') + instr(substr(stream_id, instr(stream_id, '/') + 1), '/') + instr(substr(stream_id, instr(stream_id, '/') + instr(substr(stream_id, instr(stream_id, '/') + 1), '/') + 1), '/') + 1), '/')), global_position) WHERE stream_position = 1",
     },
+    {
+      name: 'ledger_layout',
+      sql: 'CREATE TABLE ledger_layout (one_row BOOLEAN NOT NULL PRIMARY KEY DEFAULT TRUE CHECK (one_row), layout INTEGER NOT NULL)',
+    },
     { name: 'ledger_messages_by_id', sql: 'CREATE INDEX ledger_messages_by_id ON emt_messages (message_id)' },
     {
       name: 'ledger_messages_by_playbook',

packages/ledger/src/layout/ledger-layout.ts

@@ -1 +1 @@
-export const ledgerLayout = 1;
+export const ledgerLayout = 2;

packages/ledger/src/layout/sqlite-file-in-layout.test.ts

@@ -0,0 +1,22 @@
+import { describe, expect, it, onTestFinished } from 'vitest';
+
+import { executedOn } from '../recorded/sqlite-checks.ts';
+import { temporaryDatabase } from '../testing/temporary-database.ts';
+import { sqliteLayout } from './layout-dialects.ts';
+import { layoutFound } from './layout-kept.ts';
+import { ledgerLayout } from './ledger-layout.ts';
+import { aSQLiteFileInLayout } from './sqlite-file-in-layout.ts';
+
+describe('a SQLite file put in a layout of the ledger', () => {
+  it('holds that layout in the table that keeps it, and no other table', async () => {
+    const { fileName, remove } = temporaryDatabase();
+    onTestFinished(remove);
+
+    await aSQLiteFileInLayout(fileName, ledgerLayout + 1);
+
+    expect(await executedOn(fileName, (execute) => layoutFound(execute, sqliteLayout))).toEqual({
+      layout: ledgerLayout + 1,
+      tables: 1,
+    });
+  });
+});

packages/ledger/src/layout/sqlite-file-in-layout.ts

@@ -0,0 +1,6 @@
+import { executedOn } from '../recorded/sqlite-checks.ts';
+import { layoutWritten } from './layout-kept.ts';
+
+export function aSQLiteFileInLayout(fileName: string, layout: number): Promise<void> {
+  return executedOn(fileName, (execute) => layoutWritten(execute, layout));
+}

packages/ledger/src/layout/sqlite-schema.test.ts

@@ -0,0 +1,52 @@
+import { SQL } from '@event-driven-io/dumbo';
+import { sqliteFormatter } from '@event-driven-io/dumbo/sqlite';
+import { describe, expect, it } from 'vitest';
+
+import type { StatementExecutor, TransactionalPool } from '../event-store.ts';
+import { ledgerLayout } from './ledger-layout.ts';
+import { sqliteLayoutKept } from './sqlite-schema.ts';
+
+function described(sql: SQL): string {
+  return SQL.describe(sql, sqliteFormatter).replaceAll(/\s+/gu, ' ').trim();
+}
+
+function aPoolHolding(tables: readonly string[]): { readonly pool: TransactionalPool; readonly asked: string[] } {
+  const asked: string[] = [];
+  const recordedAs = (where: string): StatementExecutor => ({
+    query: (sql) => {
+      asked.push(`${where}${described(sql)}`);
+      return Promise.resolve({ rows: tables.map((name) => ({ name })) });
+    },
+    command: (sql) => {
+      asked.push(`${where}${described(sql)}`);
+      return Promise.resolve();
+    },
+  });
+  return {
+    asked,
+    pool: {
+      execute: recordedAs('outside the transaction: '),
+      withTransaction: async (handle) => {
+        asked.push('BEGIN');
+        await handle({ execute: recordedAs('') });
+        asked.push('COMMIT');
+      },
+    },
+  };
+}
+
+describe('the layout of the ledger on SQLite, as the store migrates', () => {
+  it('lists the tables of the schema and writes the layout into a table of one row of a database that holds none, all in one transaction', async () => {
+    const { pool, asked } = aPoolHolding([]);
+
+    await sqliteLayoutKept(pool);
+
+    expect(asked).toEqual([
+      'BEGIN',
+      "SELECT name FROM sqlite_master WHERE type = 'table'",
+      'CREATE TABLE ledger_layout ( one_row BOOLEAN NOT NULL PRIMARY KEY DEFAULT TRUE CHECK (one_row), layout INTEGER NOT NULL )',
+      `INSERT INTO ledger_layout (layout) VALUES (${ledgerLayout})`,
+      'COMMIT',
+    ]);
+  });
+});

packages/ledger/src/layout/sqlite-schema.ts

@@ -1,13 +1,13 @@
 import { sqliteContent } from '../content/content-dialects.ts';
 import { createdContentTables } from '../content/content-tables.ts';
-import type { StatementExecutor } from '../event-store.ts';
+import { inTransactionOf, type StatementExecutor, type TransactionalPool } from '../event-store.ts';
 import { createSQLitePlaybookIndexes } from '../recorded/sqlite-indexes.ts';
 import { createdSecretTable } from '../secrets/kept-secrets.ts';
 import { sqliteLayout } from './layout-dialects.ts';
 import { requireTheLayoutKept } from './layout-kept.ts';
 
-export function sqliteLayoutKept(execute: StatementExecutor): Promise<void> {
-  return requireTheLayoutKept(execute, sqliteLayout);
+export function sqliteLayoutKept(pool: TransactionalPool): Promise<void> {
+  return inTransactionOf(pool)((execute) => requireTheLayoutKept(execute, sqliteLayout));
 }
 
 export async function sqliteSchemaCreated(execute: StatementExecutor): Promise<void> {