Skip to content

Commit

Permalink
make offline persist/refresh work
Browse files Browse the repository at this point in the history
  • Loading branch information
@grgbkr
grgbkr committed Dec 16, 2023
1 parent bf95cd6 commit 429cdd6
Show file tree
Hide file tree
Showing 3 changed files with 32 additions and 104 deletions.
3 changes: 1 addition & 2 deletions examples/tiptap/src/components/TextEditor.tsx
View file
Original file line number Diff line number Diff line change
Expand Up @@ -93,15 +93,14 @@ export function Editor({ roomID }: { roomID: string }) {
useEffect(() => {
console.log("creating new reflect instance");

const userID = "anon";
const userID = nanoid();

const reflect = new Reflect({
server,
userID,
roomID,
auth: userID,
mutators: yjsMutators,
kvStore: "idb",
});

const yDoc = new Y.Doc();
Expand Down
48 changes: 4 additions & 44 deletions packages/reflect-yjs/src/mutators.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ export function yjsProviderKeyPrefix(name: string): string {
return `'yjs/provider/${name}/`;
}

function yjsProviderClientUpdateKeyPrefix(name: string): string {
export function yjsProviderClientUpdateKeyPrefix(name: string): string {
return `${yjsProviderKeyPrefix(name)}/client/`;
}

Expand Down Expand Up @@ -92,19 +92,6 @@ function setClientUpdate(name: string, update: string, tx: WriteTransaction) {
return tx.set(yjsProviderClientUpdateKey(name, uuidv4()), update);
}

export function getClientUpdates(
name: string,
tx: ReadTransaction,
): Promise<string[]> {
const updatePrefix = yjsProviderClientUpdateKeyPrefix(name);
return tx
.scan({
prefix: updatePrefix,
})
.values()
.toArray() as Promise<string[]>;
}

const AVG_CHUNK_SIZE_B = 1024;
const MIN_CHUNK_SIZE_B = 256;
const MAX_CHUNK_SIZE_B = 2048;
Expand Down Expand Up @@ -174,7 +161,9 @@ async function getServerUpdate(
name: string,
tx: ReadTransaction,
): Promise<Uint8Array | undefined> {
const updateMeta = await getServerUpdateMeta(name, tx);
const updateMeta = (await tx.get(yjsProviderServerUpdateMetaKey(name))) as
| undefined
| ChunkedUpdateMeta;
if (updateMeta === undefined) {
return undefined;
}
Expand All @@ -193,35 +182,6 @@ async function getServerUpdate(
return unchunk(chunksByHash, updateMeta.chunkHashes, updateMeta.length);
}

export async function getServerUpdateMeta(
name: string,
tx: ReadTransaction,
): Promise<ChunkedUpdateMeta | undefined> {
const updateMeta = (await tx.get(yjsProviderServerUpdateMetaKey(name))) as
| undefined
| ChunkedUpdateMeta;
return updateMeta;
}

export async function getServerUpdateChunkEntries(
name: string,
tx: ReadTransaction,
): Promise<[hash: string, value: string][]> {
const chunksPrefix = yjsProviderServerUpdateChunkKeyPrefix(name);
const chunks = await tx
.scan({
prefix: chunksPrefix,
})
.entries();
const chunksPrefixLength = chunksPrefix.length;
const entries: [hash: string, value: string][] = [];
for await (const [key, value] of chunks) {
const hash = key.substring(chunksPrefixLength, key.length);
entries.push([hash, value as string]);
}
return entries;
}

export function yjsAwarenessKey(
name: string,
reflectClientID: ClientID,
Expand Down
85 changes: 27 additions & 58 deletions packages/reflect-yjs/src/provider.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,21 +4,18 @@ import * as Y from 'yjs';
import {Awareness} from './awareness.js';
import type {ChunkedUpdateMeta, Mutators} from './mutators.js';
import {
getClientUpdates,
getServerUpdateChunkEntries,
getServerUpdateMeta,
yjsProviderKeyPrefix,
yjsProviderClientUpdateKeyPrefix,
yjsProviderServerUpdateChunkKeyPrefix,
yjsProviderServerUpdateKeyPrefix,
yjsProviderServerUpdateMetaKey,
} from './mutators.js';
import {unchunk} from './chunk.js';

export class Provider {
readonly #reflect: Reflect<Mutators>;
readonly #ydoc: Y.Doc;
#destroyed = false;
#awareness: Awareness | null = null;
#cancelWatch: () => void = () => {};
readonly #cancelWatch: () => void;

readonly name: string;
#serverUpdateMeta: ChunkedUpdateMeta | null = null;
Expand All @@ -32,56 +29,27 @@ export class Provider {
ydoc.on('updateV2', this.#handleUpdateV2);
ydoc.on('destroy', this.#handleDestroy);

void this.#init();
}

async #init(): Promise<void> {
const {name} = this;
const clientUpdateKeyPrefix = yjsProviderClientUpdateKeyPrefix(name);
const serverUpdateMetaKey = yjsProviderServerUpdateMetaKey(name);
const serverUpdateChunkKeyPrefix =
yjsProviderServerUpdateChunkKeyPrefix(name);

const {clientUpdates, serverUpdateMeta, serverUpdateChunkEntries} =
await this.#reflect.query(async tx => ({
clientUpdates: await getClientUpdates(name, tx),
serverUpdateMeta: await getServerUpdateMeta(name, tx),
serverUpdateChunkEntries: await getServerUpdateChunkEntries(name, tx),
}));
if (this.#destroyed) {
return;
}

this.#serverUpdateMeta = serverUpdateMeta ?? null;
this.#serverUpdateChunks = new Map(
serverUpdateChunkEntries.map(([hash, value]) => [
hash,
base64.toByteArray(value),
]),
);
if (this.#serverUpdateMeta) {
Y.applyUpdateV2(
this.#ydoc,
unchunk(
this.#serverUpdateChunks,
this.#serverUpdateMeta.chunkHashes,
this.#serverUpdateMeta.length,
),
this,
);
}
for (const clientUpdate of clientUpdates) {
Y.applyUpdateV2(this.#ydoc, base64.toByteArray(clientUpdate), this);
}

this.#cancelWatch = this.#reflect.experimentalWatch(
this.#cancelWatch = reflect.experimentalWatch(
diff => {
const newClientUpdates: Uint8Array[] = [];
let serverUpdateChange = false;
for (const diffOp of diff) {
const {key} = diffOp;
switch (diffOp.op) {
case 'add':
case 'change':
if (key === serverUpdateMetaKey) {
if (key.startsWith(clientUpdateKeyPrefix)) {
newClientUpdates.push(
base64.toByteArray(diffOp.newValue as string),
);
} else if (key === serverUpdateMetaKey) {
this.#serverUpdateMeta = diffOp.newValue as ChunkedUpdateMeta;
serverUpdateChange = true;
} else if (key.startsWith(serverUpdateChunkKeyPrefix)) {
this.#serverUpdateChunks.set(
key.substring(serverUpdateChunkKeyPrefix.length),
Expand All @@ -92,6 +60,7 @@ export class Provider {
case 'del':
if (key === serverUpdateMetaKey) {
this.#serverUpdateMeta = null;
serverUpdateChange = true;
} else if (key.startsWith(serverUpdateChunkKeyPrefix)) {
this.#serverUpdateChunks.delete(
key.substring(serverUpdateChunkKeyPrefix.length),
Expand All @@ -100,20 +69,21 @@ export class Provider {
break;
}
}
if (this.#serverUpdateMeta !== null) {
Y.applyUpdateV2(
this.#ydoc,
unchunk(
this.#serverUpdateChunks,
this.#serverUpdateMeta.chunkHashes,
this.#serverUpdateMeta.length,
),
this,
if (serverUpdateChange && this.#serverUpdateMeta !== null) {
const serverUpdate = unchunk(
this.#serverUpdateChunks,
this.#serverUpdateMeta.chunkHashes,
this.#serverUpdateMeta.length,
);
Y.applyUpdateV2(ydoc, serverUpdate, this);
}
for (const clientUpdate of newClientUpdates) {
Y.applyUpdateV2(ydoc, clientUpdate, this);
}
},
{
prefix: yjsProviderServerUpdateKeyPrefix(name),
prefix: yjsProviderKeyPrefix(name),
initialValuesInFirstDiff: true,
},
);
}
Expand All @@ -125,13 +95,13 @@ export class Provider {
return this.#awareness;
}

#handleUpdateV2 = async (update: Uint8Array, origin: unknown) => {
#handleUpdateV2 = async (updateV2: Uint8Array, origin: unknown) => {
if (origin === this) {
return;
}
await this.#reflect.mutate.updateYJS({
name: this.name,
update: base64.fromByteArray(update),
update: base64.fromByteArray(updateV2),
});
};

Expand All @@ -140,7 +110,6 @@ export class Provider {
};

destroy(): void {
this.#destroyed = true;
this.#cancelWatch();
this.#serverUpdateMeta = null;
this.#serverUpdateChunks.clear();
Expand Down

0 comments on commit 429cdd6

Please sign in to comment.