core: handle inbox item sync & decryption (#8733)

* core: handle inbox item sync & decryption
Signed-off-by: 01zulfi <85733202+01zulfi@users.noreply.github.com>

* core: minor refactors in handling inbox item sync
Signed-off-by: 01zulfi <85733202+01zulfi@users.noreply.github.com>

* core: use inbox item salt
Signed-off-by: 01zulfi <85733202+01zulfi@users.noreply.github.com>

* core: check inbox item version
Signed-off-by: 01zulfi <85733202+01zulfi@users.noreply.github.com>

packages/core/src/api/sync/index.ts

@@ -29,7 +29,7 @@ import Constants from "../../utils/constants.js";
 import TokenManager from "../token-manager.js";
 import Collector from "./collector.js";
 import { type HubConnection } from "@microsoft/signalr";
-import Merger from "./merger.js";
+import Merger, { handleInboxItems } from "./merger.js";
 import { AutoSync } from "./auto-sync.js";
 import { logger } from "../../logger.js";
 import { Mutex } from "async-mutex";
@@ -49,6 +49,7 @@ import {
 import {
   SYNC_COLLECTIONS_MAP,
   SyncableItemType,
+  SyncInboxItem,
   SyncTransferItem
 } from "./types.js";
 import { DownloadableFile } from "../../database/fs.js";
@@ -233,7 +234,19 @@ class Sync {
   async fetch(deviceId: string, options: SyncOptions) {
     await this.checkConnection();
 
-    await this.connection?.invoke("RequestFetchV2", deviceId);
+    try {
+      await this.connection?.invoke("RequestFetchV3", deviceId);
+    } catch (error) {
+      if (
+        error instanceof Error &&
+        error.message.includes("HubException: Method does not exist")
+      ) {
+        this.logger.warn(
+          "RequestFetchV3 failed, falling back to RequestFetchV2"
+        );
+        await this.connection?.invoke("RequestFetchV2", deviceId);
+      }
+    }
 
     if (this.conflictedNoteIds.length > 0) {
       await this.db
@@ -478,6 +491,19 @@ class Sync {
 
       return true;
     });
+
+    this.connection.on(
+      "SendInboxItems",
+      async (inboxItems: SyncInboxItem[]) => {
+        if (this.connection?.state !== HubConnectionState.Connected) {
+          return false;
+        }
+
+        await handleInboxItems(inboxItems, this.db);
+
+        return true;
+      }
+    );
   }
 
   private async getKey() {

packages/core/src/api/sync/merger.ts

@@ -27,6 +27,7 @@ import {
   MaybeDeletedItem,
   isDeleted
 } from "../../types.js";
+import { ParsedInboxItem, SyncInboxItem } from "./types.js";
 
 const THRESHOLD = process.env.NODE_ENV === "test" ? 6 * 1000 : 60 * 1000;
 class Merger {
@@ -146,3 +147,81 @@ export function isContentConflicted(
     return "merge";
   }
 }
+
+export async function handleInboxItems(
+  inboxItems: SyncInboxItem[],
+  db: Database
+) {
+  const inboxKeys = await db.user.getInboxKeys();
+  if (!inboxKeys) {
+    logger.error("No inbox keys found, cannot process inbox items.");
+    return;
+  }
+
+  for (const item of inboxItems) {
+    try {
+      if (await db.notes.exists(item.id)) {
+        logger.info("Inbox item already exists, skipping.", {
+          inboxItemId: item.id
+        });
+        continue;
+      }
+
+      const decryptedKey = await db.storage().decryptAsymmetric(inboxKeys, {
+        alg: item.key.alg,
+        cipher: item.key.cipher,
+        format: "base64",
+        length: item.key.length
+      });
+      const decryptedItem = await db.storage().decrypt(
+        { key: decryptedKey },
+        {
+          alg: item.alg,
+          iv: item.iv,
+          cipher: item.cipher,
+          format: "base64",
+          length: item.length,
+          salt: item.salt
+        }
+      );
+      const parsed = JSON.parse(decryptedItem) as ParsedInboxItem;
+      if (parsed.type !== "note") {
+        continue;
+      }
+      if (parsed.version !== 1) {
+        continue;
+      }
+
+      await db.notes.add({
+        id: item.id,
+        title: parsed.title,
+        favorite: parsed.favorite,
+        pinned: parsed.pinned,
+        readonly: parsed.readonly,
+        content: {
+          data: parsed?.content?.data ?? "",
+          type: "tiptap"
+        }
+      });
+      if (parsed.archived !== undefined) {
+        await db.notes.archive(parsed.archived, item.id);
+      }
+      for (const notebookId of parsed.notebookIds || []) {
+        if (!(await db.notebooks.exists(notebookId))) continue;
+        await db.notes.addToNotebook(notebookId, item.id);
+      }
+      for (const tagId of parsed.tagIds || []) {
+        if (!(await db.tags.exists(tagId))) continue;
+        await db.relations.add(
+          { type: "tag", id: tagId },
+          { type: "note", id: item.id }
+        );
+      }
+    } catch (e) {
+      logger.error(e, "Failed to process inbox item.", {
+        inboxItem: item
+      });
+      continue;
+    }
+  }
+}

packages/core/src/api/sync/types.ts

@@ -48,3 +48,24 @@ export type SyncTransferItem = {
   items: SyncItem[];
   type: SyncableItemType;
 };
+
+export type SyncInboxItem = Omit<SyncItem, "format"> & {
+  key: Omit<Cipher<"base64">, "format" | "salt" | "iv">;
+};
+
+export type ParsedInboxItem = {
+  title: string;
+  pinned?: boolean;
+  favorite?: boolean;
+  readonly?: boolean;
+  archived?: boolean;
+  notebookIds?: string[];
+  tagIds?: string[];
+  type: "note";
+  source: string;
+  version: 1;
+  content?: {
+    type: "html";
+    data: string;
+  };
+};