feat: Use notify push for sync messages during editing

Signed-off-by: Julius Härtl <[email protected]>

lib/Service/ApiService.php

@@ -29,6 +29,7 @@
 use Exception;
 use InvalidArgumentException;
 use OCA\Files_Sharing\SharedStorage;
+use OCA\NotifyPush\Queue\IQueue;
 use OCA\Text\AppInfo\Application;
 use OCA\Text\Db\Document;
 use OCA\Text\Db\Session;
@@ -45,7 +46,10 @@
 use OCP\IL10N;
 use OCP\IRequest;
 use OCP\Lock\LockedException;
+use OCP\Server;
 use OCP\Share\IShare;
+use Psr\Container\ContainerExceptionInterface;
+use Psr\Container\NotFoundExceptionInterface;
 use Psr\Log\LoggerInterface;
 
 class ApiService {
@@ -203,6 +207,7 @@ public function push(Session $session, Document $document, int $version, array $
 		}
 		try {
 			$result = $this->documentService->addStep($document, $session, $steps, $version, $token);
+			$this->addToPushQueue($document, [$awareness, ...array_values($steps)]);
 		} catch (InvalidArgumentException $e) {
 			return new DataResponse($e->getMessage(), 422);
 		} catch (DoesNotExistException|NotPermittedException) {
@@ -212,6 +217,26 @@ public function push(Session $session, Document $document, int $version, array $
 		return new DataResponse($result);
 	}
 
+	private function addToPushQueue(Document $document, array $steps): void {
+		try {
+			$queue = Server::get(IQueue::class);
+			$sessions = $this->sessionService->getActiveSessions($document->getId());
+			$sessions = array_values(array_filter(array_unique(array_map(fn ($session): ?string => $session['userId'], $sessions))));
+			foreach ($sessions as $userId) {
+				// Get sync response
+				$queue->push('notify_custom', [
+					'user' => $userId,
+					'message' => 'text_steps',
+					'body' => [
+						'documentId' => $document->getId(),
+						'steps' => $steps,
+					],
+				]);
+			}
+		} catch (NotFoundExceptionInterface|ContainerExceptionInterface) {
+		}
+	}
+
 	public function sync(Session $session, Document $document, int $version = 0, ?string $shareToken = null): DataResponse {
 		$documentId = $session->getDocumentId();
 		$result = [];

package.json

@@ -47,6 +47,7 @@
     "@nextcloud/l10n": "^2.2.0",
     "@nextcloud/logger": "^2.7.0",
     "@nextcloud/moment": "^1.3.1",
+    "@nextcloud/notify_push": "^1.1.4",
     "@nextcloud/router": "^3.0.0",
     "@nextcloud/vue": "^8.10.0",
     "@quartzy/markdown-it-mentions": "^0.2.0",

src/services/NotifyService.js

@@ -0,0 +1,34 @@
+/*
+ * @copyright Copyright (c) 2023 Julius Härtl <[email protected]>
+ *
+ * @author Julius Härtl <[email protected]>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+import mitt from 'mitt'
+import { listen } from '@nextcloud/notify_push'
+
+if (!window._nc_text_notify) {
+	const useNotifyPush = listen('text_steps', (messageType, messageBody) => {
+		window._nc_text_notify?.emit('notify_push', { messageType, messageBody })
+	})
+	window._nc_text_notify = useNotifyPush ? mitt() : null
+}
+
+export default () => {
+	return window._nc_text_notify
+}

src/services/PollingBackend.js

@@ -22,6 +22,7 @@
 import { logger } from '../helpers/logger.js'
 import { SyncService, ERROR_TYPE } from './SyncService.js'
 import { Connection } from './SessionApi.js'
+import getNotifyBus from './NotifyService.js'
 
 /**
  * Minimum inverval to refetch the document changes
@@ -50,7 +51,9 @@ const FETCH_INTERVAL_SINGLE_EDITOR = 5000
  *
  * @type {number} time in ms
  */
-const FETCH_INTERVAL_INVISIBLE = 60000
+const FETCH_INTERVAL_INVISIBLE = 30000
+
+const FETCH_INTERVAL_NOTIFY = 30000
 
 /* Maximum number of retries for fetching before emitting a connection error */
 const MAX_RETRY_FETCH_COUNT = 5
@@ -73,6 +76,7 @@ class PollingBackend {
 	#fetchRetryCounter
 	#pollActive
 	#initialLoadingFinished
+	#notifyPushBus
 
 	constructor(syncService, connection) {
 		this.#syncService = syncService
@@ -90,6 +94,7 @@ class PollingBackend {
 		this.#initialLoadingFinished = false
 		this.fetcher = setInterval(this._fetchSteps.bind(this), 50)
 		document.addEventListener('visibilitychange', this.visibilitychange.bind(this))
+		this.#notifyPushBus = getNotifyBus()
 	}
 
 	/**
@@ -121,6 +126,13 @@ class PollingBackend {
 		this.#pollActive = false
 	}
 
+	handleNotifyPush({ messageType, messageBody }) {
+		if (messageBody.documentId !== this.#connection.document.id) {
+			return
+		}
+		this._handleResponse({ data: messageBody.response })
+	}
+
 	_handleResponse({ data }) {
 		const { document, sessions } = data
 		this.#fetchRetryCounter = 0
@@ -195,15 +207,26 @@ class PollingBackend {
 	}
 
 	resetRefetchTimer() {
+		if (this.#notifyPushBus && this.#initialLoadingFinished) {
+			this.#fetchInterval = FETCH_INTERVAL_NOTIFY
+			return
+		}
 		this.#fetchInterval = FETCH_INTERVAL
-
 	}
 
 	increaseRefetchTimer() {
+		if (this.#notifyPushBus && this.#initialLoadingFinished) {
+			this.#fetchInterval = FETCH_INTERVAL_NOTIFY
+			return
+		}
 		this.#fetchInterval = Math.min(this.#fetchInterval * 2, FETCH_INTERVAL_MAX)
 	}
 
 	maximumRefetchTimer() {
+		if (this.#notifyPushBus && this.#initialLoadingFinished) {
+			this.#fetchInterval = FETCH_INTERVAL_NOTIFY
+			return
+		}
 		this.#fetchInterval = FETCH_INTERVAL_SINGLE_EDITOR
 	}
 

src/services/WebSocketPolyfill.js

@@ -22,6 +22,7 @@
 
 import { logger } from '../helpers/logger.js'
 import { encodeArrayBuffer, decodeArrayBuffer } from '../helpers/base64.js'
+import getNotifyBus from './NotifyService.js'
 
 /**
  *
@@ -42,8 +43,11 @@ export default function initWebSocketPolyfill(syncService, fileId, initialSessio
 		onclose
 		onopen
 		#handlers
+		#notifyPushBus
 
 		constructor(url) {
+			this.#notifyPushBus = getNotifyBus()
+			this.#notifyPushBus?.on('notify_push', this.#onNotifyPush.bind(this))
 			this.url = url
 			logger.debug('WebSocketPolyfill#constructor', { url, fileId, initialSession })
 			this.#registerHandlers({
@@ -118,6 +122,7 @@ export default function initWebSocketPolyfill(syncService, fileId, initialSessio
 			Object.entries(this.#handlers)
 				.forEach(([key, value]) => syncService.off(key, value))
 			this.#handlers = []
+			this.#notifyPushBus?.off('notify_push', this.#onNotifyPush.bind(this))
 			syncService.close().then(() => {
 				this.onclose()
 			})
@@ -146,5 +151,15 @@ export default function initWebSocketPolyfill(syncService, fileId, initialSessio
 			}
 		}
 
+		#onNotifyPush({ messageType, messageBody }) {
+			if (messageBody.documentId !== fileId) {
+				return
+			}
+			messageBody.steps.forEach(step => {
+				const data = decodeArrayBuffer(step)
+				this.onmessage({ data })
+			})
+		}
+
 	}
 }

tests/stub.php

@@ -49,3 +49,15 @@ abstract public function setWantsNotification(bool $wantsNotification): void;
 		abstract public function setNotificationTarget(?string $notificationTarget): void;
 	}
 }
+
+
+namespace OCA\NotifyPush\Queue {
+	interface IQueue {
+		/**
+		 * @param string $channel
+		 * @param mixed $message
+		 * @return void
+		 */
+		public function push(string $channel, $message);
+	}
+}