Add persistent EphemeralStore adaptors

README.md

@@ -157,7 +157,7 @@ The Rust workspace contains a minimal async WebSocket server (`loro-websocket-se
 
 ## Protocol Highlights
 
-- Magic bytes per CRDT: "%LOR" (Loro doc), "%EPH" (Loro ephemeral), "%YJS", "%YAW", …
+- Magic bytes per CRDT: "%LOR" (Loro doc), "%EPH" (Loro ephemeral), "%EPS" (persisted Loro ephemeral – tells the server to keep the latest state so new peers can load it immediately), "%YJS", "%YAW", …
 - Messages: JoinRequest/JoinResponseOk/JoinError, DocUpdate, DocUpdateFragmentHeader/Fragment, UpdateError, Leave
 - Limits: 256 KiB per message; large updates must be fragmented; default reassembly timeout 10s
 - Multi‑room: room ID is part of every message; one connection can join multiple rooms

packages/loro-adaptors/README.md

@@ -14,6 +14,7 @@ The websocket client (`loro-websocket`) speaks the binary wire protocol. These a
 
 - `LoroAdaptor`: wraps a `LoroDoc` and streams local updates to the connection; applies remote updates on receipt
 - `LoroEphemeralAdaptor`: wraps an `EphemeralStore` for transient presence/cursor data
+- `LoroPersistentStoreAdaptor`: wraps an `EphemeralStore` but marks updates as persisted so the server stores them for new peers
 - `EloLoroAdaptor`: wraps a `LoroDoc` and packages updates into %ELO containers with AES‑GCM; decrypts inbound containers and imports plaintext.
 
 ## Usage
@@ -23,6 +24,7 @@ import { LoroWebsocketClient } from "loro-websocket";
 import {
   LoroAdaptor,
   LoroEphemeralAdaptor,
+  LoroPersistentStoreAdaptor,
   EloLoroAdaptor,
 } from "loro-adaptors";
 import { LoroDoc, EphemeralStore } from "loro-crdt";
@@ -41,6 +43,14 @@ const eph = new EphemeralStore(30_000);
 const ephAdaptor = new LoroEphemeralAdaptor(eph);
 const roomEph = await client.join({ roomId: "demo", crdtAdaptor: ephAdaptor });
 
+// Persisted presence that should be available to late joiners
+const persistedStore = new EphemeralStore(30_000);
+const persistedAdaptor = new LoroPersistentStoreAdaptor(persistedStore);
+const roomPersisted = await client.join({
+  roomId: "demo-persisted",
+  crdtAdaptor: persistedAdaptor,
+});
+
 // %ELO (end‑to‑end encrypted Loro)
 const key = new Uint8Array(32);
 const elo = new EloLoroAdaptor({
@@ -55,13 +65,15 @@ doc.commit();
 // Cleanup
 await roomEph.destroy();
 await roomDoc.destroy();
+await roomPersisted.destroy();
 await secure.destroy();
 ```
 
 ## API
 
 - `new LoroAdaptor(doc?: LoroDoc, config?: { onImportError?, onUpdateError? })`
 - `new LoroEphemeralAdaptor(store?: EphemeralStore)`
+- `new LoroPersistentStoreAdaptor(store?: EphemeralStore)`
 - `new EloLoroAdaptor(docOrConfig: LoroDoc | { getPrivateKey, ivFactory?, onDecryptError?, onUpdateError? })`
   - `getPrivateKey: (keyId?) => Promise<{ keyId: string, key: CryptoKey | Uint8Array }>`
   - Optional `ivFactory()` for testing (12‑byte IV)

packages/loro-adaptors/src/adaptors.ts

@@ -1,4 +1,5 @@
 export * from "./loro-adaptor";
 export * from "./loro-ephemeral-adaptor";
+export * from "./loro-persistent-store-adaptor";
 export * from "./elo-adaptor";
 export * from "./flock-adaptor";

packages/loro-adaptors/src/loro-persistent-store-adaptor.ts

@@ -0,0 +1,86 @@
+import { EphemeralStore } from "loro-crdt";
+import { CrdtType, JoinResponseOk } from "loro-protocol";
+import type { CrdtAdaptorContext, CrdtDocAdaptor } from "./types";
+
+export class LoroPersistentStoreAdaptor implements CrdtDocAdaptor {
+  readonly crdtType = CrdtType.LoroEphemeralStorePersisted;
+
+  private store: EphemeralStore;
+  private ctx?: CrdtAdaptorContext;
+  private localUpdateUnsubscribe?: () => void;
+  private destroyed = false;
+
+  constructor(store?: EphemeralStore) {
+    this.store = store || new EphemeralStore();
+  }
+
+  waitForReachingServerVersion(): Promise<void> {
+    return Promise.resolve();
+  }
+
+  cmpVersion(_v: Uint8Array) {
+    return 0 as const;
+  }
+
+  handleJoinErr?: undefined;
+
+  getStore(): EphemeralStore {
+    return this.store;
+  }
+
+  setCtx(ctx: CrdtAdaptorContext): void {
+    this.ctx = ctx;
+    this.localUpdateUnsubscribe = this.store.subscribeLocalUpdates(updates => {
+      if (!this.destroyed && this.ctx) {
+        this.ctx.send([updates]);
+      }
+    });
+  }
+
+  getVersion(): Uint8Array {
+    return new Uint8Array();
+  }
+
+  getAlternativeVersion(_currentVersion: Uint8Array): Uint8Array | undefined {
+    return undefined;
+  }
+
+  async handleJoinOk(_res: JoinResponseOk): Promise<void> {
+    if (this.destroyed) return;
+    if (this.ctx) {
+      const allState = this.store.encodeAll();
+      if (allState.length > 0) {
+        this.ctx.send([allState]);
+      }
+    }
+  }
+
+  applyUpdate(updates: Uint8Array[]): void {
+    if (this.destroyed || !updates?.length) return;
+    for (const update of updates) {
+      if (update?.length > 0) {
+        try {
+          this.store.apply(update);
+        } catch (error) {
+          this.ctx?.onImportError(
+            error instanceof Error ? error : new Error(String(error)),
+            [update]
+          );
+        }
+      }
+    }
+  }
+
+  destroy(): void {
+    if (this.destroyed) return;
+    this.destroyed = true;
+
+    if (this.localUpdateUnsubscribe) {
+      this.localUpdateUnsubscribe();
+      this.localUpdateUnsubscribe = undefined;
+    }
+
+    this.store.destroy();
+    this.ctx = undefined;
+  }
+}

packages/loro-adaptors/src/server/index.ts

@@ -1,5 +1,6 @@
 export * from "./server-registry";
 export * from "./server-loro-adaptor";
 export * from "./server-loro-ephemeral-adaptor";
+export * from "./server-loro-persistent-store-adaptor";
 export * from "./server-yjs-awareness-adaptor";
 export * from "./server-flock-adaptor";

packages/loro-adaptors/src/server/server-loro-persistent-store-adaptor.ts

@@ -0,0 +1,136 @@
+import { EphemeralStore } from "loro-crdt";
+import {
+  CrdtType,
+  Permission,
+  MessageType,
+  JoinResponseOk,
+  UpdateError,
+  UpdateErrorCode,
+} from "loro-protocol";
+import type { CrdtServerAdaptor } from "../types";
+
+export interface LoroPersistentStoreServerAdaptorConfig {
+  timeout?: number;
+}
+
+export class LoroPersistentStoreServerAdaptor implements CrdtServerAdaptor {
+  readonly crdtType = CrdtType.LoroEphemeralStorePersisted;
+  private readonly timeout: number;
+
+  constructor(config: LoroPersistentStoreServerAdaptorConfig = {}) {
+    this.timeout = config.timeout ?? 10_000;
+  }
+
+  createEmpty(): Uint8Array {
+    const store = new EphemeralStore(this.timeout);
+    try {
+      return store.encodeAll();
+    } finally {
+      store.inner.free();
+    }
+  }
+
+  handleJoinRequest(
+    documentData: Uint8Array,
+    _clientVersion: Uint8Array,
+    permission: Permission
+  ): {
+    response: JoinResponseOk;
+    updates?: Uint8Array[];
+  } {
+    const response: JoinResponseOk = {
+      type: MessageType.JoinResponseOk,
+      crdt: this.crdtType,
+      roomId: "",
+      permission,
+      version: new Uint8Array(),
+    };
+
+    const updates = documentData.length > 0 ? [documentData] : undefined;
+    return { response, updates };
+  }
+
+  applyUpdates(
+    documentData: Uint8Array,
+    updates: Uint8Array[],
+    permission: Permission
+  ): {
+    success: boolean;
+    newDocumentData?: Uint8Array;
+    error?: UpdateError;
+    broadcastUpdates?: Uint8Array[];
+  } {
+    if (permission === "read") {
+      return {
+        success: false,
+        error: {
+          type: MessageType.UpdateError,
+          crdt: this.crdtType,
+          roomId: "",
+          code: UpdateErrorCode.PermissionDenied,
+          message: "Read-only permission, cannot apply updates",
+        },
+      };
+    }
+
+    const store = new EphemeralStore(this.timeout);
+    const broadcastUpdates: Uint8Array[] = [];
+
+    try {
+      if (documentData.length > 0) {
+        store.apply(documentData);
+      }
+      for (const update of updates) {
+        if (update.length > 0) {
+          store.apply(update);
+          broadcastUpdates.push(update);
+        }
+      }
+
+      const newDocumentData = store.encodeAll();
+
+      return {
+        success: true,
+        newDocumentData,
+        broadcastUpdates:
+          broadcastUpdates.length > 0 ? broadcastUpdates : undefined,
+      };
+    } catch (error) {
+      return {
+        success: false,
+        error: {
+          type: MessageType.UpdateError,
+          crdt: this.crdtType,
+          roomId: "",
+          code: UpdateErrorCode.InvalidUpdate,
+          message: error instanceof Error ? error.message : "Invalid update",
+        },
+      };
+    } finally {
+      store.destroy();
+      store.inner.free();
+    }
+  }
+
+  getVersion(_documentData: Uint8Array): Uint8Array {
+    return new Uint8Array();
+  }
+
+  merge(documents: Uint8Array[]): Uint8Array {
+    const store = new EphemeralStore(this.timeout);
+    for (const data of documents) {
+      if (data.length > 0) {
+        store.apply(data);
+      }
+    }
+    try {
+      return store.encodeAll();
+    } finally {
+      store.destroy();
+      store.inner.free();
+    }
+  }
+}
+
+export const loroPersistentStoreServerAdaptor =
+  new LoroPersistentStoreServerAdaptor();

packages/loro-adaptors/tests/adaptors.test.ts

@@ -1,6 +1,10 @@
 import { describe, it, expect, beforeEach, vi } from "vitest";
 import { LoroDoc, EphemeralStore, VersionVector } from "loro-crdt";
-import { LoroAdaptor, LoroEphemeralAdaptor } from "../src/adaptors";
+import {
+  LoroAdaptor,
+  LoroEphemeralAdaptor,
+  LoroPersistentStoreAdaptor,
+} from "../src/adaptors";
 import { CrdtType, JoinResponseOk, MessageType } from "loro-protocol";
 
 describe("LoroAdaptor", () => {
@@ -125,6 +129,46 @@ describe("LoroEphemeralAdaptor", () => {
   });
 });
 
+describe("LoroPersistentStoreAdaptor", () => {
+  let adaptor: LoroPersistentStoreAdaptor;
+  let store: EphemeralStore;
+
+  beforeEach(() => {
+    store = new EphemeralStore();
+    adaptor = new LoroPersistentStoreAdaptor(store);
+  });
+
+  it("should have correct CRDT type", () => {
+    expect(adaptor.crdtType).toBe(CrdtType.LoroEphemeralStorePersisted);
+  });
+
+  it("should return the underlying EphemeralStore", () => {
+    expect(adaptor.getStore()).toBe(store);
+  });
+
+  it("should send full state on join", async () => {
+    const mockSend = vi.fn();
+    adaptor.setCtx({
+      send: mockSend,
+      onJoinFailed: vi.fn(),
+      onImportError: vi.fn(),
+    });
+
+    store.set("cursor", { x: 1 });
+    const joinOk: JoinResponseOk = {
+      type: MessageType.JoinResponseOk,
+      crdt: CrdtType.LoroEphemeralStorePersisted,
+      roomId: "room",
+      permission: "write",
+      version: new Uint8Array(),
+    };
+
+    await adaptor.handleJoinOk(joinOk);
+
+    expect(mockSend).toHaveBeenCalledWith([expect.any(Uint8Array)]);
+  });
+});
+
 describe("Utility functions", () => {
   it("should create LoroAdaptor with new LoroDoc", () => {
     const adaptor = new LoroAdaptor();

packages/loro-protocol/src/encoding.ts

@@ -22,6 +22,7 @@ import {
 const validCrdtTypes = [
   CrdtType.Loro,
   CrdtType.LoroEphemeralStore,
+  CrdtType.LoroEphemeralStorePersisted,
   CrdtType.Yjs,
   CrdtType.YjsAwareness,
   CrdtType.Elo,
@@ -35,6 +36,7 @@ function isCrdtType(x: unknown): x is CrdtType {
     typeof x === "string" &&
     (x === CrdtType.Loro ||
       x === CrdtType.LoroEphemeralStore ||
+      x === CrdtType.LoroEphemeralStorePersisted ||
       x === CrdtType.Yjs ||
       x === CrdtType.YjsAwareness ||
       x === CrdtType.Flock ||

packages/loro-protocol/src/protocol.ts

@@ -5,6 +5,7 @@ export const MAX_MESSAGE_SIZE = 256 * 1024;
 export const CrdtType = {
   Loro: "%LOR",
   LoroEphemeralStore: "%EPH",
+  LoroEphemeralStorePersisted: "%EPS",
   Yjs: "%YJS",
   YjsAwareness: "%YAW",
   Elo: "%ELO",
@@ -13,6 +14,7 @@ export const CrdtType = {
 export const CrdtTypeId = {
   Loro: "loro",
   LoroEphemeralStore: "eph",
+  LoroEphemeralStorePersisted: "eps",
   Yjs: "yjs",
   YjsAwareness: "yaw",
   Elo: "elo",
@@ -23,6 +25,7 @@ export type CrdtId = (typeof CrdtTypeId)[keyof typeof CrdtTypeId];
 export const MagicBytesToCrdtId = {
   "%LOR": "loro",
   "%EPH": "eph",
+  "%EPS": "eps",
   "%YJS": "yjs",
   "%YAW": "yaw",
   "%ELO": "elo",