feat: Use notify push for sync messages during editing

Signed-off-by: Julius Härtl <jus@bitgrid.net>

lib/Service/ApiService.php

@@ -12,6 +12,7 @@ namespace OCA\Text\Service;
 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;
@@ -32,7 +33,6 @@ use OCP\Share\IShare;
 use Psr\Log\LoggerInterface;
 
 class ApiService {
-
 	public function __construct(
 		private IRequest $request,
 		private SessionService $sessionService,
@@ -41,6 +41,7 @@ class ApiService {
 		private LoggerInterface $logger,
 		private IL10N $l10n,
 		private ?string $userId,
+		private ?IQueue $queue,
 	) {
 	}
 
@@ -181,6 +182,7 @@ class ApiService {
 		}
 		try {
 			$result = $this->documentService->addStep($document, $session, $steps, $version, $token);
+			$this->addToPushQueue($document, [$awareness, ...array_values($steps)]);
 		} catch (InvalidArgumentException $e) {
 			return new DataResponse(['error' => $e->getMessage()], Http::STATUS_UNPROCESSABLE_ENTITY);
 		} catch (DoesNotExistException|NotPermittedException) {
@@ -190,6 +192,27 @@ class ApiService {
 		return new DataResponse($result);
 	}
 
+	private function addToPushQueue(Document $document, array $steps): void {
+		if ($this->queue === null) {
+			return;
+		}
+
+		$sessions = $this->sessionService->getActiveSessions($document->getId());
+		$userIds = array_values(array_filter(array_unique(
+			array_map(fn ($session): ?string => $session['userId'], $sessions)
+		)));
+		foreach ($userIds as $userId) {
+			$this->queue->push('notify_custom', [
+				'user' => $userId,
+				'message' => 'text_steps',
+				'body' => [
+					'documentId' => $document->getId(),
+					'steps' => $steps,
+				],
+			]);
+		}
+	}
+
 	public function sync(Session $session, Document $document, int $version = 0, ?string $shareToken = null): DataResponse {
 		$documentId = $session->getDocumentId();
 		$result = [];

package-lock.json

@@ -21,6 +21,7 @@
         "@nextcloud/l10n": "^3.1.0",
         "@nextcloud/logger": "^3.0.2",
         "@nextcloud/moment": "^1.3.1",
+        "@nextcloud/notify_push": "^1.3.0",
         "@nextcloud/router": "^3.0.1",
         "@nextcloud/vue": "^8.14.0",
         "@quartzy/markdown-it-mentions": "^0.2.0",
@@ -4304,6 +4305,16 @@
         "npm": "^10.0.0"
       }
     },
+    "node_modules/@nextcloud/notify_push": {
+      "version": "1.3.0",
+      "resolved": "https://registry.npmjs.org/@nextcloud/notify_push/-/notify_push-1.3.0.tgz",
+      "integrity": "sha512-WmyINTP/RynrfrOdyxzcntwV79b88uhXHU3cVJEcMzuh7wt6YT66kitjuQHMGlrG/xlEwk4qUKEM/NpFqVcvJg==",
+      "dependencies": {
+        "@nextcloud/axios": "^2.5.0",
+        "@nextcloud/capabilities": "^1.2.0",
+        "@nextcloud/event-bus": "^3.3.0"
+      }
+    },
     "node_modules/@nextcloud/paths": {
       "version": "2.1.0",
       "resolved": "https://registry.npmjs.org/@nextcloud/paths/-/paths-2.1.0.tgz",
@@ -31196,6 +31207,16 @@
         }
       }
     },
+    "@nextcloud/notify_push": {
+      "version": "1.3.0",
+      "resolved": "https://registry.npmjs.org/@nextcloud/notify_push/-/notify_push-1.3.0.tgz",
+      "integrity": "sha512-WmyINTP/RynrfrOdyxzcntwV79b88uhXHU3cVJEcMzuh7wt6YT66kitjuQHMGlrG/xlEwk4qUKEM/NpFqVcvJg==",
+      "requires": {
+        "@nextcloud/axios": "^2.5.0",
+        "@nextcloud/capabilities": "^1.2.0",
+        "@nextcloud/event-bus": "^3.3.0"
+      }
+    },
     "@nextcloud/paths": {
       "version": "2.1.0",
       "resolved": "https://registry.npmjs.org/@nextcloud/paths/-/paths-2.1.0.tgz",

package.json

@@ -45,6 +45,7 @@
     "@nextcloud/l10n": "^3.1.0",
     "@nextcloud/logger": "^3.0.2",
     "@nextcloud/moment": "^1.3.1",
+    "@nextcloud/notify_push": "^1.3.0",
     "@nextcloud/router": "^3.0.1",
     "@nextcloud/vue": "^8.14.0",
     "@quartzy/markdown-it-mentions": "^0.2.0",

src/services/NotifyService.js

@@ -0,0 +1,34 @@
+/*
+ * @copyright Copyright (c) 2023 Julius Härtl <jus@bitgrid.net>
+ *
+ * @author Julius Härtl <jus@bitgrid.net>
+ *
+ * @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

@@ -5,6 +5,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
@@ -39,7 +40,9 @@ const FETCH_INTERVAL_READ_ONLY = 30000
  *
  * @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
@@ -62,6 +65,7 @@ class PollingBackend {
 	#fetchRetryCounter
 	#pollActive
 	#initialLoadingFinished
+	#notifyPushBus
 
 	constructor(syncService, connection) {
 		this.#syncService = syncService
@@ -79,6 +83,7 @@ class PollingBackend {
 		this.#initialLoadingFinished = false
 		this.fetcher = setInterval(this._fetchSteps.bind(this), 50)
 		document.addEventListener('visibilitychange', this.visibilitychange.bind(this))
+		this.#notifyPushBus = getNotifyBus()
 	}
 
 	/**
@@ -110,6 +115,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
@@ -189,15 +201,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

@@ -6,6 +6,7 @@
 import { logger } from '../helpers/logger.js'
 import { decodeArrayBuffer } from '../helpers/base64.ts'
 import { getSteps, getAwareness } from '../helpers/yjs.js'
+import getNotifyBus from './NotifyService.js'
 
 /**
  *
@@ -26,8 +27,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({
@@ -91,9 +95,21 @@ 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))
 			this.onclose()
 			logger.debug('Websocket closed')
 		}
 
+		#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 @@ namespace OCA\TpAssistant\Event {
 		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);
+	}
+}