⚡ Firestore Realtime - Advanced

Introduction

Advanced real-time patterns enable sophisticated collaborative applications, presence tracking, and operational transformation for multi-user editing.

Key Learning Outcomes

By the end of this lesson, you'll understand:

• Advanced listener patterns and optimization

• Presence detection and user awareness

• Conflict resolution strategies

• Operational transformation for collaborative editing

• Subscription management at scale

• Real-time aggregations and denormalization

• Offline-first architectures

Advanced Listener Patterns

Managed Listener Lifecycle

class FirestoreListenerManager {
  constructor() {
    this.listeners = new Map();
    this.subscriptions = new Map();
  }

  subscribe(key, query, callback, onError) {
    if (this.listeners.has(key)) {
      this.unsubscribe(key);
    }

    const unsubscribe = onSnapshot(
      query,
      (snapshot) => {
        const data = snapshot.docs.map(doc => ({
          id: doc.id,
          ...doc.data()
        }));
        callback(data);
      },
      (error) => {
        console.error(`Listener error for ${key}:`, error);
        onError?.(error);
      }
    );

    this.listeners.set(key, unsubscribe);
    return key;
  }

  unsubscribe(key) {
    const unsubscribe = this.listeners.get(key);
    if (unsubscribe) {
      unsubscribe();
      this.listeners.delete(key);
    }
  }

  unsubscribeAll() {
    this.listeners.forEach(unsubscribe => unsubscribe());
    this.listeners.clear();
  }
}

Presence Detection

User Presence System

class PresenceManager {
  constructor(userId) {
    this.userId = userId;
    this.presenceRef = doc(db, 'presence', userId);
  }

  async comeOnline() {
    await setDoc(this.presenceRef, {
      userId: this.userId,
      online: true,
      lastSeen: serverTimestamp(),
      status: 'active'
    });

    // Auto-cleanup on disconnect
    window.addEventListener('beforeunload', () => this.goOffline());
  }

  async goOffline() {
    await updateDoc(this.presenceRef, {
      online: false,
      lastSeen: serverTimestamp()
    });
  }

  watchPresence(callback) {
    return onSnapshot(
      collection(db, 'presence'),
      (snapshot) => {
        const users = {};
        snapshot.forEach(doc => {
          users[doc.id] = doc.data();
        });
        callback(users);
      }
    );
  }
}

// Usage
const presence = new PresenceManager(userId);
await presence.comeOnline();

presence.watchPresence((users) => {
  const activeUsers = Object.values(users).filter(u => u.online);
  console.log('Active users:', activeUsers);
});

Collaborative Editing with Operational Transformation

class CollaborativeDocument {
  constructor(docId) {
    this.docId = docId;
    this.content = '';
    this.version = 0;
    this.operations = [];
  }

  async insertText(position, text) {
    const op = {
      type: 'insert',
      position,
      text,
      userId: currentUserId,
      timestamp: Date.now(),
      version: this.version
    };

    // Save operation
    await addDoc(collection(db, `docs/${this.docId}/operations`), op);
    this.operations.push(op);
    this.version++;
  }

  async deleteText(position, length) {
    const op = {
      type: 'delete',
      position,
      length,
      userId: currentUserId,
      timestamp: Date.now(),
      version: this.version
    };

    await addDoc(collection(db, `docs/${this.docId}/operations`), op);
    this.operations.push(op);
    this.version++;
  }

  transform(op1, op2) {
    // Operational transformation logic
    if (op1.type === 'insert' && op2.type === 'insert') {
      if (op1.position < op2.position) {
        op2.position += op1.text.length;
      } else if (op1.position > op2.position) {
        op1.position += op2.text.length;
      } else {
        // Same position, use userId to break tie
        if (op1.userId < op2.userId) {
          op2.position += op1.text.length;
        } else {
          op1.position += op2.text.length;
        }
      }
    }
    return [op1, op2];
  }

  watchOperations(callback) {
    const operationsRef = collection(db, `docs/${this.docId}/operations`);
    const q = query(operationsRef, orderBy('version', 'asc'));

    return onSnapshot(q, (snapshot) => {
      snapshot.docChanges().forEach((change) => {
        if (change.type === 'added') {
          const op = change.doc.data();
          this.applyOperation(op);
          callback(this.content);
        }
      });
    });
  }

  applyOperation(op) {
    if (op.type === 'insert') {
      this.content = 
        this.content.slice(0, op.position) + 
        op.text + 
        this.content.slice(op.position);
    } else if (op.type === 'delete') {
      this.content = 
        this.content.slice(0, op.position) + 
        this.content.slice(op.position + op.length);
    }
  }
}

Subscription Management at Scale

class SubscriptionManager {
  constructor(maxSubscriptions = 50) {
    this.subscriptions = [];
    this.maxSubscriptions = maxSubscriptions;
  }

  addSubscription(key, unsubscribe, priority = 0) {
    this.subscriptions.push({key, unsubscribe, priority, createdAt: Date.now()});

    if (this.subscriptions.length > this.maxSubscriptions) {
      // Remove lowest priority or oldest subscription
      this.subscriptions.sort((a, b) => {
        if (a.priority !== b.priority) return b.priority - a.priority;
        return a.createdAt - b.createdAt;
      });

      const removed = this.subscriptions.pop();
      removed.unsubscribe();
      console.log(`Removed subscription: ${removed.key}`);
    }
  }

  removeSubscription(key) {
    const index = this.subscriptions.findIndex(s => s.key === key);
    if (index >= 0) {
      this.subscriptions[index].unsubscribe();
      this.subscriptions.splice(index, 1);
    }
  }
}

Real-Time Aggregations

class RealtimeAggregator {
  constructor() {
    this.aggregates = new Map();
  }

  watchAggregation(collectionName, aggregateField) {
    const q = collection(db, collectionName);

    return onSnapshot(q, (snapshot) => {
      let sum = 0;
      let count = 0;
      const values = [];

      snapshot.forEach(doc => {
        const value = doc.data()[aggregateField];
        if (typeof value === 'number') {
          sum += value;
          count++;
          values.push(value);
        }
      });

      const agg = {
        sum,
        count,
        average: count > 0 ? sum / count : 0,
        min: Math.min(...values),
        max: Math.max(...values)
      };

      this.aggregates.set(collectionName, agg);
      return agg;
    });
  }

  getAggregation(collectionName) {
    return this.aggregates.get(collectionName);
  }
}

Offline-First Architecture

class OfflineFirstManager {
  constructor() {
    this.pendingWrites = [];
  }

  async queueWrite(op) {
    // Store write for later sync
    localStorage.setItem(
      `pending_${Date.now()}`,
      JSON.stringify(op)
    );
    this.pendingWrites.push(op);
  }

  async syncPendingWrites() {
    const keys = Object.keys(localStorage).filter(k => k.startsWith('pending_'));

    for (const key of keys) {
      try {
        const op = JSON.parse(localStorage.getItem(key));
        await this.executeWrite(op);
        localStorage.removeItem(key);
      } catch (error) {
        console.error(`Failed to sync ${key}:`, error);
      }
    }

    this.pendingWrites = [];
  }

  async executeWrite(op) {
    if (op.type === 'create') {
      await addDoc(collection(db, op.collection), op.data);
    } else if (op.type === 'update') {
      await updateDoc(doc(db, op.collection, op.id), op.data);
    } else if (op.type === 'delete') {
      await deleteDoc(doc(db, op.collection, op.id));
    }
  }

  watchConnectivity() {
    window.addEventListener('online', () => {
      console.log('Back online, syncing...');
      this.syncPendingWrites();
    });

    window.addEventListener('offline', () => {
      console.log('Going offline, operations queued');
    });
  }
}

Performance Optimization for Real-Time

Listener Debouncing

class DebouncedListener {
  constructor(delay = 500) {
    this.delay = delay;
    this.timeout = null;
  }

  debounce(fn) {
    return (...args) => {
      clearTimeout(this.timeout);
      this.timeout = setTimeout(() => fn(...args), this.delay);
    };
  }

  watch(query, onDataChange) {
    const debouncedCallback = this.debounce(onDataChange);

    return onSnapshot(query, (snapshot) => {
      debouncedCallback(snapshot);
    });
  }
}

Key Takeaways

• **Listener management** prevents memory leaks at scale

• **Presence detection** enables awareness features

• **Operational transformation** handles collaborative editing

• **Subscription management** controls resource usage

• **Real-time aggregations** enable live statistics

• **Offline-first** provides seamless offline experience

• **Debouncing** optimizes real-time updates

Next Steps

Explore distributed locks for coordinated updates, or learn about conflict-free replicated data types (CRDTs).