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

/**
 * Plaintext server implementation for the event-log remote protocol.
 *
 * This module accepts unencrypted event batches from remote clients, runs the
 * registered event handlers, stores journal entries, and streams backlog plus
 * live changes through the shared `EventLogServer` RPC protocol. It is intended
 * for trusted deployments, local development, and tests where event data does
 * not need a server-side encryption layer. The module also provides the
 * services and layers needed to authorize requests, map stores, persist entries,
 * and install the plaintext server.
 *
 * @since 4.0.0
 */
import * as Arr from "../../Array.js";
import * as Context from "../../Context.js";
import * as Data from "../../Data.js";
import * as Effect from "../../Effect.js";
import * as Layer from "../../Layer.js";
import * as Option from "../../Option.js";
import * as PubSub from "../../PubSub.js";
import * as RcMap from "../../RcMap.js";
import * as Redacted from "../../Redacted.js";
import * as Schema from "../../Schema.js";
import * as Semaphore from "../../Semaphore.js";
import * as Stream from "../../Stream.js";
import * as RpcServer from "../rpc/RpcServer.js";
import * as EventJournal from "./EventJournal.js";
import { Entry, makeEntryIdUnsafe, makeRemoteIdUnsafe, RemoteEntry } from "./EventJournal.js";
import * as EventLog from "./EventLog.js";
import { ChangesRpc, EventLogProtocolError, EventLogRemoteRpcs, WriteEntriesUnencrypted } from "./EventLogMessage.js";
import * as EventLogServer from "./EventLogServer.js";
/**
 * Service that writes plaintext event-log entries directly to
 * unencrypted storage through registered event handlers.
 *
 * **When to use**
 *
 * Use to access or provide the server service that handles plaintext
 * event-log writes.
 *
 * @category services
 * @since 4.0.0
 */
export class EventLogServerUnencrypted extends /*#__PURE__*/Context.Service()("effect/eventlog/EventLogServerUnencrypted") {}
/**
 * Creates a typed server-side write function for events in the supplied
 * `EventLogSchema`.
 *
 * @category EventLogServerUnencrypted
 * @since 4.0.0
 */
export const makeWrite = schema => EventLogServerUnencrypted.useSync(_ => _.makeWrite(schema));
/**
 * Provides RPC handlers for the unencrypted event-log server.
 *
 * **Details**
 *
 * Incoming plaintext entries are authorized, mapped to a server store, checked
 * for conflicts, run through registered handlers, and persisted; change streams
 * include compacted backlog entries when compactors are registered.
 *
 * @category layers
 * @since 4.0.0
 */
export const layerRpcHandlers = /*#__PURE__*/Layer.unwrap(/*#__PURE__*/Effect.gen(function* () {
  const storage = yield* Storage;
  const mapping = yield* StoreMapping;
  const auth = yield* EventLogServerAuthorization;
  const registry = yield* EventLog.Registry;
  const handler = yield* makeServerHandler;
  const remoteId = yield* storage.getId;
  const processEntries = Effect.fnUntraced(function* (options) {
    const entries = Arr.sort(options.entries, Entry.Order);
    let history = yield* storage.entriesAfter(options.storeId, entries[0]);
    const persistedEntries = Arr.empty();
    for (const entry of entries) {
      const [duplicate, conflicts, newHistory] = toConflicts(history, entry);
      if (duplicate) continue;
      history = newHistory;
      yield* handler({
        publicKey: options.publicKey,
        storeId: options.storeId,
        entry,
        conflicts
      });
      persistedEntries.push(entry);
    }
    yield* storage.write(options.storeId, persistedEntries);
  }, storage.withTransaction);
  return EventLogServer.layerRpcHandlers({
    remoteId,
    getOrCreateSessionAuthBinding: (publicKey, signingPublicKey) => storage.getOrCreateSessionAuthBinding(publicKey, signingPublicKey),
    onWrite: Effect.fnUntraced(function* (data) {
      const request = yield* WriteEntriesUnencrypted.decode(data).pipe(Effect.mapError(_ => new EventLogProtocolError({
        requestTag: "WriteEntries",
        publicKey: undefined,
        code: "InternalServerError",
        message: "Decoding failure"
      })));
      if (!Arr.isReadonlyArrayNonEmpty(request.entries)) return;
      const resolvedStoreId = yield* mapping.resolve({
        publicKey: request.publicKey,
        storeId: request.storeId
      }).pipe(Effect.mapError(_ => new EventLogProtocolError({
        requestTag: "WriteEntries",
        publicKey: request.publicKey,
        storeId: request.storeId,
        code: "Unauthorized",
        message: _.message
      })));
      yield* auth.authorizeWrite({
        publicKey: request.publicKey,
        storeId: resolvedStoreId,
        entries: request.entries
      }).pipe(Effect.mapError(_ => new EventLogProtocolError({
        requestTag: "WriteEntries",
        publicKey: request.publicKey,
        storeId: request.storeId,
        code: "Unauthorized",
        message: _.message
      })));
      yield* processEntries({
        publicKey: request.publicKey,
        storeId: resolvedStoreId,
        entries: request.entries
      }).pipe(Effect.catchCause(_ => Effect.fail(new EventLogProtocolError({
        requestTag: "WriteEntries",
        publicKey: request.publicKey,
        code: "InternalServerError",
        message: "Persistence failure"
      }))), Effect.provideService(EventLog.Identity, makeClientIdentity(request.publicKey)));
    }),
    changes: Effect.fnUntraced(function* (request) {
      const storeId = yield* mapping.resolve({
        publicKey: request.publicKey,
        storeId: request.storeId
      });
      yield* auth.authorizeRead({
        publicKey: request.publicKey,
        storeId
      });
      return storage.changes({
        storeId,
        startSequence: request.startSequence,
        compactors: registry.compactors
      }).pipe(Stream.mapArrayEffect(entries => Effect.map(ChangesRpc.encodeUnencrypted(entries), Arr.of)));
    }, Stream.unwrap)
  });
}));
/**
 * Error raised by unencrypted server storage and store mapping operations.
 *
 * @category errors
 * @since 4.0.0
 */
export class EventLogServerStoreError extends /*#__PURE__*/Data.TaggedError("EventLogServerStoreError") {}
/**
 * Error raised when unencrypted server authorization rejects an identity or store
 * operation.
 *
 * @category errors
 * @since 4.0.0
 */
export class EventLogServerAuthError extends /*#__PURE__*/Data.TaggedError("EventLogServerAuthError") {}
/**
 * Service that validates unencrypted event-log server
 * write access, read access, and identities.
 *
 * **When to use**
 *
 * Use to provide authorization checks for plaintext event-log writes, reads,
 * and identity authentication.
 *
 * @category services
 * @since 4.0.0
 */
export class EventLogServerAuthorization extends /*#__PURE__*/Context.Service()("effect/eventlog/EventLogServerUnencrypted/EventLogServerAuthorization") {}
/**
 * Service that resolves client-requested store ids to server store ids and checks
 * whether a store exists.
 *
 * **When to use**
 *
 * Use to map client-visible store identifiers to server storage identifiers
 * before authorizing or serving unencrypted event-log requests.
 *
 * @category services
 * @since 4.0.0
 */
export class StoreMapping extends /*#__PURE__*/Context.Service()("effect/eventlog/EventLogServerUnencrypted/StoreMapping") {}
const toStoreNotFoundError = options => new EventLogServerStoreError({
  reason: "NotFound",
  publicKey: options.publicKey,
  storeId: options.storeId,
  message: options.publicKey === undefined ? `No provisioned store found for store id: ${options.storeId}` : `No provisioned store found for public key: ${options.publicKey} and store id: ${options.storeId}`
});
/**
 * Provides a `StoreMapping` that accepts only one configured store id and fails
 * all other store ids as not found.
 *
 * @category store
 * @since 4.0.0
 */
export const layerStoreMappingStatic = options => Layer.succeed(StoreMapping, {
  resolve(request) {
    if (request.storeId === options.storeId) {
      return Effect.succeed(options.storeId);
    }
    return Effect.fail(toStoreNotFoundError(request));
  },
  hasStore: ({
    storeId
  }) => Effect.succeed(storeId === options.storeId)
});
/**
 * Defines the backing store service used by the unencrypted event-log server.
 *
 * **When to use**
 *
 * Use to provide durable event-log persistence for an unencrypted event-log
 * server layer.
 *
 * **Details**
 *
 * It provides the server remote id, stores session authentication bindings,
 * allocates remote sequence numbers, persists entries, streams changes, and
 * exposes a transaction boundary.
 *
 * @category storage
 * @since 4.0.0
 */
export class Storage extends /*#__PURE__*/Context.Service()("effect/eventlog/EventLogServerUnencrypted/Storage") {}
const makeClientIdentity = publicKey => ({
  publicKey,
  privateKey: constEmptyPrivateKey
});
const constEmptyPrivateKey = /*#__PURE__*/Redacted.make(/*#__PURE__*/new Uint8Array(32));
const makeServerWriteIdentityPublicKey = storeId => `effect-eventlog-server-write:${storeId}`;
const entriesAfter = (journal, startSequence) => journal.filter(entry => entry.remoteSequence >= startSequence);
const toConflicts = (history, originEntry) => {
  let duplicate = false;
  for (let i = 0; i < history.length; i++) {
    const entry = history[i];
    if (entry.createdAtMillis < originEntry.createdAtMillis) {
      continue;
    } else if (entry.idString === originEntry.idString) {
      duplicate = true;
      continue;
    }
    const newHistory = history.slice(i);
    let conflicts = [];
    for (let j = 0; j < newHistory.length; j++) {
      const scannedEntry = history[j];
      if (scannedEntry.event === originEntry.event && scannedEntry.primaryKey === originEntry.primaryKey) {
        conflicts.push(scannedEntry);
      }
    }
    return [duplicate, conflicts, newHistory];
  }
  return [duplicate, [], []];
};
const representativeSequences = options => {
  if (options.compactedCount === 0) {
    return [];
  }
  if (options.compactedCount > options.remoteEntries.length) {
    return undefined;
  }
  const maxSequence = options.remoteEntries[options.remoteEntries.length - 1].remoteSequence;
  if (options.compactedCount === 1) {
    return [maxSequence];
  }
  const selected = options.remoteEntries.slice(0, options.compactedCount - 1).map(entry => entry.remoteSequence);
  selected.push(maxSequence);
  for (let i = 1; i < selected.length; i++) {
    if (selected[i] <= selected[i - 1]) {
      return undefined;
    }
  }
  return selected;
};
const toCompactedRemoteEntries = options => {
  const sequences = representativeSequences({
    remoteEntries: options.remoteEntries,
    compactedCount: options.compacted.length
  });
  if (sequences === undefined) {
    return undefined;
  }
  return options.compacted.map((entry, index) => new RemoteEntry({
    remoteSequence: sequences[index],
    entry
  }, {
    disableChecks: true
  }));
};
/**
 * Runs the registered compactors over a backlog of remote entries.
 *
 * **When to use**
 *
 * Use to reduce stored remote entries before replaying them to an unencrypted
 * event-log client.
 *
 * **Details**
 *
 * Contiguous entries handled by the same compactor may be replaced with compacted
 * entries when the replacement count can be mapped back to increasing remote
 * sequence numbers; otherwise the original entries are kept.
 *
 * @category compaction
 * @since 4.0.0
 */
export const compactBacklog = /*#__PURE__*/Effect.fnUntraced(function* (options) {
  if (options.compactors.size === 0 || options.remoteEntries.length === 0) {
    return options.remoteEntries;
  }
  const compactedRemoteEntries = [];
  let index = 0;
  while (index < options.remoteEntries.length) {
    const remoteEntry = options.remoteEntries[index];
    const compactor = options.compactors.get(remoteEntry.entry.event);
    if (compactor === undefined) {
      compactedRemoteEntries.push(remoteEntry);
      index++;
      continue;
    }
    const entries = [remoteEntry.entry];
    const remoteGroup = [remoteEntry];
    const compacted = [];
    index++;
    while (index < options.remoteEntries.length) {
      const nextRemoteEntry = options.remoteEntries[index];
      const nextCompactor = options.compactors.get(nextRemoteEntry.entry.event);
      if (nextCompactor !== compactor) {
        break;
      }
      entries.push(nextRemoteEntry.entry);
      remoteGroup.push(nextRemoteEntry);
      index++;
    }
    yield* compactor.effect({
      entries,
      write(entry) {
        return Effect.sync(() => {
          compacted.push(entry);
        });
      }
    }).pipe(Effect.orDie);
    const projected = toCompactedRemoteEntries({
      compacted,
      remoteEntries: remoteGroup
    });
    if (projected === undefined) {
      compactedRemoteEntries.push(...remoteGroup);
      continue;
    }
    compactedRemoteEntries.push(...projected);
  }
  return compactedRemoteEntries;
});
/**
 * Creates an in-memory unencrypted server `Storage`.
 *
 * **Details**
 *
 * The implementation keeps per-store journals and session authentication bindings
 * in memory, publishes live changes, and serializes transactions with a
 * semaphore.
 *
 * @category storage
 * @since 4.0.0
 */
export const makeStorageMemory = /*#__PURE__*/Effect.gen(function* () {
  const knownIds = new Map();
  const journals = new Map();
  const sessionAuthBindings = new Map();
  const remoteId = makeRemoteIdUnsafe();
  const ensureKnownIds = storeId => {
    let storeKnownIds = knownIds.get(storeId);
    if (storeKnownIds) return storeKnownIds;
    storeKnownIds = new Map();
    knownIds.set(storeId, storeKnownIds);
    return storeKnownIds;
  };
  const ensureJournal = storeId => {
    let journal = journals.get(storeId);
    if (journal) return journal;
    journal = [];
    journals.set(storeId, journal);
    return journal;
  };
  const pubsubs = yield* RcMap.make({
    lookup: _storeId => Effect.acquireRelease(PubSub.unbounded(), PubSub.shutdown),
    idleTimeToLive: 60000
  });
  const write = Effect.fnUntraced(function* (storeId, entries) {
    const sequenceNumbers = [];
    const committed = [];
    const storeKnownIds = ensureKnownIds(storeId);
    const journal = ensureJournal(storeId);
    let lastSequenceNumber = Arr.last(journal).pipe(Option.map(entry => entry.remoteSequence), Option.getOrElse(() => 0));
    if (entries.some(entry => storeKnownIds.has(entry.idString))) {
      return yield* Effect.die("Duplicate entries");
    }
    for (const entry of entries) {
      const remoteEntry = new RemoteEntry({
        remoteSequence: ++lastSequenceNumber,
        entry
      }, {
        disableChecks: true
      });
      sequenceNumbers.push(remoteEntry.remoteSequence);
      committed.push(remoteEntry);
      journal.push(remoteEntry);
      storeKnownIds.set(entry.idString, remoteEntry.remoteSequence);
    }
    const pubsub = yield* RcMap.get(pubsubs, storeId);
    yield* PubSub.publishAll(pubsub, committed);
    return committed;
  }, Effect.scoped);
  const transactionSemaphore = yield* Semaphore.make(1);
  return Storage.of({
    getId: Effect.succeed(remoteId),
    getOrCreateSessionAuthBinding: (publicKey, signingPublicKey) => Effect.sync(() => {
      const existing = sessionAuthBindings.get(publicKey);
      if (existing) return existing;
      sessionAuthBindings.set(publicKey, signingPublicKey);
      return signingPublicKey;
    }),
    entriesAfter: (storeId, entry) => Effect.sync(() => {
      const journal = ensureJournal(storeId);
      return journal.filter(e => Entry.Order(e.entry, entry) >= 0).map(e => e.entry);
    }),
    write,
    changes: Effect.fnUntraced(function* ({
      storeId,
      startSequence,
      compactors
    }) {
      const pubsub = yield* RcMap.get(pubsubs, storeId);
      const subscription = yield* PubSub.subscribe(pubsub);
      const backlog = yield* compactBacklog({
        remoteEntries: entriesAfter(ensureJournal(storeId), startSequence),
        compactors
      });
      const replayedUpTo = 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 > replayedUpTo))));
    }, Stream.unwrap),
    withTransaction: transactionSemaphore.withPermits(1)
  });
});
/**
 * Provides unencrypted server `Storage` using the in-memory implementation.
 *
 * @category storage
 * @since 4.0.0
 */
export const layerStorageMemory = /*#__PURE__*/Layer.effect(Storage)(makeStorageMemory);
/**
 * Creates the `EventLogServerUnencrypted` service from the configured storage and
 * registered event handlers.
 *
 * **When to use**
 *
 * Use when you need the unencrypted event-log server service from provided
 * `Storage` and an event-log `Registry`.
 *
 * **Details**
 *
 * The constructed service exposes `makeWrite`, which builds a typed server-side
 * write function from an `EventLogSchema`. Each write encodes the payload with
 * the event schema, runs the registered handler, and persists the generated
 * entry inside `Storage.withTransaction`.
 *
 * **Gotchas**
 *
 * The write function dies if the requested event tag is not present in the
 * schema passed to `makeWrite`; it does not report that case as a typed failure.
 *
 * @see {@link makeWrite} for the accessor that retrieves the typed server-side write function from the service environment
 * @see {@link layerServer} for the layer form that provides this service together with an event-log `Registry`
 *
 * @category constructors
 * @since 4.0.0
 */
export const make = /*#__PURE__*/Effect.gen(function* () {
  const storage = yield* Storage;
  const handler = yield* makeServerHandler;
  return EventLogServerUnencrypted.of({
    makeWrite(schema) {
      const events = new Map();
      for (const group of schema.groups) {
        for (const [tag, event] of Object.entries(group.events)) {
          events.set(tag, event);
        }
      }
      return Effect.fnUntraced(function* (options) {
        const publicKey = makeServerWriteIdentityPublicKey(options.storeId);
        const schemaEvent = events.get(options.event);
        if (schemaEvent === undefined) {
          return yield* Effect.die(`Event schema not found for: "${options.event}"`);
        }
        const entry = new EventJournal.Entry({
          id: makeEntryIdUnsafe(),
          event: options.event,
          primaryKey: schemaEvent.primaryKey(options.payload),
          payload: yield* Schema.encodeUnknownEffect(schemaEvent.payloadMsgPack)(options.payload).pipe(Effect.mapError(_ => new EventLogServerStoreError({
            reason: "PersistenceFailure",
            publicKey: publicKey,
            storeId: options.storeId,
            message: "Failed to encode event"
          })))
        }, {
          disableChecks: true
        });
        const result = yield* handler({
          publicKey,
          storeId: options.storeId,
          entry,
          conflicts: []
        }).pipe(Effect.provideService(EventLog.Identity, makeClientIdentity(publicKey)));
        yield* storage.write(options.storeId, [entry]);
        return result;
      }, storage.withTransaction);
    }
  });
});
/**
 * Provides `EventLogServerUnencrypted` and an event-log `Registry` using the
 * configured unencrypted server `Storage`.
 *
 * **When to use**
 *
 * Use to provide the unencrypted event-log server service together with the
 * registry needed by event handlers.
 *
 * @category layers
 * @since 4.0.0
 */
export const layerServer = /*#__PURE__*/Layer.effect(EventLogServerUnencrypted, make).pipe(/*#__PURE__*/Layer.provideMerge(EventLog.layerRegistry));
/**
 * Builds a full unencrypted event-log RPC server for the supplied schema and
 * event-group handler layer.
 *
 * **When to use**
 *
 * Use when you need the full unencrypted event-log RPC server layer with
 * storage, authorization, RPC protocol, and event-group handler dependencies
 * supplied externally.
 *
 * **Details**
 *
 * The layer installs `EventLogRemoteRpcs`, wires `layerRpcHandlers`, registers
 * the supplied event-group handler layer, and provides `layerServer`, leaving
 * only the required infrastructure services in the environment.
 *
 * **Gotchas**
 *
 * Entries are persisted and streamed in plaintext. Protect the backing
 * `Storage` with the surrounding infrastructure, and use durable storage that
 * preserves session authentication bindings when the server must survive
 * restarts.
 *
 * @see {@link layerNoRpcServer} for installing the same unencrypted handlers when an `RpcServer.Protocol` is provided elsewhere
 * @see {@link layerRpcHandlers} for wiring the unencrypted RPC handlers directly
 * @see {@link layerServer} for constructing the server service and event-log registry without RPC handlers
 *
 * @category layers
 * @since 4.0.0
 */
export const layer = (_schema, layer) => RpcServer.layer(EventLogRemoteRpcs).pipe(Layer.provide(layerRpcHandlers), Layer.provide(layer), Layer.provide(layerServer));
/**
 * Builds the unencrypted event-log server handlers without installing an
 * `RpcServer.Protocol` implementation.
 *
 * @category layers
 * @since 4.0.0
 */
export const layerNoRpcServer = (_schema, layer) => layerRpcHandlers.pipe(Layer.merge(layer), Layer.provide(layerServer));
const makeServerHandler = /*#__PURE__*/Effect.gen(function* () {
  const registry = yield* EventLog.Registry;
  return Effect.fnUntraced(function* (options) {
    const handler = registry.handlers.get(options.entry.event);
    if (handler === undefined) {
      return yield* Effect.logDebug(`Event handler not found for: "${options.entry.event}"`);
    }
    const decodePayload = Schema.decodeUnknownEffect(handler.event.payloadMsgPack);
    const decodedConflicts = [];
    for (const conflict of options.conflicts) {
      decodedConflicts.push({
        entry: conflict,
        payload: yield* decodePayload(conflict.payload).pipe(Effect.updateContext(input => Context.merge(handler.context, input)))
      });
    }
    const payloadEffect = "payload" in options ? Effect.succeed(options.payload) : decodePayload(options.entry.payload);
    return yield* payloadEffect.pipe(Effect.mapError(_ => new EventLogServerStoreError({
      reason: "PersistenceFailure",
      publicKey: options.publicKey,
      storeId: options.storeId,
      message: "Failed to decode event"
    })), Effect.flatMap(payload => handler.handler({
      storeId: options.storeId,
      payload,
      entry: options.entry,
      conflicts: decodedConflicts
    })), Effect.updateContext(input => Context.merge(handler.context, input)));
  });
});
//# sourceMappingURL=EventLogServerUnencrypted.js.map