.opencode/node_modules/effect/dist/unstable/eventlog/SqlEventLogServerUnencrypted.js

/**
 * SQL-backed storage for unencrypted event-log servers.
 *
 * This module provides the durable `Storage` implementation used by
 * `EventLogServerUnencrypted` when remote entries should be stored in a SQL
 * database and streamed back to clients by store sequence. It creates
 * dialect-specific tables for the server remote id, per-store sequence state,
 * plaintext entries, and session authentication bindings.
 *
 * @since 4.0.0
 */
import * as Arr from "../../Array.js";
import * as Effect from "../../Effect.js";
import * as Layer from "../../Layer.js";
import * as PubSub from "../../PubSub.js";
import * as RcMap from "../../RcMap.js";
import * as Schema from "../../Schema.js";
import * as Stream from "../../Stream.js";
import * as SqlClient from "../sql/SqlClient.js";
import * as SqlError from "../sql/SqlError.js";
import { Entry, EntryId, makeRemoteIdUnsafe, RemoteEntry } from "./EventJournal.js";
import * as EventLogServerUnencrypted from "./EventLogServerUnencrypted.js";
/**
 * Creates unencrypted event-log server `Storage` backed by SQL.
 *
 * **Details**
 *
 * The implementation creates tables for the server remote id, store sequences,
 * entries, and session authentication bindings, then persists and streams
 * plaintext remote entries.
 *
 * @category constructors
 * @since 4.0.0
 */
export const makeStorage = options => Effect.gen(function* () {
  const sql = (yield* SqlClient.SqlClient).withoutTransforms();
  const entriesTable = options?.entryTablePrefix ?? "effect_events";
  const remoteIdTable = options?.remoteIdTable ?? "effect_remote_id";
  const insertBatchSize = options?.insertBatchSize ?? 200;
  const storesTable = `${entriesTable}_stores`;
  const sessionAuthBindingsTable = `${entriesTable}_session_auth_bindings`;
  const remoteIdTableSql = sql(remoteIdTable);
  const entriesTableSql = sql(entriesTable);
  const storesTableSql = sql(storesTable);
  const sessionAuthBindingsTableSql = sql(sessionAuthBindingsTable);
  yield* sql.onDialectOrElse({
    pg: () => sql`
          CREATE TABLE IF NOT EXISTS ${remoteIdTableSql} (
            singleton INT PRIMARY KEY,
            remote_id BYTEA NOT NULL
          )`,
    mysql: () => sql`
          CREATE TABLE IF NOT EXISTS ${remoteIdTableSql} (
            singleton INT PRIMARY KEY,
            remote_id BINARY(16) NOT NULL
          )`,
    mssql: () => sql`
          CREATE TABLE IF NOT EXISTS ${remoteIdTableSql} (
            singleton INT PRIMARY KEY,
            remote_id VARBINARY(16) NOT NULL
          )`,
    orElse: () => sql`
          CREATE TABLE IF NOT EXISTS ${remoteIdTableSql} (
            singleton INTEGER PRIMARY KEY,
            remote_id BLOB NOT NULL
          )`
  });
  yield* sql.onDialectOrElse({
    pg: () => sql`
          CREATE TABLE IF NOT EXISTS ${entriesTableSql} (
            store_id TEXT NOT NULL,
            sequence BIGINT NOT NULL,
            entry_id BYTEA NOT NULL,
            event TEXT NOT NULL,
            primary_key TEXT NOT NULL,
            payload BYTEA NOT NULL,
            PRIMARY KEY (store_id, sequence),
            UNIQUE (store_id, entry_id)
          )`,
    mysql: () => sql`
          CREATE TABLE IF NOT EXISTS ${entriesTableSql} (
            store_id VARCHAR(191) NOT NULL,
            sequence BIGINT NOT NULL,
            entry_id BINARY(16) NOT NULL,
            event TEXT NOT NULL,
            primary_key TEXT NOT NULL,
            payload BLOB NOT NULL,
            PRIMARY KEY (store_id, sequence),
            UNIQUE (store_id, entry_id)
          )`,
    mssql: () => sql`
          CREATE TABLE IF NOT EXISTS ${entriesTableSql} (
            store_id NVARCHAR(191) NOT NULL,
            sequence BIGINT NOT NULL,
            entry_id VARBINARY(16) NOT NULL,
            event NVARCHAR(MAX) NOT NULL,
            primary_key NVARCHAR(MAX) NOT NULL,
            payload VARBINARY(MAX) NOT NULL,
            PRIMARY KEY (store_id, sequence),
            UNIQUE (store_id, entry_id)
          )`,
    orElse: () => sql`
          CREATE TABLE IF NOT EXISTS ${entriesTableSql} (
            store_id TEXT NOT NULL,
            sequence INTEGER NOT NULL,
            entry_id BLOB NOT NULL,
            event TEXT NOT NULL,
            primary_key TEXT NOT NULL,
            payload BLOB NOT NULL,
            PRIMARY KEY (store_id, sequence),
            UNIQUE (store_id, entry_id)
          )`
  });
  yield* sql.onDialectOrElse({
    pg: () => sql`
          CREATE TABLE IF NOT EXISTS ${storesTableSql} (
            store_id TEXT PRIMARY KEY,
            next_sequence BIGINT NOT NULL
          )`,
    mysql: () => sql`
          CREATE TABLE IF NOT EXISTS ${storesTableSql} (
            store_id VARCHAR(191) PRIMARY KEY,
            next_sequence BIGINT NOT NULL
          )`,
    mssql: () => sql`
          CREATE TABLE IF NOT EXISTS ${storesTableSql} (
            store_id NVARCHAR(191) PRIMARY KEY,
            next_sequence BIGINT NOT NULL
          )`,
    orElse: () => sql`
          CREATE TABLE IF NOT EXISTS ${storesTableSql} (
            store_id TEXT PRIMARY KEY,
            next_sequence INTEGER NOT NULL
          )`
  });
  yield* sql.onDialectOrElse({
    pg: () => sql`
          CREATE TABLE IF NOT EXISTS ${sessionAuthBindingsTableSql} (
            public_key TEXT PRIMARY KEY,
            signing_public_key BYTEA NOT NULL
          )`,
    mysql: () => sql`
          CREATE TABLE IF NOT EXISTS ${sessionAuthBindingsTableSql} (
            public_key VARCHAR(191) PRIMARY KEY,
            signing_public_key BINARY(32) NOT NULL
          )`,
    mssql: () => sql`
          CREATE TABLE IF NOT EXISTS ${sessionAuthBindingsTableSql} (
            public_key NVARCHAR(191) PRIMARY KEY,
            signing_public_key VARBINARY(32) NOT NULL
          )`,
    orElse: () => sql`
          CREATE TABLE IF NOT EXISTS ${sessionAuthBindingsTableSql} (
            public_key TEXT PRIMARY KEY,
            signing_public_key BLOB NOT NULL
          )`
  });
  const selectRemoteId = sql`
      SELECT remote_id
      FROM ${remoteIdTableSql}
      WHERE singleton = 1
    `;
  const remoteId = yield* selectRemoteId.pipe(Effect.flatMap(rows => {
    const existing = rows[0];
    if (existing !== undefined) {
      return Effect.succeed(existing.remote_id);
    }
    const created = makeRemoteIdUnsafe();
    return sql`
          INSERT INTO ${remoteIdTableSql} (singleton, remote_id)
          VALUES (1, ${created})
        `.pipe(Effect.catchIf(isConstraintConflict, () => Effect.void), Effect.andThen(selectRemoteId), Effect.map(rows => rows[0]?.remote_id), Effect.map(persisted => persisted ?? created));
  }));
  const pubsubs = yield* RcMap.make({
    lookup: _storeId => Effect.acquireRelease(PubSub.unbounded(), PubSub.shutdown),
    idleTimeToLive: "5 minutes"
  });
  const ensureStore = storeId => sql.onDialectOrElse({
    pg: () => sql`
            INSERT INTO ${storesTableSql} (store_id, next_sequence)
            VALUES (${storeId}, 1)
            ON CONFLICT (store_id) DO NOTHING
          `,
    mysql: () => sql`
            INSERT INTO ${storesTableSql} (store_id, next_sequence)
            VALUES (${storeId}, 1)
            ON DUPLICATE KEY UPDATE store_id = store_id
          `,
    mssql: () => sql`
            MERGE ${storesTableSql} WITH (HOLDLOCK) AS target
            USING (SELECT ${storeId} AS store_id, 1 AS next_sequence) AS source
              ON target.store_id = source.store_id
            WHEN NOT MATCHED THEN
              INSERT (store_id, next_sequence)
              VALUES (source.store_id, source.next_sequence);
          `,
    orElse: () => sql`
            INSERT INTO ${storesTableSql} (store_id, next_sequence)
            VALUES (${storeId}, 1)
            ON CONFLICT DO NOTHING
          `
  });
  const lockStore = storeId => sql.onDialectOrElse({
    pg: () => sql`
            SELECT next_sequence
            FROM ${storesTableSql}
            WHERE store_id = ${storeId}
            FOR UPDATE
          `,
    mysql: () => sql`
            SELECT next_sequence
            FROM ${storesTableSql}
            WHERE store_id = ${storeId}
            FOR UPDATE
          `,
    mssql: () => sql`
            SELECT next_sequence
            FROM ${storesTableSql} WITH (UPDLOCK, HOLDLOCK)
            WHERE store_id = ${storeId}
          `,
    orElse: () => sql`
            UPDATE ${storesTableSql}
            SET next_sequence = next_sequence
            WHERE store_id = ${storeId}
            RETURNING next_sequence
          `
  }).pipe(Effect.flatMap(decodeStoreSequence));
  const setNextSequence = (storeId, nextSequence) => sql`
        UPDATE ${storesTableSql}
        SET next_sequence = ${nextSequence}
        WHERE store_id = ${storeId}
      `;
  const selectEntriesAfter = (storeId, startSequence) => sql`
        SELECT sequence, entry_id, event, primary_key, payload
        FROM ${entriesTableSql}
        WHERE store_id = ${storeId} AND sequence >= ${startSequence}
        ORDER BY sequence ASC
      `.pipe(Effect.flatMap(decodeRemoteEntries));
  const getSessionAuthBinding = publicKey => sql`
        SELECT public_key, signing_public_key
        FROM ${sessionAuthBindingsTableSql}
        WHERE public_key = ${publicKey}
      `.pipe(Effect.flatMap(decodeSessionAuthBindings), Effect.map(rows => {
    const row = rows[0];
    return row === undefined ? undefined : row.signing_public_key;
  }), Effect.orDie);
  return EventLogServerUnencrypted.Storage.of({
    getId: Effect.succeed(remoteId),
    getOrCreateSessionAuthBinding: Effect.fnUntraced(function* (publicKey, signingPublicKey) {
      const existing = yield* getSessionAuthBinding(publicKey);
      if (existing !== undefined) {
        return existing;
      }
      return yield* sql`
            INSERT INTO ${sessionAuthBindingsTableSql} (public_key, signing_public_key)
            VALUES (${publicKey}, ${signingPublicKey})
          `.pipe(Effect.as(signingPublicKey));
    }, sql.withTransaction, Effect.orDie, withTracerDisabled),
    entriesAfter: (storeId, entry) => sql`
          SELECT sequence, entry_id, event, primary_key, payload
          FROM ${entriesTableSql}
          WHERE store_id = ${storeId} AND entry_id >= ${entry.id}
          ORDER BY sequence ASC
        `.pipe(Effect.flatMap(decodeRemoteEntries), Effect.map(Arr.map(r => r.entry)), Effect.orDie, withTracerDisabled),
    write: Effect.fnUntraced(function* (storeId, entries) {
      if (entries.length === 0) {
        return [];
      }
      yield* ensureStore(storeId);
      const currentNextSequence = yield* lockStore(storeId);
      const committed = [];
      let rowsForInsert = [];
      for (let index = 0; index < entries.length; index++) {
        const entry = entries[index];
        const remoteSequence = currentNextSequence + index;
        committed.push(new RemoteEntry({
          remoteSequence,
          entry
        }, {
          disableChecks: true
        }));
        rowsForInsert.push({
          store_id: storeId,
          sequence: remoteSequence,
          entry_id: entry.id,
          event: entry.event,
          primary_key: entry.primaryKey,
          payload: entry.payload
        });
        if (rowsForInsert.length >= insertBatchSize) {
          yield* sql`INSERT INTO ${entriesTableSql} ${sql.insert(rowsForInsert)}`;
          rowsForInsert = [];
        }
      }
      if (rowsForInsert.length > 0) {
        yield* sql`INSERT INTO ${entriesTableSql} ${sql.insert(rowsForInsert)}`;
      }
      const nextSequence = currentNextSequence + entries.length;
      yield* setNextSequence(storeId, nextSequence);
      const pubsub = yield* RcMap.get(pubsubs, storeId);
      yield* PubSub.publishAll(pubsub, committed);
      return committed;
    }, Effect.scoped, sql.withTransaction, withTracerDisabled, Effect.orDie),
    changes: Effect.fnUntraced(function* ({
      storeId,
      startSequence,
      compactors
    }) {
      const pubsub = yield* RcMap.get(pubsubs, storeId);
      const subscription = yield* PubSub.subscribe(pubsub);
      const backlog = yield* EventLogServerUnencrypted.compactBacklog({
        compactors,
        remoteEntries: yield* selectEntriesAfter(storeId, startSequence)
      });
      let watermark = backlog.length > 0 ? backlog[backlog.length - 1].remoteSequence : startSequence - 1;
      return Stream.fromArray(backlog).pipe(Stream.concat(Stream.fromSubscription(subscription).pipe(Stream.filter(entry => entry.remoteSequence > watermark))));
    }, Effect.orDie, withTracerDisabled, Stream.unwrap),
    withTransaction: effect => sql.withTransaction(effect).pipe(Effect.catchIf(SqlError.isSqlError, Effect.die))
  });
}).pipe(withTracerDisabled);
/**
 * Provides unencrypted server `Storage` using the SQL-backed implementation.
 *
 * @category layers
 * @since 4.0.0
 */
export const layerStorage = options => Layer.effect(EventLogServerUnencrypted.Storage)(makeStorage(options));
const EntrySql = /*#__PURE__*/Schema.Struct({
  entry_id: EntryId,
  event: Schema.String,
  primary_key: Schema.String,
  payload: Schema.Uint8Array
});
const SqlNumber = /*#__PURE__*/Schema.Union([Schema.Number, Schema.NumberFromString]);
const RemoteEntrySql = /*#__PURE__*/Schema.Struct({
  ...EntrySql.fields,
  sequence: SqlNumber
});
const StoreSequenceSql = /*#__PURE__*/Schema.Struct({
  next_sequence: SqlNumber
});
const SessionAuthBindingSql = /*#__PURE__*/Schema.Struct({
  public_key: Schema.String,
  signing_public_key: Schema.Uint8Array
});
const decodeRemoteEntryRows = /*#__PURE__*/Schema.decodeUnknownEffect(/*#__PURE__*/Schema.mutable(/*#__PURE__*/Schema.Array(RemoteEntrySql)));
const decodeStoreSequenceRows = /*#__PURE__*/Schema.decodeUnknownEffect(/*#__PURE__*/Schema.Array(StoreSequenceSql));
const decodeSessionAuthBindingRows = /*#__PURE__*/Schema.decodeUnknownEffect(/*#__PURE__*/Schema.Array(SessionAuthBindingSql));
const toEntry = row => new Entry({
  id: row.entry_id,
  event: row.event,
  primaryKey: row.primary_key,
  payload: row.payload
}, {
  disableChecks: true
});
const toRemoteEntry = row => new RemoteEntry({
  remoteSequence: row.sequence,
  entry: toEntry(row)
}, {
  disableChecks: true
});
const decodeRemoteEntries = rows => decodeRemoteEntryRows(rows).pipe(Effect.map(rows => rows.map(toRemoteEntry)));
const decodeStoreSequence = rows => decodeStoreSequenceRows(rows).pipe(Effect.flatMap(rows => {
  const row = rows[0];
  if (row === undefined) {
    return Effect.die("SqlEventLogServerUnencrypted missing store sequence row");
  }
  return Effect.succeed(row.next_sequence);
}));
const decodeSessionAuthBindings = rows => decodeSessionAuthBindingRows(rows);
const withTracerDisabled = /*#__PURE__*/Effect.withTracerEnabled(false);
const isConstraintConflict = error => error.reason._tag === "ConstraintError" || error.reason._tag === "UniqueViolation";
//# sourceMappingURL=SqlEventLogServerUnencrypted.js.map