tessl/npm-pouchdb

PouchDB is a pocket-sized database inspired by Apache CouchDB that runs in the browser and Node.js.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Changes and Monitoring

Name: tessl/npm-pouchdb
Author: tessl

Real-time monitoring of database changes with filtering, continuous feeds, and event-driven updates. The changes feed is essential for building reactive applications and implementing real-time synchronization.

Capabilities

Changes Feed

Monitors database changes with support for continuous feeds, filtering, and detailed change information.

/**
 * Monitor database changes with various options
 * @param options - Options for changes monitoring
 * @returns Changes object with event emitter interface
 */
db.changes(options?: ChangesOptions): Changes;

interface ChangesOptions {
  since?: number | string;       // Starting sequence number or 'now'
  limit?: number;                // Maximum number of changes to return
  descending?: boolean;          // Return changes in reverse order
  include_docs?: boolean;        // Include full document content
  conflicts?: boolean;           // Include conflicting revisions
  attachments?: boolean;         // Include attachment data
  live?: boolean;                // Continuous changes feed
  continuous?: boolean;          // Alias for live
  heartbeat?: number;            // Heartbeat interval (milliseconds)
  timeout?: number;              // Request timeout (milliseconds)
  filter?: string | Function;    // Filter function or design doc filter
  view?: string;                 // View-based filter
  doc_ids?: string[];            // Monitor specific document IDs only
  query_params?: object;         // Parameters for filter functions
}

interface Changes extends EventEmitter {
  cancel(): void;                // Stop the changes feed
  on(event: 'change', listener: (info: ChangeInfo) => void): Changes;
  on(event: 'complete', listener: (info: ChangesComplete) => void): Changes;
  on(event: 'error', listener: (error: Error) => void): Changes;
}

interface ChangeInfo {
  id: string;                    // Document ID
  seq: number | string;          // Sequence number
  changes: ChangeItem[];         // Array of changes
  doc?: Document;                // Full document (if include_docs: true)
  deleted?: boolean;             // Whether document was deleted
}

interface ChangeItem {
  rev: string;                   // Revision ID
}

interface ChangesComplete {
  results: ChangeInfo[];         // All changes (for non-live feeds)
  last_seq: number | string;     // Last sequence number
}

Usage Examples:

// Basic changes monitoring
const changes = db.changes({
  since: 'now',
  live: true,
  include_docs: true
}).on('change', function(info) {
  console.log('Document changed:', info.id);
  if (info.deleted) {
    console.log('Document was deleted');
  } else {
    console.log('New revision:', info.doc._rev);
  }
}).on('complete', function(info) {
  console.log('Changes feed complete');
}).on('error', function(err) {
  console.error('Changes feed error:', err);
});

// Stop the changes feed
setTimeout(() => {
  changes.cancel();
}, 30000); // Stop after 30 seconds

Historical Changes

Retrieve changes that occurred in the past with pagination and filtering.

// Get recent changes (non-live)
const result = await new Promise((resolve, reject) => {
  db.changes({
    limit: 50,
    include_docs: true
  }).on('complete', resolve).on('error', reject);
});

console.log(`Found ${result.results.length} changes`);
result.results.forEach(change => {
  console.log(`${change.id}: ${change.changes[0].rev}`);
});

// Get changes since specific sequence
const recentChanges = await new Promise((resolve, reject) => {
  db.changes({
    since: 100,
    limit: 20
  }).on('complete', resolve).on('error', reject);
});

Filtered Changes

Monitor changes for specific documents or using custom filter functions.

// Monitor specific documents only
const specificChanges = db.changes({
  live: true,
  doc_ids: ['user1', 'user2', 'user3'],
  include_docs: true
}).on('change', function(info) {
  console.log(`User document changed: ${info.id}`);
});

// Custom filter function
const filteredChanges = db.changes({
  live: true,
  include_docs: true,
  filter: function(doc) {
    return doc.type === 'user' && doc.active === true;
  }
}).on('change', function(info) {
  console.log('Active user changed:', info.doc.name);
});

// Filter with parameters
const parameterizedChanges = db.changes({
  live: true,
  filter: 'myapp/users',  // Design document filter
  query_params: {
    department: 'engineering',
    active: true
  }
});

Continuous Changes with Heartbeat

Set up reliable continuous monitoring with heartbeat and error handling.

function startChangesMonitoring() {
  const changes = db.changes({
    live: true,
    since: 'now',
    heartbeat: 10000,  // 10 second heartbeat
    timeout: 30000,    // 30 second timeout
    include_docs: true,
    retry: true
  });

  changes.on('change', function(info) {
    console.log('Change detected:', info.id);
    processChange(info);
  });

  changes.on('error', function(err) {
    console.error('Changes feed error:', err);
    // Restart after delay
    setTimeout(startChangesMonitoring, 5000);
  });

  changes.on('complete', function(info) {
    console.log('Changes feed completed');
    // Restart if needed
    setTimeout(startChangesMonitoring, 1000);
  });

  return changes;
}

function processChange(changeInfo) {
  if (changeInfo.deleted) {
    handleDocumentDeletion(changeInfo.id);
  } else {
    handleDocumentUpdate(changeInfo.doc);
  }
}

// Start monitoring
const changesMonitor = startChangesMonitoring();

// Stop monitoring when needed
function stopMonitoring() {
  changesMonitor.cancel();
}

Change Sequence Management

Work with sequence numbers for resuming changes feeds and synchronization.

// Store last sequence for resumption
let lastSequence = null;

const changes = db.changes({
  since: lastSequence || 0,
  live: true,
  include_docs: true
}).on('change', function(info) {
  // Update last sequence
  lastSequence = info.seq;
  localStorage.setItem('lastSequence', lastSequence);
  
  console.log(`Processed change ${info.id} at sequence ${info.seq}`);
}).on('complete', function(info) {
  // Save final sequence
  if (info.last_seq) {
    lastSequence = info.last_seq;
    localStorage.setItem('lastSequence', lastSequence);
  }
});

// Resume from stored sequence on restart
function resumeChanges() {
  const storedSequence = localStorage.getItem('lastSequence');
  return db.changes({
    since: storedSequence || 'now',
    live: true,
    include_docs: true
  });
}

Changes with Conflicts

Monitor and handle document conflicts in replicated environments.

const conflictMonitor = db.changes({
  live: true,
  include_docs: true,
  conflicts: true
}).on('change', function(info) {
  if (info.doc._conflicts && info.doc._conflicts.length > 0) {
    console.log(`Conflict detected in document ${info.id}`);
    handleConflict(info.doc);
  }
});

async function handleConflict(doc) {
  console.log(`Document ${doc._id} has conflicts:`, doc._conflicts);
  
  // Get all conflicting revisions
  const conflicts = await Promise.all(
    doc._conflicts.map(rev => db.get(doc._id, { rev }))
  );
  
  // Implement conflict resolution logic
  const resolved = resolveConflict(doc, conflicts);
  
  // Save resolved document
  await db.put(resolved);
  
  // Remove conflicting revisions
  for (const conflict of conflicts) {
    await db.remove(conflict);
  }
}

function resolveConflict(mainDoc, conflictDocs) {
  // Implement your conflict resolution strategy
  // This example uses "last write wins" based on timestamp
  const allDocs = [mainDoc, ...conflictDocs];
  return allDocs.reduce((latest, current) => {
    const latestTime = new Date(latest.updated || 0);
    const currentTime = new Date(current.updated || 0);
    return currentTime > latestTime ? current : latest;
  });
}

Event Handling Patterns

Promise-based Changes

Convert changes feed to Promise for one-time change retrieval:

// Get changes as Promise
function getChanges(options) {
  return new Promise((resolve, reject) => {
    db.changes(options)
      .on('complete', resolve)
      .on('error', reject);
  });
}

// Usage
const changes = await getChanges({
  since: 'now',
  limit: 10,
  include_docs: true
});

console.log(`Received ${changes.results.length} changes`);

Async Iterator Pattern

Create async iterator for changes processing:

async function* changesIterator(options) {
  const changes = db.changes({ ...options, live: true });
  
  try {
    while (true) {
      const change = await new Promise((resolve, reject) => {
        changes.once('change', resolve);
        changes.once('error', reject);
        changes.once('complete', () => resolve(null));
      });
      
      if (change === null) break; // Complete
      yield change;
    }
  } finally {
    changes.cancel();
  }
}

// Usage
for await (const change of changesIterator({ include_docs: true })) {
  console.log('Processing change:', change.id);
  await processChange(change);
}

Error Handling

Handle various error scenarios in changes monitoring:

const changes = db.changes({
  live: true,
  since: 'now'
}).on('change', function(info) {
  console.log('Change:', info.id);
}).on('error', function(err) {
  if (err.status === 401) {
    console.log('Authentication required');
    // Handle auth error
  } else if (err.status === 404) {
    console.log('Database not found');
    // Handle missing database
  } else if (err.name === 'timeout') {
    console.log('Changes feed timeout');
    // Restart feed
  } else {
    console.error('Unexpected error:', err);
  }
});

Common error scenarios: