New database structure for sync engine

2026-09-03 12:52:05 +02:00 · 2024-10-26 11:38:06 +02:00
parent f7d1d12e95
commit 35ee77ea06
7 changed files with 259 additions and 150 deletions
--- a/desktop/src/main/handlers/mutations/server-node-sync.ts
+++ b/desktop/src/main/handlers/mutations/server-node-sync.ts
@@ -115,6 +115,7 @@ export class ServerNodeSyncMutationHandler
      type: 'local_node_sync',
      nodeId: input.id,
      versionId: input.versionId,
+      workspaceId: input.workspaceId,
    });

    return {
--- a/desktop/src/operations/messages/local-node-sync.ts
+++ b/desktop/src/operations/messages/local-node-sync.ts
@@ -2,6 +2,7 @@ export type LocalNodeSyncMessageInput = {
  type: 'local_node_sync';
  nodeId: string;
  versionId: string;
+  workspaceId: string;
 };

 declare module '@/operations/messages' {
--- a/server/src/consumers/changes-subcriber.ts
+++ b/server/src/consumers/changes-subcriber.ts
@@ -12,27 +12,34 @@ export const initChangesSubscriber = async () => {
 const handleEvent = async (event: string) => {
  const data: ServerNodeChangeEvent = JSON.parse(event);

-  const deviceIds: string[] = [];
+  const userDevices = new Map<string, string[]>();
  for (const connection of socketManager.getConnections()) {
-    deviceIds.push(connection.getDeviceId());
+    const workspaceUsers = connection.getWorkspaceUsers();
+    for (const workspaceUser of workspaceUsers) {
+      if (workspaceUser.workspaceId !== data.workspaceId) {
+        continue;
+      }
+
+      const userIds = userDevices.get(workspaceUser.userId) ?? [];
+      userIds.push(connection.getDeviceId());
+      userDevices.set(workspaceUser.userId, userIds);
+    }
  }

-  if (deviceIds.length === 0) {
+  const userIds = Array.from(userDevices.keys());
+  if (userIds.length === 0) {
    return;
  }

-  const nodeVersions = await database
-    .selectFrom('device_node_versions')
+  const nodeUserStates = await database
+    .selectFrom('node_user_states')
    .selectAll()
    .where((eb) =>
-      eb.and([
-        eb('device_id', 'in', deviceIds),
-        eb('node_id', '=', data.nodeId),
-      ]),
+      eb.and([eb('user_id', 'in', userIds), eb('node_id', '=', data.nodeId)]),
    )
    .execute();

-  if (nodeVersions.length === 0) {
+  if (nodeUserStates.length === 0) {
    return;
  }

@@ -52,32 +59,43 @@ const handleEvent = async (event: string) => {
    .where('id', '=', data.nodeId)
    .executeTakeFirst();

-  for (const nodeVersion of nodeVersions) {
-    const socketConnection = socketManager.getConnection(nodeVersion.device_id);
-    if (!socketConnection) {
+  if (!node) {
+    return;
+  }
+
+  for (const nodeUserState of nodeUserStates) {
+    const deviceIds = userDevices.get(nodeUserState.user_id) ?? [];
+    if (deviceIds.length === 0) {
      continue;
    }

-    if (nodeVersion.access_removed_at !== null || !node) {
-      socketConnection.send({
-        type: 'server_node_delete',
-        id: data.nodeId,
-        workspaceId: data.workspaceId,
-      });
-    } else {
-      socketConnection.send({
-        type: 'server_node_sync',
-        id: node.id,
-        workspaceId: data.workspaceId,
-        state: node.state!,
-        createdAt: node.created_at.toISOString(),
-        createdBy: node.created_by,
-        updatedAt: node.updated_at?.toISOString() ?? null,
-        updatedBy: node.updated_by ?? null,
-        serverCreatedAt: node.server_created_at.toISOString(),
-        serverUpdatedAt: node.server_updated_at?.toISOString() ?? null,
-        versionId: node.version_id,
-      });
+    for (const deviceId of deviceIds) {
+      const socketConnection = socketManager.getConnection(deviceId);
+      if (socketConnection === undefined) {
+        continue;
+      }
+
+      if (nodeUserState.access_removed_at !== null) {
+        socketConnection.send({
+          type: 'server_node_delete',
+          id: data.nodeId,
+          workspaceId: data.workspaceId,
+        });
+      } else {
+        socketConnection.send({
+          type: 'server_node_sync',
+          id: node.id,
+          workspaceId: data.workspaceId,
+          state: node.state!,
+          createdAt: node.created_at.toISOString(),
+          createdBy: node.created_by,
+          updatedAt: node.updated_at?.toISOString() ?? null,
+          updatedBy: node.updated_by ?? null,
+          serverCreatedAt: node.server_created_at.toISOString(),
+          serverUpdatedAt: node.server_updated_at?.toISOString() ?? null,
+          versionId: node.version_id,
+        });
+      }
    }
  }
 };
--- a/server/src/data/migrations.ts
+++ b/server/src/data/migrations.ts
@@ -20,6 +20,32 @@ const createAccountsTable: Migration = {
  },
 };

+const createDevicesTable: Migration = {
+  up: async (db) => {
+    await db.schema
+      .createTable('devices')
+      .addColumn('id', 'varchar(30)', (col) => col.notNull().primaryKey())
+      .addColumn('account_id', 'varchar(30)', (col) => col.notNull())
+      .addColumn('token_hash', 'varchar(100)', (col) => col.notNull())
+      .addColumn('token_salt', 'varchar(100)', (col) => col.notNull())
+      .addColumn('token_generated_at', 'timestamptz', (col) => col.notNull())
+      .addColumn('previous_token_hash', 'varchar(100)')
+      .addColumn('previous_token_salt', 'varchar(100)')
+      .addColumn('type', 'integer', (col) => col.notNull())
+      .addColumn('version', 'varchar(30)', (col) => col.notNull())
+      .addColumn('platform', 'varchar(30)')
+      .addColumn('cpu', 'varchar(30)')
+      .addColumn('hostname', 'varchar(30)')
+      .addColumn('created_at', 'timestamptz', (col) => col.notNull())
+      .addColumn('last_online_at', 'timestamptz')
+      .addColumn('last_active_at', 'timestamptz')
+      .execute();
+  },
+  down: async (db) => {
+    await db.schema.dropTable('devices').execute();
+  },
+};
+
 const createWorkspacesTable: Migration = {
  up: async (db) => {
    await db.schema
@@ -255,62 +281,60 @@ const createNodeCollaboratorsTable: Migration = {
  },
 };

-const createDevicesTable: Migration = {
+const createNodeUserStatesTable: Migration = {
  up: async (db) => {
    await db.schema
-      .createTable('devices')
-      .addColumn('id', 'varchar(30)', (col) => col.notNull().primaryKey())
-      .addColumn('account_id', 'varchar(30)', (col) => col.notNull())
-      .addColumn('token_hash', 'varchar(100)', (col) => col.notNull())
-      .addColumn('token_salt', 'varchar(100)', (col) => col.notNull())
-      .addColumn('token_generated_at', 'timestamptz', (col) => col.notNull())
-      .addColumn('previous_token_hash', 'varchar(100)')
-      .addColumn('previous_token_salt', 'varchar(100)')
-      .addColumn('type', 'integer', (col) => col.notNull())
-      .addColumn('version', 'varchar(30)', (col) => col.notNull())
-      .addColumn('platform', 'varchar(30)')
-      .addColumn('cpu', 'varchar(30)')
-      .addColumn('hostname', 'varchar(30)')
+      .createTable('node_user_states')
+      .addColumn('node_id', 'varchar(30)', (col) => col.notNull())
+      .addColumn('user_id', 'varchar(30)', (col) => col.notNull())
+      .addColumn('workspace_id', 'varchar(30)', (col) => col.notNull())
+      .addColumn('last_seen_version_id', 'varchar(30)')
+      .addColumn('last_seen_at', 'timestamptz')
+      .addColumn('mentions_count', 'integer', (col) =>
+        col.notNull().defaultTo(0),
+      )
      .addColumn('created_at', 'timestamptz', (col) => col.notNull())
-      .addColumn('last_online_at', 'timestamptz')
-      .addColumn('last_active_at', 'timestamptz')
+      .addColumn('updated_at', 'timestamptz')
+      .addColumn('access_removed_at', 'timestamptz')
+      .addColumn('version_id', 'varchar(30)', (col) => col.notNull())
+      .addPrimaryKeyConstraint('node_user_states_pkey', ['node_id', 'user_id'])
      .execute();
  },
  down: async (db) => {
-    await db.schema.dropTable('devices').execute();
+    await db.schema.dropTable('node_user_states').execute();
  },
 };

-const createDeviceNodeVersionTable: Migration = {
+const createNodeDeviceStatesTable: Migration = {
  up: async (db) => {
    await db.schema
-      .createTable('device_node_versions')
-      .addColumn('device_id', 'varchar(30)', (col) =>
-        col.notNull().references('devices.id').onDelete('cascade'),
-      )
+      .createTable('node_device_states')
      .addColumn('node_id', 'varchar(30)', (col) => col.notNull())
+      .addColumn('device_id', 'varchar(30)', (col) => col.notNull())
      .addColumn('workspace_id', 'varchar(30)', (col) => col.notNull())
-      .addColumn('version_id', 'varchar(30)', (col) => col.notNull())
-      .addColumn('synced_at', 'timestamptz')
-      .addColumn('access_removed_at', 'timestamptz')
-      .addPrimaryKeyConstraint('device_node_versions_pkey', [
-        'device_id',
+      .addColumn('node_version_id', 'varchar(30)')
+      .addColumn('user_state_version_id', 'varchar(30)')
+      .addColumn('node_synced_at', 'timestamptz')
+      .addColumn('user_state_synced_at', 'timestamptz')
+      .addPrimaryKeyConstraint('node_device_states_pkey', [
        'node_id',
+        'device_id',
      ])
      .execute();
  },
  down: async (db) => {
-    await db.schema.dropTable('device_node_versions').execute();
+    await db.schema.dropTable('node_device_states').execute();
  },
 };

 export const databaseMigrations: Record<string, Migration> = {
  '00001_create_accounts_table': createAccountsTable,
-  '00002_create_workspaces_table': createWorkspacesTable,
-  '00003_create_workspace_users_table': createWorkspaceUsersTable,
-  '00004_create_nodes_table': createNodesTable,
-  '00005_create_node_paths_table': createNodePathsTable,
-  '00006_create_node_collaborators_table': createNodeCollaboratorsTable,
-  '00007_create_devices_table': createDevicesTable,
-  '00008_create_device_node_version_table': createDeviceNodeVersionTable,
+  '00002_create_devices_table': createDevicesTable,
+  '00003_create_workspaces_table': createWorkspacesTable,
+  '00004_create_workspace_users_table': createWorkspaceUsersTable,
+  '00005_create_nodes_table': createNodesTable,
+  '00006_create_node_paths_table': createNodePathsTable,
+  '00007_create_node_collaborators_table': createNodeCollaboratorsTable,
+  '00008_create_node_user_states_table': createNodeUserStatesTable,
+  '00009_create_node_device_states_table': createNodeDeviceStatesTable,
 };
--- a/server/src/data/schema.ts
+++ b/server/src/data/schema.ts
@@ -23,6 +23,28 @@ export type SelectAccount = Selectable<AccountTable>;
 export type CreateAccount = Insertable<AccountTable>;
 export type UpdateAccount = Updateable<AccountTable>;

+interface DeviceTable {
+  id: ColumnType<string, string, never>;
+  account_id: ColumnType<string, string, never>;
+  token_hash: ColumnType<string, string, string>;
+  token_salt: ColumnType<string, string, string>;
+  token_generated_at: ColumnType<Date, Date, Date>;
+  previous_token_hash: ColumnType<string | null, string | null, string | null>;
+  previous_token_salt: ColumnType<string | null, string | null, string | null>;
+  type: ColumnType<number, number, number>;
+  version: ColumnType<string, string, string>;
+  platform: ColumnType<string | null, string | null, string | null>;
+  cpu: ColumnType<string | null, string | null, string | null>;
+  hostname: ColumnType<string | null, string | null, string | null>;
+  created_at: ColumnType<Date, Date, never>;
+  last_online_at: ColumnType<Date | null, Date | null, Date>;
+  last_active_at: ColumnType<Date | null, Date | null, Date>;
+}
+
+export type SelectDevice = Selectable<DeviceTable>;
+export type CreateDevice = Insertable<DeviceTable>;
+export type UpdateDevice = Updateable<DeviceTable>;
+
 interface WorkspaceTable {
  id: ColumnType<string, string, never>;
  name: ColumnType<string, string, string>;
@@ -106,48 +128,45 @@ export type SelectNodeCollaborator = Selectable<NodeCollaboratorTable>;
 export type CreateNodeCollaborator = Insertable<NodeCollaboratorTable>;
 export type UpdateNodeCollaborator = Updateable<NodeCollaboratorTable>;

-interface DeviceTable {
-  id: ColumnType<string, string, never>;
-  account_id: ColumnType<string, string, never>;
-  token_hash: ColumnType<string, string, string>;
-  token_salt: ColumnType<string, string, string>;
-  token_generated_at: ColumnType<Date, Date, Date>;
-  previous_token_hash: ColumnType<string | null, string | null, string | null>;
-  previous_token_salt: ColumnType<string | null, string | null, string | null>;
-  type: ColumnType<number, number, number>;
-  version: ColumnType<string, string, string>;
-  platform: ColumnType<string | null, string | null, string | null>;
-  cpu: ColumnType<string | null, string | null, string | null>;
-  hostname: ColumnType<string | null, string | null, string | null>;
-  created_at: ColumnType<Date, Date, never>;
-  last_online_at: ColumnType<Date | null, Date | null, Date>;
-  last_active_at: ColumnType<Date | null, Date | null, Date>;
-}
-
-export type SelectDevice = Selectable<DeviceTable>;
-export type CreateDevice = Insertable<DeviceTable>;
-export type UpdateDevice = Updateable<DeviceTable>;
-
-interface DeviceNodeVersionTable {
-  device_id: ColumnType<string, string, never>;
+interface NodeUserStateTable {
  node_id: ColumnType<string, string, never>;
-  version_id: ColumnType<string, string, string>;
+  user_id: ColumnType<string, string, never>;
  workspace_id: ColumnType<string, string, never>;
-  synced_at: ColumnType<Date | null, Date | null, Date>;
+  last_seen_version_id: ColumnType<string | null, string | null, string | null>;
+  last_seen_at: ColumnType<Date | null, Date | null, Date>;
+  mentions_count: ColumnType<number, number, number>;
+  created_at: ColumnType<Date, Date, never>;
+  updated_at: ColumnType<Date | null, Date | null, Date>;
  access_removed_at: ColumnType<Date | null, Date | null, Date>;
+  version_id: ColumnType<string, string, string>;
 }

-export type SelectDeviceNodeVersion = Selectable<DeviceNodeVersionTable>;
-export type CreateDeviceNodeVersion = Insertable<DeviceNodeVersionTable>;
-export type UpdateDeviceNodeVersion = Updateable<DeviceNodeVersionTable>;
+export type SelectNodeUserState = Selectable<NodeUserStateTable>;
+export type CreateNodeUserState = Insertable<NodeUserStateTable>;
+export type UpdateNodeUserState = Updateable<NodeUserStateTable>;
+
+interface NodeDeviceStateTable {
+  node_id: ColumnType<string, string, never>;
+  device_id: ColumnType<string, string, never>;
+  workspace_id: ColumnType<string, string, string>;
+  node_version_id: ColumnType<string | null, string | null, string | null>;
+  user_state_version_id: ColumnType<
+    string | null,
+    string | null,
+    string | null
+  >;
+  node_synced_at: ColumnType<Date | null, Date | null, Date>;
+  user_state_synced_at: ColumnType<Date | null, Date | null, Date>;
+}

 export interface DatabaseSchema {
  accounts: AccountTable;
+  devices: DeviceTable;
  workspaces: WorkspaceTable;
  workspace_users: WorkspaceUserTable;
  nodes: NodeTable;
  node_paths: NodePathTable;
  node_collaborators: NodeCollaboratorTable;
-  devices: DeviceTable;
-  device_node_versions: DeviceNodeVersionTable;
+  node_user_states: NodeUserStateTable;
+  node_device_states: NodeDeviceStateTable;
 }
--- a/server/src/queues/events.ts
+++ b/server/src/queues/events.ts
@@ -1,7 +1,8 @@
 import { database } from '@/data/database';
 import { CHANNEL_NAMES, redis } from '@/data/redis';
-import { CreateDeviceNodeVersion } from '@/data/schema';
+import { CreateNodeUserState } from '@/data/schema';
 import { NodeTypes } from '@/lib/constants';
+import { generateId, IdType } from '@/lib/id';
 import { fetchNodeCollaborators, fetchWorkspaceUsers } from '@/lib/nodes';
 import {
  NodeCreatedEvent,
@@ -79,17 +80,20 @@ const handleNodeCreatedEvent = async (
    return;
  }

-  const deviceIds = await getDeviceIdsForCollaborators(collaboratorIds);
  await database
-    .insertInto('device_node_versions')
+    .insertInto('node_user_states')
    .values(
-      deviceIds.map((deviceId) => ({
-        device_id: deviceId,
+      collaboratorIds.map((collaboratorId) => ({
+        user_id: collaboratorId,
        node_id: event.id,
-        version_id: 'null',
+        last_seen_version_id: null,
        workspace_id: event.workspaceId,
-        synced_at: null,
+        last_seen_at: null,
+        mentions_count: 0,
+        created_at: new Date(),
        access_removed_at: null,
+        version_id: generateId(IdType.Version),
+        updated_at: null,
      })),
    )
    .onConflict((cb) => cb.doNothing())
@@ -115,7 +119,6 @@ const handleNodeUpdatedEvent = async (
  );

  if (addedCollaborators.length > 0) {
-    const deviceIds = await getDeviceIdsForCollaborators(addedCollaborators);
    const descendants = await database
      .selectFrom('node_paths')
      .select('descendant_id')
@@ -123,40 +126,41 @@ const handleNodeUpdatedEvent = async (
      .execute();

    const descendantIds = descendants.map((d) => d.descendant_id);
-    const deviceNodeVersionsToCreated: CreateDeviceNodeVersion[] = [];
-    for (const deviceId of deviceIds) {
+    const userStatesToCreated: CreateNodeUserState[] = [];
+    for (const collaboratorId of addedCollaborators) {
      for (const descendantId of descendantIds) {
-        deviceNodeVersionsToCreated.push({
-          device_id: deviceId,
+        userStatesToCreated.push({
+          user_id: collaboratorId,
          node_id: descendantId,
-          version_id: 'null',
+          last_seen_version_id: null,
          workspace_id: event.workspaceId,
-          synced_at: null,
+          last_seen_at: null,
+          mentions_count: 0,
+          created_at: new Date(),
          access_removed_at: null,
+          version_id: generateId(IdType.Version),
        });
      }
    }

-    if (deviceNodeVersionsToCreated.length > 0) {
+    if (userStatesToCreated.length > 0) {
      await database
-        .insertInto('device_node_versions')
-        .values(deviceNodeVersionsToCreated)
+        .insertInto('node_user_states')
+        .values(userStatesToCreated)
        .onConflict((cb) => cb.doNothing())
        .execute();
    }
  }

  if (removedCollaborators.length > 0) {
-    const deviceIds = await getDeviceIdsForCollaborators(removedCollaborators);
-
    await database
-      .updateTable('device_node_versions')
+      .updateTable('node_user_states')
      .set({
        access_removed_at: new Date(),
      })
      .where((eb) =>
        eb.and([
-          eb('device_id', 'in', deviceIds),
+          eb('user_id', 'in', removedCollaborators),
          eb(
            'node_id',
            'in',
@@ -187,19 +191,6 @@ const handleNodeDeletedEvent = async (
  await publishChange(event.id, event.workspaceId);
 };

-const getDeviceIdsForCollaborators = async (
-  collaboratorIds: string[],
-): Promise<string[]> => {
-  const devices = await database
-    .selectFrom('devices as d')
-    .select('d.id')
-    .innerJoin('workspace_users as wu', 'd.account_id', 'wu.account_id')
-    .where('wu.id', 'in', collaboratorIds)
-    .execute();
-
-  return devices.map((d) => d.id);
-};
-
 const publishChange = async (
  nodeId: string,
  workspaceId: string,
--- a/server/src/sockets/socket-connection.ts
+++ b/server/src/sockets/socket-connection.ts
@@ -3,15 +3,22 @@ import { MessageInput } from '@/messages';
 import { NeuronRequestAccount } from '@/types/api';
 import { WebSocket } from 'ws';

+interface WorkspaceUser {
+  workspaceId: string;
+  userId: string;
+}
+
 export class SocketConnection {
  private readonly socket: WebSocket;
  private readonly accountId: string;
  private readonly deviceId: string;
+  private readonly workspaceUsers: WorkspaceUser[];

  constructor(socket: WebSocket, account: NeuronRequestAccount) {
    this.socket = socket;
    this.accountId = account.id;
    this.deviceId = account.deviceId;
+    this.workspaceUsers = [];

    socket.on('message', (message) => {
      this.handleMessage(message.toString());
@@ -23,40 +30,85 @@ export class SocketConnection {
  }

  public init(): void {
-    this.sendPendingChanges();
+    this.fetchWorkspaceUsers().then(() => {
+      this.sendPendingChanges();
+    });
  }

  public getDeviceId(): string {
    return this.deviceId;
  }

+  public getWorkspaceUsers(): WorkspaceUser[] {
+    return this.workspaceUsers;
+  }
+
+  private async fetchWorkspaceUsers(): Promise<void> {
+    const workspaceUsers = await database
+      .selectFrom('workspace_users')
+      .selectAll()
+      .where('account_id', '=', this.accountId)
+      .execute();
+
+    for (const workspaceUser of workspaceUsers) {
+      this.workspaceUsers.push({
+        workspaceId: workspaceUser.workspace_id,
+        userId: workspaceUser.id,
+      });
+    }
+  }
+
  private async handleMessage(message: string): Promise<void> {
    const messageInput: MessageInput = JSON.parse(message);
    console.log(messageInput);
    if (messageInput.type === 'local_node_sync') {
      await database
-        .updateTable('device_node_versions')
-        .set({
-          version_id: messageInput.versionId,
-          synced_at: new Date(),
+        .insertInto('node_device_states')
+        .values({
+          node_id: messageInput.nodeId,
+          device_id: this.deviceId,
+          node_version_id: messageInput.versionId,
+          user_state_version_id: null,
+          user_state_synced_at: null,
+          workspace_id: messageInput.workspaceId,
+          node_synced_at: new Date(),
        })
-        .where('device_id', '=', this.deviceId)
-        .where('node_id', '=', messageInput.nodeId)
+        .onConflict((cb) =>
+          cb.columns(['node_id', 'device_id']).doUpdateSet({
+            workspace_id: messageInput.workspaceId,
+            node_version_id: messageInput.versionId,
+            node_synced_at: new Date(),
+          }),
+        )
        .execute();
    } else if (messageInput.type === 'local_node_delete') {
      await database
-        .deleteFrom('device_node_versions')
+        .deleteFrom('node_device_states')
        .where('device_id', '=', this.deviceId)
        .where('node_id', '=', messageInput.nodeId)
-        .where('workspace_id', '=', messageInput.workspaceId)
        .execute();
    }
  }

  private async sendPendingChanges() {
+    const userIds = this.workspaceUsers.map(
+      (workspaceUser) => workspaceUser.userId,
+    );
+
+    if (userIds.length === 0) {
+      return;
+    }
+
+    console.log('userIds', userIds);
+
    const unsyncedNodes = await database
-      .selectFrom('device_node_versions as dnv')
-      .leftJoin('nodes as n', 'n.id', 'dnv.node_id')
+      .selectFrom('node_user_states as nus')
+      .leftJoin('nodes as n', 'n.id', 'nus.node_id')
+      .leftJoin('node_device_states as nds', (join) =>
+        join
+          .onRef('nds.node_id', '=', 'n.id')
+          .on('nds.device_id', '=', this.deviceId),
+      )
      .select([
        'n.id',
        'n.state',
@@ -66,18 +118,19 @@ export class SocketConnection {
        'n.updated_by',
        'n.server_created_at',
        'n.server_updated_at',
-        'dnv.access_removed_at',
+        'nus.access_removed_at',
        'n.version_id',
-        'dnv.node_id',
-        'dnv.workspace_id',
+        'nus.node_id',
+        'nus.workspace_id',
      ])
      .where((eb) =>
        eb.and([
-          eb('dnv.device_id', '=', this.deviceId),
+          eb('nus.user_id', 'in', userIds),
          eb.or([
            eb('n.id', 'is', null),
-            eb('dnv.version_id', '!=', eb.ref('n.version_id')),
-            eb('dnv.access_removed_at', 'is not', null),
+            eb('nds.node_version_id', 'is', null),
+            eb('nds.node_version_id', '!=', eb.ref('n.version_id')),
+            eb('nus.access_removed_at', 'is not', null),
          ]),
        ]),
      )
@@ -85,6 +138,8 @@ export class SocketConnection {
      .limit(100)
      .execute();

+    console.log('unsyncedNodes', unsyncedNodes);
+
    if (unsyncedNodes.length === 0) {
      return;
    }