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

[guyofeck]

Summary

• Add watch(options, callback) method to EntityHandler for live query subscriptions
• Add WatchEvent type with changeType ('added', 'modified', 'removed')
• Add WatchOptions type for filter, sort, fields, limit configuration
• Add subscribeQuery/unsubscribeQuery socket events for server communication

Usage

const unsubscribe = base44.entities.Task.watch(
  {
    filter: { status: 'active', priority: 'high' },
    sort: '-created_date',
    limit: 10
  },
  (event) => {
    console.log(`Change type: ${event.changeType}`); // 'added', 'modified', 'removed'
    console.log(`Event type: ${event.eventType}`);   // 'create', 'update', 'delete'
    console.log(`Data:`, event.data);
  }
);

// Later, clean up
unsubscribe();

Test plan

• Unit tests for watch method
• Tests for change type parsing
• Tests for error handling

Depends on: Backend PRs #2894, #2895, #2896

Fixes #83

🤖 Generated with Claude Code

[claude]

Claude finished @guyofeck's task —— View job

---

Code Review for PR #86: feat(entities): Implement watch() method for live query subscriptions

I've thoroughly reviewed the implementation of the watch() method for live query subscriptions. Here's my comprehensive feedback:

✅ Code Quality and Best Practices

Strengths:

• Clean and consistent API design that aligns well with the existing subscribe() method
• Excellent TypeScript typing with clear interfaces and type definitions
• Good separation of concerns between entities, socket utilities, and types
• Follows existing code conventions and patterns in the codebase

Minor Improvements:

• Consider using an enum for WatchChangeType instead of string literals for better type safety:

  export enum WatchChangeType {
    ADDED = "added",
    MODIFIED = "modified", 
    REMOVED = "removed"
  }

✅ Potential Bugs or Issues

No critical bugs found, but a few considerations:

1. Room naming convention - The watch room uses a different pattern (entities:${appId}:${entityName}:watch) compared to the regular subscribe room (entities:${appId}:${entityName}). Ensure the backend is expecting this exact format.

2. Error recovery - While errors are logged to console, there's no mechanism for the consumer to handle connection errors or subscription failures. Consider adding an error callback option:

   interface WatchOptions {
     // ... existing fields
     onError?: (error: Error) => void;
   }

3. Missing timestamp fallback - Good that you provide a fallback timestamp when missing, but consider if the backend should always provide this.

⚠️ Performance Considerations

1. Memory management - The current implementation stores handlers in roomsToListeners. With many active subscriptions, this could grow. Consider:

   • Adding a maximum subscription limit per entity
   • Implementing subscription cleanup on connection loss
   • Warning when subscription count is high

2. Message parsing efficiency - Both parseWatchMessage and parseRealtimeMessage are nearly identical. Consider consolidating or using a generic parser to reduce code duplication.

3. Subscription lifecycle - No automatic reconnection logic. If the socket disconnects, subscriptions may need manual re-establishment.

✅ Security Concerns

No major security issues identified. Good practices observed:

• Uses existing authentication token from getAccessToken()
• No sensitive data exposure in error logs
• Proper data sanitization through JSON parsing

Recommendation: Ensure the backend validates subscription filters to prevent potential query injection or excessive resource consumption from complex filters.

✅ Test Coverage

Excellent test coverage with comprehensive unit tests covering:

• ✅ Basic subscription setup
• ✅ All change types (added, modified, removed)
• ✅ Error handling (callback errors, invalid JSON)
• ✅ Unsubscribe functionality

Additional test suggestions:

1. Test concurrent subscriptions to the same entity
2. Test subscription with all optional parameters (fields, limit)
3. Test subscription cleanup when socket disconnects
4. Integration tests with actual socket connection (if feasible)

📝 Documentation

The inline documentation is excellent and comprehensive. The examples clearly demonstrate usage patterns. Consider adding:

• A note about subscription limits (if any)
• Guidance on when to use watch() vs subscribe()
• Best practices for filter performance

🎯 Overall Assessment

This is a well-implemented feature that cleanly extends the existing entities module. The code is production-ready with only minor improvements suggested. The implementation follows established patterns and provides a intuitive API for developers.

Recommendation: Approved with minor suggestions ✅

Great work on this feature! The live query functionality will be very useful for real-time applications.

[guyofeck]

@claude the callback should get the snapshot of all of the entities that returns from the query, and not a specific entity

[claude]

Claude finished @guyofeck's task —— View job

---

Todo List:

• Review current watch() implementation
• Analyze the callback behavior
• Provide feedback on the architectural difference

---

Analysis of the watch() Implementation

I've reviewed the current implementation, and I understand the concern. Looking at the code and tests, the current watch() method is designed to receive individual entity change events rather than a snapshot of all entities matching the query.

Current Implementation Behavior

The current implementation:

1. Subscribes to a filtered query with options (filter, sort, fields, limit)
2. Receives individual change events for entities that match the filter
3. Each callback invocation gets a single WatchEvent with:
   • changeType: 'added' | 'modified' | 'removed'
   • eventType: 'create' | 'update' | 'delete'
   • data: The single entity that changed
   • id: The ID of that specific entity

Architectural Implications

If the callback should receive a snapshot of all entities matching the query (as you're suggesting), this would be a significant architectural change:

1. Different Event Model: Instead of individual entity events, each callback would receive the entire result set
2. Performance Considerations: Sending full snapshots on every change could be bandwidth-intensive
3. State Management: The client would need to diff snapshots to understand what changed
4. Backend Requirements: The backend would need to:
   • Maintain the query state for each subscription
   • Re-execute the query on every relevant change
   • Send the complete result set each time

Potential Implementation Approaches

If you want to implement snapshot-based updates, here are two approaches:

Option 1: Full Snapshot on Every Change

interface WatchSnapshot {
  entities: any[];  // All entities matching the query
  changeType: 'added' | 'modified' | 'removed';
  changedEntityId: string;  // Which entity triggered this update
  timestamp: string;
}

Option 2: Initial Snapshot + Individual Updates

// Initial callback with full snapshot
callback({ type: 'snapshot', entities: [...] });

// Subsequent callbacks with individual changes
callback({ type: 'change', changeType: 'added', entity: {...} });

Questions for Clarification

1. Should the initial subscription immediately return the current snapshot?
2. On each change, should the callback receive the entire filtered dataset?
3. How should sorting and limiting work with snapshots? (e.g., if limit=10, does removing one entity cause a new 10th entity to appear?)
4. What's the expected behavior for large result sets?

The current implementation follows a more traditional event-based pattern (similar to Firebase Realtime Database), while what you're describing sounds more like a reactive query pattern (similar to MongoDB Change Streams with full document).

Would you like me to suggest specific implementation changes for the snapshot-based approach?