Real-time collaboration: Enable CRDT doc state map changes

packages/core-data/src/entities.js

@@ -13,7 +13,7 @@ import { __ } from '@wordpress/i18n';
 /**
  * Internal dependencies
  */
-import { getSyncManager } from './sync';
+import { LOCAL_EDITOR_ORIGIN, getSyncManager } from './sync';
 import {
 	applyPostChangesToCRDTDoc,
 	defaultApplyChangesToCRDTDoc,
@@ -280,6 +280,13 @@ export const prePersistPostType = (
 				...edits.meta,
 				...meta,
 			};
+
+			// Mark the doc as having been persisted.
+			getSyncManager()?.markPersisted(
+				objectType,
+				objectId,
+				LOCAL_EDITOR_ORIGIN
+			);
 		}
 	}
 

packages/sync/src/config.ts

@@ -25,6 +25,8 @@ export const CRDT_STATE_MAP_KEY = 'state';
 
 // Y.Map keys for the state map.
 export const CRDT_STATE_VERSION_KEY = 'version';
+export const CRDT_STATE_PERSISTED_AT_KEY = 'persistedAt';
+export const CRDT_STATE_PERSISTED_BY_KEY = 'persistedBy';
 
 /**
  * Origin string for CRDT document changes originating from the local editor.

packages/sync/src/manager.ts

@@ -9,6 +9,8 @@ import * as Y from 'yjs';
 import {
 	CRDT_RECORD_MAP_KEY as RECORD_KEY,
 	LOCAL_SYNC_MANAGER_ORIGIN,
+	CRDT_STATE_PERSISTED_AT_KEY as PERSISTED_AT_KEY,
+	CRDT_STATE_PERSISTED_BY_KEY as PERSISTED_BY_KEY,
 } from './config';
 import { createPersistedCRDTDoc, getPersistedCrdtDoc } from './persistence';
 import { getProviderCreators } from './providers';
@@ -305,11 +307,49 @@ export function createSyncManager(): SyncManager {
 		return createPersistedCRDTDoc( entityState.ydoc );
 	}
 
+	/**
+	 * Update the last persisted timestamp, and the user who last persisted the document in the CRDT document state map.
+	 *
+	 * This can be used to send notifications to other peers about when the document
+	 * was last updated, and by whom.
+	 *
+	 * @param {ObjectType} objectType Object type.
+	 * @param {ObjectID}   objectId   Object ID.
+	 * @param {string}     origin     The source of change.
+	 */
+	function recordPersistence(
+		objectType: ObjectType,
+		objectId: ObjectID,
+		origin: string
+	): void {
+		const entityId = getEntityId( objectType, objectId );
+		const entityState = entityStates.get( entityId );
+
+		if ( ! entityState ) {
+			return;
+		}
+
+		const { ydoc } = entityState;
+
+		const lastPersistedAt = Date.now();
+
+		Y.transact(
+			ydoc,
+			() => {
+				const stateMap = ydoc.getMap( 'state' );
+				stateMap.set( PERSISTED_AT_KEY, lastPersistedAt );
+				stateMap.set( PERSISTED_BY_KEY, ydoc.clientID );
+			},
+			origin
+		);
+	}
+
 	return {
 		createMeta: createEntityMeta,
 		load: loadEntity,
 		undoManager,
 		unload: unloadEntity,
 		update: updateCRDTDoc,
+		markPersisted: recordPersistence,
 	};
 }

packages/sync/src/test/manager.ts

@@ -19,6 +19,9 @@ import {
 import { createSyncManager } from '../manager';
 import {
 	CRDT_RECORD_MAP_KEY,
+	CRDT_STATE_MAP_KEY,
+	CRDT_STATE_PERSISTED_AT_KEY,
+	CRDT_STATE_PERSISTED_BY_KEY,
 	WORDPRESS_META_KEY_FOR_CRDT_DOC_PERSISTENCE,
 } from '../config';
 import { createPersistedCRDTDoc } from '../persistence';
@@ -714,4 +717,139 @@ describe( 'SyncManager', () => {
 			expect( mockHandlers.editRecord ).not.toHaveBeenCalled();
 		} );
 	} );
+
+	describe( 'markPersisted', () => {
+		it( 'updates the state map with timestamp and client ID', async () => {
+			// Capture the Y.Doc from provider creator.
+			let capturedDoc: Y.Doc | null = null;
+			mockProviderCreator.mockImplementation(
+				async (
+					_objectType: string,
+					_objectId: string,
+					ydoc: Y.Doc
+				) => {
+					capturedDoc = ydoc;
+					return mockProviderResult;
+				}
+			);
+
+			const manager = createSyncManager();
+
+			await manager.load(
+				mockSyncConfig,
+				'post',
+				'123',
+				mockRecord,
+				mockHandlers
+			);
+
+			expect( capturedDoc ).not.toBeNull();
+			const ydoc = capturedDoc as unknown as Y.Doc;
+			const stateMap = ydoc.getMap( CRDT_STATE_MAP_KEY );
+
+			const beforeTime = Date.now();
+			manager.markPersisted( 'post', '123', 'test-origin' );
+			const afterTime = Date.now();
+
+			const persistedAt = stateMap.get(
+				CRDT_STATE_PERSISTED_AT_KEY
+			) as number;
+			const persistedBy = stateMap.get(
+				CRDT_STATE_PERSISTED_BY_KEY
+			) as number;
+
+			expect( persistedAt ).toBeGreaterThanOrEqual( beforeTime );
+			expect( persistedAt ).toBeLessThanOrEqual( afterTime );
+			expect( persistedBy ).toBe( ydoc.clientID );
+		} );
+
+		it( 'does not throw when entity is not loaded', () => {
+			const manager = createSyncManager();
+
+			expect( () => {
+				manager.markPersisted( 'post', '999', 'test-origin' );
+			} ).not.toThrow();
+		} );
+
+		it( 'applies changes with specified origin', async () => {
+			// Capture the Y.Doc from provider creator.
+			let capturedDoc: Y.Doc | null = null;
+			mockProviderCreator.mockImplementation(
+				async (
+					_objectType: string,
+					_objectId: string,
+					ydoc: Y.Doc
+				) => {
+					capturedDoc = ydoc;
+					return mockProviderResult;
+				}
+			);
+
+			const manager = createSyncManager();
+
+			await manager.load(
+				mockSyncConfig,
+				'post',
+				'123',
+				mockRecord,
+				mockHandlers
+			);
+
+			expect( capturedDoc ).not.toBeNull();
+			const ydoc = capturedDoc as unknown as Y.Doc;
+
+			let capturedOrigin: string | undefined;
+			ydoc.on( 'afterTransaction', ( transaction: Y.Transaction ) => {
+				capturedOrigin = transaction.origin as string | undefined;
+			} );
+
+			const customOrigin = 'custom-persist-origin';
+			manager.markPersisted( 'post', '123', customOrigin );
+
+			expect( capturedOrigin ).toBe( customOrigin );
+		} );
+
+		it( 'updates timestamp each time markPersisted is called', async () => {
+			// Capture the Y.Doc from provider creator.
+			let capturedDoc: Y.Doc | null = null;
+			mockProviderCreator.mockImplementation(
+				async (
+					_objectType: string,
+					_objectId: string,
+					ydoc: Y.Doc
+				) => {
+					capturedDoc = ydoc;
+					return mockProviderResult;
+				}
+			);
+
+			const manager = createSyncManager();
+
+			await manager.load(
+				mockSyncConfig,
+				'post',
+				'123',
+				mockRecord,
+				mockHandlers
+			);
+
+			expect( capturedDoc ).not.toBeNull();
+			const ydoc = capturedDoc as unknown as Y.Doc;
+			const stateMap = ydoc.getMap( CRDT_STATE_MAP_KEY );
+
+			manager.markPersisted( 'post', '123', 'test-origin' );
+			const firstTimestamp = stateMap.get(
+				CRDT_STATE_PERSISTED_AT_KEY
+			) as number;
+
+			await new Promise( ( resolve ) => setTimeout( resolve, 10 ) );
+
+			manager.markPersisted( 'post', '123', 'test-origin' );
+			const secondTimestamp = stateMap.get(
+				CRDT_STATE_PERSISTED_AT_KEY
+			) as number;
+
+			expect( secondTimestamp ).toBeGreaterThan( firstTimestamp );
+		} );
+	} );
 } );

packages/sync/src/types.ts

@@ -98,6 +98,11 @@ export interface SyncManager {
 		changes: Partial< ObjectData >,
 		origin: string
 	) => void;
+	markPersisted: (
+		objectType: ObjectType,
+		objectId: ObjectID,
+		origin: string
+	) => void;
 }
 
 export interface SyncUndoManager extends WPUndoManager< ObjectData > {