feat(entities): Implement watch() method for live query subscriptions

This adds a watch() method to EntityHandler that allows subscribing to
filtered subsets of entities with real-time change notifications.

Features:
- watch(options, callback) method with filter, sort, fields, limit options
- WatchEvent with changeType ('added', 'modified', 'removed')
- subscribeQuery/unsubscribeQuery socket events for server communication
- Proper cleanup on unsubscribe

The watch() method differs from subscribe() in that:
- It supports filter, sort, fields, and limit options
- It returns changeType indicating how the change affects the filtered results
- It uses dedicated subscribe_query socket events

Example usage:
```typescript
const unsubscribe = base44.entities.Task.watch(
  { filter: { status: 'active' }, sort: '-created_date', limit: 10 },
  (event) => {
    if (event.changeType === 'added') {
      console.log('New matching task:', event.data);
    }
  }
);
```

Fixes #83

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: guyofeck

src/modules/entities.ts

@@ -5,6 +5,10 @@ import {
   RealtimeCallback,
   RealtimeEvent,
   RealtimeEventType,
+  WatchCallback,
+  WatchChangeType,
+  WatchEvent,
+  WatchOptions,
 } from "./entities.types";
 import { RoomsSocket } from "../utils/socket-utils.js";
 
@@ -69,6 +73,26 @@ function parseRealtimeMessage(dataStr: string): RealtimeEvent | null {
   }
 }
 
+/**
+ * Parses the watch (live query) message data and extracts event information.
+ * @internal
+ */
+function parseWatchMessage(dataStr: string): WatchEvent | null {
+  try {
+    const parsed = JSON.parse(dataStr);
+    return {
+      changeType: parsed.change_type as WatchChangeType,
+      eventType: parsed.type as RealtimeEventType,
+      data: parsed.data,
+      id: parsed.id || parsed.data?.id,
+      timestamp: parsed.timestamp || new Date().toISOString(),
+    };
+  } catch (error) {
+    console.warn("[Base44 SDK] Failed to parse watch message:", error);
+    return null;
+  }
+}
+
 /**
  * Creates a handler for a specific entity.
  *
@@ -186,5 +210,38 @@ function createEntityHandler(
 
       return unsubscribe;
     },
+
+    // Watch for changes to a filtered subset (live query)
+    watch(options: WatchOptions, callback: WatchCallback): () => void {
+      const socket = getSocket();
+
+      // Use subscribeQuery to send subscription options to the server
+      const unsubscribe = socket.subscribeQuery(
+        appId,
+        entityName,
+        {
+          filter: options.filter,
+          sort: options.sort,
+          fields: options.fields,
+          limit: options.limit,
+        },
+        {
+          update_model: (msg) => {
+            const event = parseWatchMessage(msg.data);
+            if (!event) {
+              return;
+            }
+
+            try {
+              callback(event);
+            } catch (error) {
+              console.error("[Base44 SDK] Watch callback error:", error);
+            }
+          },
+        }
+      );
+
+      return unsubscribe;
+    },
   };
 }

src/modules/entities.types.ts

@@ -3,6 +3,14 @@
  */
 export type RealtimeEventType = "create" | "update" | "delete";
 
+/**
+ * Change types for live query (watch) subscriptions.
+ * - "added": Entity now matches the filter but didn't before (or was created matching)
+ * - "modified": Entity matched before and still matches
+ * - "removed": Entity matched before but no longer matches (or was deleted)
+ */
+export type WatchChangeType = "added" | "modified" | "removed";
+
 /**
  * Payload received when a realtime event occurs.
  */
@@ -17,6 +25,41 @@ export interface RealtimeEvent {
   timestamp: string;
 }
 
+/**
+ * Payload received when a watch (live query) event occurs.
+ */
+export interface WatchEvent {
+  /** The type of change relative to the subscription filter */
+  changeType: WatchChangeType;
+  /** The CUD event type that triggered this change */
+  eventType: RealtimeEventType;
+  /** The entity data after the change */
+  data: any;
+  /** The unique identifier of the affected entity */
+  id: string;
+  /** ISO 8601 timestamp of when the event occurred */
+  timestamp: string;
+}
+
+/**
+ * Options for watch (live query) subscriptions.
+ */
+export interface WatchOptions {
+  /** MongoDB-style filter query */
+  filter?: Record<string, any>;
+  /** Sort field with optional '-' prefix for descending */
+  sort?: string;
+  /** Array of field names to include in the response */
+  fields?: string[];
+  /** Maximum number of results */
+  limit?: number;
+}
+
+/**
+ * Callback function invoked when a watch (live query) event occurs.
+ */
+export type WatchCallback = (event: WatchEvent) => void;
+
 /**
  * Callback function invoked when a realtime event occurs.
  */
@@ -312,6 +355,64 @@ export interface EntityHandler {
    * ```
    */
   subscribe(callback: RealtimeCallback): () => void;
+
+  /**
+   * Watches for changes to a filtered subset of records (live query).
+   *
+   * Similar to `subscribe`, but allows you to specify filter, sort, fields,
+   * and limit options. The callback receives events with a `changeType` that
+   * indicates how the change affects the filtered result set:
+   * - `'added'`: A record now matches your filter (created or updated to match)
+   * - `'modified'`: A record still matches your filter but was updated
+   * - `'removed'`: A record no longer matches your filter (deleted or updated to not match)
+   *
+   * @param options - Options for the watch subscription:
+   * - `filter`: MongoDB-style filter query to match records
+   * - `sort`: Sort field with optional '-' prefix for descending order
+   * - `fields`: Array of field names to include in the response
+   * - `limit`: Maximum number of records to track
+   * @param callback - Callback function called when a matching entity changes. The callback receives an event object with:
+   * - `changeType`: How the change affects your filtered results - `'added'`, `'modified'`, or `'removed'`
+   * - `eventType`: The underlying CUD operation - `'create'`, `'update'`, or `'delete'`
+   * - `data`: The entity data after the change
+   * - `id`: The unique identifier of the affected entity
+   * - `timestamp`: ISO 8601 timestamp of when the event occurred
+   * @returns Unsubscribe function to stop receiving updates.
+   *
+   * @example
+   * ```typescript
+   * // Watch for changes to active high-priority tasks
+   * const unsubscribe = base44.entities.Task.watch(
+   *   {
+   *     filter: { status: 'active', priority: 'high' },
+   *     sort: '-created_date',
+   *     limit: 10
+   *   },
+   *   (event) => {
+   *     if (event.changeType === 'added') {
+   *       console.log('New high-priority task:', event.data);
+   *     } else if (event.changeType === 'removed') {
+   *       console.log('Task no longer high-priority:', event.id);
+   *     }
+   *   }
+   * );
+   *
+   * // Later, clean up the subscription
+   * unsubscribe();
+   * ```
+   *
+   * @example
+   * ```typescript
+   * // Watch for changes to current user's tasks
+   * const unsubscribe = base44.entities.Task.watch(
+   *   { filter: { assignee: currentUser.id } },
+   *   (event) => {
+   *     console.log(`My task ${event.id} was ${event.changeType}:`, event.data);
+   *   }
+   * );
+   * ```
+   */
+  watch(options: WatchOptions, callback: WatchCallback): () => void;
 }
 
 /**

src/utils/socket-utils.ts

@@ -12,18 +12,46 @@ export interface RoomsSocketConfig {
 export type TSocketRoom = string;
 export type TJsonStr = string;
 
+/**
+ * Options for watch (live query) subscriptions.
+ */
+export interface WatchSubscriptionOptions {
+  filter?: Record<string, any>;
+  sort?: string;
+  fields?: string[];
+  limit?: number;
+}
+
 type RoomsSocketEventsMap = {
   listen: {
     connect: () => Promise<void> | void;
     update_model: (msg: {
       room: string;
       data: TJsonStr;
     }) => Promise<void> | void;
+    subscribed: (msg: {
+      room: string;
+      entity_name: string;
+      options: WatchSubscriptionOptions;
+    }) => Promise<void> | void;
+    unsubscribed: (msg: {
+      room: string;
+      entity_name: string;
+    }) => Promise<void> | void;
     error: (error: Error) => Promise<void> | void;
   };
   emit: {
     join: (room: string) => void;
     leave: (room: string) => void;
+    subscribe_query: (data: {
+      app_id: string;
+      entity_name: string;
+      options: WatchSubscriptionOptions;
+    }) => void;
+    unsubscribe_query: (data: {
+      app_id: string;
+      entity_name: string;
+    }) => void;
   };
 };
 
@@ -53,6 +81,14 @@ function initializeSocket(
     return handlers.update_model?.(msg);
   });
 
+  socket.on("subscribed", async (msg) => {
+    return handlers.subscribed?.(msg);
+  });
+
+  socket.on("unsubscribed", async (msg) => {
+    return handlers.unsubscribed?.(msg);
+  });
+
   socket.on("error", async (error) => {
     return handlers.error?.(error);
   });
@@ -162,9 +198,53 @@ export function RoomsSocket({ config }: { config: RoomsSocketConfig }) {
     };
   };
 
+  /**
+   * Subscribe to a live query with filter, sort, fields, and limit options.
+   * This sends subscribe_query to the server and sets up listeners for updates.
+   */
+  const subscribeQuery = (
+    appId: string,
+    entityName: string,
+    options: WatchSubscriptionOptions,
+    handlers: Partial<{ [k in TEvent]: THandler<k> }>
+  ) => {
+    // The room name matches the backend format
+    const room = `entities:${appId}:${entityName}:watch`;
+
+    // Add handlers for this room
+    if (!roomsToListeners[room]) {
+      roomsToListeners[room] = [];
+    }
+    roomsToListeners[room].push(handlers);
+
+    // Send subscribe_query event to server
+    socket.emit("subscribe_query", {
+      app_id: appId,
+      entity_name: entityName,
+      options,
+    });
+
+    // Return unsubscribe function
+    return () => {
+      roomsToListeners[room] =
+        roomsToListeners[room]?.filter((listener) => listener !== handlers) ??
+        [];
+
+      if (roomsToListeners[room].length === 0) {
+        // Send unsubscribe_query event to server
+        socket.emit("unsubscribe_query", {
+          app_id: appId,
+          entity_name: entityName,
+        });
+        delete roomsToListeners[room];
+      }
+    };
+  };
+
   return {
     socket,
     subscribeToRoom,
+    subscribeQuery,
     updateConfig,
     updateModel,
     disconnect,