chore: Basic backend y.js message decoder

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

composer/composer/autoload_classmap.php

@@ -61,4 +61,5 @@
     'OCA\\Text\\Service\\SessionService' => $baseDir . '/../lib/Service/SessionService.php',
     'OCA\\Text\\Service\\WorkspaceService' => $baseDir . '/../lib/Service/WorkspaceService.php',
     'OCA\\Text\\TextFile' => $baseDir . '/../lib/TextFile.php',
+    'OCA\\Text\\YjsMessage' => $baseDir . '/../lib/YjsMessage.php',
 );

composer/composer/autoload_static.php

@@ -76,6 +76,7 @@ class ComposerStaticInitText
         'OCA\\Text\\Service\\SessionService' => __DIR__ . '/..' . '/../lib/Service/SessionService.php',
         'OCA\\Text\\Service\\WorkspaceService' => __DIR__ . '/..' . '/../lib/Service/WorkspaceService.php',
         'OCA\\Text\\TextFile' => __DIR__ . '/..' . '/../lib/TextFile.php',
+        'OCA\\Text\\YjsMessage' => __DIR__ . '/..' . '/../lib/YjsMessage.php',
     );
 
     public static function getInitializer(ClassLoader $loader)

lib/Service/DocumentService.php

@@ -36,6 +36,7 @@
 use OCA\Text\Db\StepMapper;
 use OCA\Text\Exception\DocumentHasUnsavedChangesException;
 use OCA\Text\Exception\DocumentSaveConflictException;
+use OCA\Text\YjsMessage;
 use OCP\AppFramework\Db\DoesNotExistException;
 use OCP\AppFramework\Db\Entity;
 use OCP\Constants;
@@ -227,11 +228,9 @@ public function addStep($documentId, $sessionId, $steps, $version): array {
 		$getStepsSinceVersion = null;
 		$newVersion = $version;
 		foreach ($steps as $step) {
-			// Steps are base64 encoded messages of the yjs protocols
-			// https://github.com/yjs/y-protocols
-			// Base64 encoded values smaller than "AAE" belong to sync step 1 messages.
-			// These messages query other participants for their current state.
-			if ($step < "AAE") {
+			$message = YjsMessage::fromBase64($step);
+			// Filter out query steps as they would just trigger clients to send their steps again
+			if ($message->getYjsMessageType() === YjsMessage::YJS_MESSAGE_SYNC && $message->getYjsSyncType() === YjsMessage::YJS_MESSAGE_SYNC_STEP1) {
 				array_push($querySteps, $step);
 			} else {
 				array_push($stepsToInsert, $step);
@@ -245,8 +244,9 @@ public function addStep($documentId, $sessionId, $steps, $version): array {
 		$allSteps = $this->getSteps($documentId, $getStepsSinceVersion);
 		$stepsToReturn = [];
 		foreach ($allSteps as $step) {
-			if ($step < "AAQ") {
-				array_push($stepsToReturn, $step);
+			$message = YjsMessage::fromBase64($step);
+			if ($message->getYjsMessageType() === YjsMessage::YJS_MESSAGE_SYNC) {
+				$stepsToReturn[] = $step;
 			}
 		}
 		return [

lib/Service/SessionService.php

@@ -28,6 +28,7 @@
 
 use OCA\Text\Db\Session;
 use OCA\Text\Db\SessionMapper;
+use OCA\Text\YjsMessage;
 use OCP\AppFramework\Db\DoesNotExistException;
 use OCP\AppFramework\Utility\ITimeFactory;
 use OCP\DirectEditing\IManager;
@@ -240,6 +241,12 @@ public function updateSessionAwareness(int $documentId, int $sessionId, string $
 		if (empty($message)) {
 			return $session;
 		}
+
+		$decoded = YjsMessage::fromBase64($message);
+		if ($decoded->getYjsMessageType() !== YjsMessage::YJS_MESSAGE_AWARENESS) {
+			throw new \ValueError('Message passed was not an awareness message');
+		}
+
 		$session->setLastAwarenessMessage($message);
 		return $this->sessionMapper->update($session);
 	}

lib/YjsMessage.php

@@ -0,0 +1,88 @@
+<?php
+
+namespace OCA\Text;
+
+use InvalidArgumentException;
+
+/**
+ * Steps are base64 encoded messages of the yjs protocols
+ * https://github.com/yjs/y-protocols
+ *
+ * This class is a simple representation of a message containing some methods
+ * to decode parts of it for what we need on the backend
+ *
+ * Relevant resources:
+ * https://github.com/yjs/y-protocols/blob/master/PROTOCOL.md
+ * https://github.com/yjs/y-websocket/blob/master/src/y-websocket.js#L19-L22
+ * https://github.com/yjs/y-protocols/blob/master/sync.js#L38-L40
+ * https://github.com/dmonad/lib0/blob/master/decoding.js
+ */
+class YjsMessage {
+
+	public const YJS_MESSAGE_SYNC = 0;
+	public const YJS_MESSAGE_AWARENESS = 1;
+	public const YJS_MESSAGE_AWARENESS_QUERY = 3;
+
+	public const YJS_MESSAGE_SYNC_STEP1 = 0;
+	public const YJS_MESSAGE_SYNC_STEP2 = 1;
+	public const YJS_MESSAGE_SYNC_UPDATE = 2;
+
+	private int $pos = 0;
+
+	public function __construct(
+		private string $data = ''
+	) {
+	}
+
+	public static function fromBase64(string $data = ''): self {
+		return new self(base64_decode($data));
+	}
+
+	/**
+	 * https://github.com/dmonad/lib0/blob/bd69ab4dc701d77e808f2bab08d96d63acd297da/decoding.js#L242
+	 */
+	public function readVarUint(): int {
+		$bytes = unpack('C*', substr($this->data, $this->pos, 4));
+		$segments = [];
+		$offset = 0;
+		foreach ($bytes as $byte) {
+			$offset++;
+			$segments[] = (0x7f & $byte);
+
+			if (($byte & 0x80) === 0) {
+				$integer = 0;
+				foreach($segments as $segment) {
+					$integer <<= 7;
+					$integer |= (0x7f & $segment);
+				}
+				$this->pos += $offset;
+				return $integer;
+			}
+		}
+		if (($byte & 0x80) !== 0) {
+			throw new InvalidArgumentException('Incomplete byte sequence.');
+		}
+		throw new InvalidArgumentException('Incomplete byte sequence.');
+	}
+
+	public function getYjsMessageType(): int {
+		$oldPos = $this->pos;
+		$this->pos = 0;
+		$messageType = $this->readVarUint();
+		$this->pos = $oldPos;
+		return $messageType;
+	}
+
+	public function getYjsSyncType(): int {
+		$oldPos = $this->pos;
+		$this->pos = 0;
+		$messageType = $this->readVarUint();
+		if ($messageType !== self::YJS_MESSAGE_SYNC) {
+			throw new \ValueError('Message is not a sync message');
+		}
+		$syncType = $this->readVarUint();
+		$this->pos = $oldPos;
+		return $syncType;
+	}
+
+}

tests/unit/YjsMessageTest.php

@@ -0,0 +1,42 @@
+<?php
+
+namespace OCA\Text;
+
+use Test\TestCase;
+
+class YjsMessageTest extends TestCase {
+	protected function setUp(): void {
+		parent::setUp();
+	}
+
+	// https://github.com/yjs/y-dat/blob/745d25f9690fceae5901d1225575fe8b6bcafdd7/src/y-dat.js#LL207C59-L210C1
+	public function dataMessageTypes() {
+		return [
+			//
+			['AAABAA==', 0, 0],
+			// messageSync messageYjsSyncStep1
+			['AAAyCIqghaQNBvrS2LoMB8L10I8KrQPD+t3RB2DcrYL4A40Ema2O4AMHz9bk8AIOtbm0PAE=', 0, 0],
+			// messageAwareness
+			['AQoBkZWK7gMAAnt9', 1, null],
+			['AUIBwvXQjwqWAzl7InVzZXIiOnsibmFtZSI6ImFkbWluIiwiY29sb3IiOiIjZDA5ZTZkIn0sImN1cnNvciI6bnVsbH0=', 1, null],
+			['AegBAcP63dEHtwHeAXsidXNlciI6eyJuYW1lIjoiR3Vlc3QiLCJjb2xvciI6IiM5M2IyN2IifSwiY3Vyc29yIjp7ImFuY2hvciI6eyJ0eXBlIjp7ImNsaWVudCI6MjcxNzEzNzYwMiwiY2xvY2siOjl9LCJ0bmFtZSI6bnVsbCwiaXRlbSI6bnVsbCwiYXNzb2MiOjB9LCJoZWFkIjp7InR5cGUiOnsiY2xpZW50IjoyNzE3MTM3NjAyLCJjbG9jayI6OX0sInRuYW1lIjpudWxsLCJpdGVtIjpudWxsLCJhc3NvYyI6MH19fQ==', 1, null],
+			['AbsBAZGViu4DArIBeyJ1c2VyIjp7Im5hbWUiOiJHdWVzdCIsImNvbG9yIjoiI2I4YmU2OCJ9LCJjdXJzb3IiOnsiYW5jaG9yIjp7InR5cGUiOm51bGwsInRuYW1lIjoiZGVmYXVsdCIsIml0ZW0iOm51bGwsImFzc29jIjowfSwiaGVhZCI6eyJ0eXBlIjpudWxsLCJ0bmFtZSI6ImRlZmF1bHQiLCJpdGVtIjpudWxsLCJhc3NvYyI6MH19fQ==', 1, null],
+			// messageSync messageYjsUpdate
+			['AAISAQHD+t3RB2CEwvXQjwpHAWEA', 0, 2],
+			['AAI0AQOKoIWkDQAHAQdkZWZhdWx0AwlwYXJhZ3JhcGgHAIqghaQNAAYEAIqghaQNAQR0ZXN0AA==', 0, 2],
+			['AAIdAQGRlYruAx2okZWK7gMbAXcCaC0BkZWK7gMBGwE=', 0, 2],
+			['AAIKAAGRlYruAwEVBA==', 0, 2],
+		];
+	}
+
+	/** @dataProvider dataMessageTypes */
+	public function testMessageTypes($data, $type, $subtype) {
+		$buffer = YjsMessage::fromBase64($data);
+		$unpack1 = $buffer->getYjsMessageType();
+		self::assertEquals($type, $unpack1, 'type');
+		if ($subtype !== null) {
+			$unpack2 = $buffer->getYjsSyncType();
+			self::assertEquals($subtype, $unpack2);
+		}
+	}
+}