MongoDB Sync API Reference

Complete API documentation for the MongoDB integration in the Storage System.

Table of Contents

1. MongoDBSyncAdapter
2. IntelligentSyncEngine
3. MongoDBRealtimeService
4. SyncConfigService
5. Provider Functions
6. Types & Interfaces

MongoDBSyncAdapter

Main adapter for MongoDB synchronization operations.

Constructor

constructor(options?: MongoDBSyncAdapterOptions)

Parameters

MongoDBSyncAdapterOptions

interface MongoDBSyncAdapterOptions {
  connectionString?: string;
  database?: string;
  collections?: Record<string, string>;
  syncStrategy?: SyncStrategy;
  localStorage?: GgStorage;
  autoConnect?: boolean;
}

Methods

connect()

Establish connection to MongoDB Atlas.

async connect(): Promise<void>

Example:

const adapter = new MongoDBSyncAdapter({
  connectionString: process.env['MONGODB_CONNECTION_STRING'],
  database: 'my-app'
});

await adapter.connect();
console.log('Connected to MongoDB');

disconnect()

Close MongoDB connection and cleanup resources.

async disconnect(): Promise<void>

Example:

await adapter.disconnect();
console.log('Disconnected from MongoDB');

push()

Sync local data to MongoDB.

async push(
  collection: string,
  data: any,
  options?: PushOptions
): Promise<SyncResult>

Parameters:

PushOptions:

interface PushOptions {
  priority?: SyncPriority;
  immediate?: boolean;
  conflictResolution?: ConflictResolution;
  metadata?: Record<string, any>;
}

Returns: Promise<SyncResult>

interface SyncResult {
  success: boolean;
  documentId?: string;
  version?: number;
  conflictResolved?: boolean;
  error?: Error;
}

Example:

const result = await adapter.push('user-preferences', {
  userId: '123',
  theme: 'dark',
  language: 'en'
}, {
  priority: 'high',
  immediate: true,
  conflictResolution: 'last-write-wins'
});

if (result.success) {
  console.log('Synced with ID:', result.documentId);
}

pull()

Fetch data from MongoDB to local storage.

async pull(
  collection: string,
  query?: QueryOptions,
  options?: PullOptions
): Promise<any[]>

Parameters:

QueryOptions:

interface QueryOptions {
  filter?: Record<string, any>;
  sort?: Record<string, 1 | -1>;
  limit?: number;
  skip?: number;
  projection?: Record<string, 0 | 1>;
}

PullOptions:

interface PullOptions {
  updateLocal?: boolean;
  cacheResults?: boolean;
  ttl?: number;
}

Example:

const userPrefs = await adapter.pull('user-preferences', {
  filter: { userId: '123' },
  limit: 10
}, {
  updateLocal: true,
  cacheResults: true,
  ttl: 3600000 // 1 hour
});

console.log('Fetched preferences:', userPrefs);

bulkPush()

Sync multiple documents in a single batch.

async bulkPush(
  collection: string,
  documents: any[],
  options?: BulkPushOptions
): Promise<BulkSyncResult>

Parameters:

BulkPushOptions:

interface BulkPushOptions {
  ordered?: boolean;
  batchSize?: number;
  onProgress?: (progress: number) => void;
}

Returns: Promise<BulkSyncResult>

interface BulkSyncResult {
  totalCount: number;
  successCount: number;
  failureCount: number;
  errors: Array<{ index: number; error: Error }>;
  duration: number;
}

Example:

const tasks = [
  { id: 1, title: 'Task 1', completed: false },
  { id: 2, title: 'Task 2', completed: true },
  { id: 3, title: 'Task 3', completed: false }
];

const result = await adapter.bulkPush('tasks', tasks, {
  batchSize: 100,
  ordered: false,
  onProgress: (progress) => console.log(`Progress: ${progress}%`)
});

console.log(`Synced ${result.successCount}/${result.totalCount} tasks`);

find()

Query documents from MongoDB.

async find(
  collection: string,
  filter: Record<string, any>,
  options?: FindOptions
): Promise<any[]>

FindOptions:

interface FindOptions {
  sort?: Record<string, 1 | -1>;
  limit?: number;
  skip?: number;
  projection?: Record<string, 0 | 1>;
}

Example:

const completedTasks = await adapter.find('tasks', {
  completed: true,
  dueDate: { $lt: new Date() }
}, {
  sort: { dueDate: -1 },
  limit: 20
});

aggregate()

Execute aggregation pipeline.

async aggregate(
  collection: string,
  pipeline: any[]
): Promise<any[]>

Example:

const taskStats = await adapter.aggregate('tasks', [
  { $match: { userId: '123' } },
  { $group: {
      _id: '$status',
      count: { $sum: 1 },
      avgPriority: { $avg: '$priority' }
    }
  },
  { $sort: { count: -1 } }
]);

console.log('Task statistics:', taskStats);

setSyncStrategy()

Update sync strategy for a collection.

setSyncStrategy(
  collection: string,
  strategy: SyncStrategy
): void

Example:

adapter.setSyncStrategy('user-preferences', {
  mode: 'realtime',
  priority: 'critical',
  conflictResolution: 'merge'
});

getSyncStatus()

Get current sync status for a collection.

getSyncStatus(collection: string): SyncStatus

Returns: SyncStatus

interface SyncStatus {
  state: 'idle' | 'syncing' | 'error';
  lastSyncTime?: number;
  pendingOperations: number;
  lastError?: Error;
}

Example:

const status = adapter.getSyncStatus('tasks');
console.log(`Pending operations: ${status.pendingOperations}`);

IntelligentSyncEngine

Orchestrates intelligent synchronization with priority-based queuing.

Constructor

constructor(options?: IntelligentSyncEngineOptions)

IntelligentSyncEngineOptions

interface IntelligentSyncEngineOptions {
  syncAdapter?: MongoDBSyncAdapter;
  localStorage?: GgStorage;
  autoSync?: boolean;
  globalSyncInterval?: number;
  enableMetrics?: boolean;
}

Reactive Signals

readonly queueStats: Signal<SyncQueueStats>;
readonly isSyncing: Signal<boolean>;
readonly isOnline: Signal<boolean>;
readonly syncProgress: Signal<number>;
readonly hasPendingOperations: Signal<boolean>; // Computed
readonly needsUrgentSync: Signal<boolean>;      // Computed

Methods

syncData()

Queue data for synchronization.

async syncData(
  collection: string,
  data: any,
  priority?: SyncPriority
): Promise<string>

SyncOperation:

interface SyncOperation {
  id?: string;
  type: 'push' | 'pull' | 'delete';
  collection: string;
  data?: any;
  query?: any;
  priority: SyncPriority;
  timestamp: number;
  retries?: number;
}

Returns: Operation ID

Example:

const syncEngine = new IntelligentSyncEngine({ 
  syncAdapter: adapter,
  autoSync: true 
});

// Queue data for sync
const opId = await syncEngine.syncData('tasks', {
  id: 1,
  title: 'New Task',
  completed: false
}, 'high');

console.log('Operation queued:', opId);

syncCollection()

Sync a specific collection immediately.

async syncCollection(collection: string): Promise<SyncResult[]>

Example:

const results = await syncEngine.syncCollection('tasks');
console.log(`Synced ${results.length} operations`);

syncNow()

Sync all pending operations across all collections.

async syncNow(): Promise<Map<string, SyncResult[]>>

Example:

const results = await syncEngine.syncNow();
console.log(`Synced ${results.size} collections`);

initialize()

Initialize the sync engine and connect to MongoDB.

async initialize(): Promise<void>

shutdown()

Gracefully shutdown the sync engine.

async shutdown(): Promise<void>

MongoDBRealtimeService

Provides real-time synchronization using MongoDB Change Streams.

Constructor

constructor(options?: MongoDBRealtimeServiceOptions)

MongoDBRealtimeServiceOptions

interface MongoDBRealtimeServiceOptions {
  adapter?: MongoDBSyncAdapter;
  collections?: string[];
  eventFilter?: ChangeStreamFilter;
  autoReconnect?: boolean;
  localStorage?: GgStorage;
}

Methods

subscribe()

Subscribe to real-time changes for a collection.

subscribe(
  collection: string,
  options?: SubscribeOptions
): Subscription

SubscribeOptions:

interface SubscribeOptions {
  filter?: ChangeStreamFilter;
  fullDocument?: 'default' | 'updateLookup';
  onInsert?: (doc: any) => void;
  onUpdate?: (doc: any) => void;
  onDelete?: (id: string) => void;
}

Returns: Subscription (with unsubscribe() method)

Example:

const realtimeService = new MongoDBRealtimeService({ adapter });

const subscription = realtimeService.subscribe('tasks', {
  filter: { 'fullDocument.userId': '123' },
  onInsert: (task) => console.log('New task:', task),
  onUpdate: (task) => console.log('Task updated:', task),
  onDelete: (id) => console.log('Task deleted:', id)
});

// Later...
subscription.unsubscribe();

unsubscribe()

Unsubscribe from a collection.

unsubscribe(collection: string): void

unsubscribeAll()

Unsubscribe from all collections.

unsubscribeAll(): void

isSubscribed()

Check if subscribed to a collection.

isSubscribed(collection: string): boolean

getConnectionStatus()

Get current connection status.

getConnectionStatus(): ConnectionStatus

ConnectionStatus:

interface ConnectionStatus {
  connected: boolean;
  activeSubscriptions: number;
  lastHeartbeat?: number;
  reconnecting: boolean;
}

SyncConfigService

Manages sync configuration for collections.

Constructor

constructor(options?: SyncConfigServiceOptions)

SyncConfigServiceOptions

interface SyncConfigServiceOptions {
  localStorage?: GgStorage;
  defaultStrategy?: SyncStrategy;
}

Methods

setCollectionConfig()

Set sync configuration for a collection.

async setCollectionConfig(
  collection: string,
  config: CollectionSyncConfig
): Promise<void>

CollectionSyncConfig:

interface CollectionSyncConfig {
  mode: 'offline-first' | 'online-first' | 'realtime';
  priority: SyncPriority;
  interval?: number;
  batchSize?: number;
  conflictResolution: ConflictResolution;
  enabled: boolean;
}

Example:

const configService = new SyncConfigService();

await configService.setCollectionConfig('user-preferences', {
  mode: 'offline-first',
  priority: 'high',
  interval: 300000, // 5 minutes
  batchSize: 50,
  conflictResolution: 'last-write-wins',
  enabled: true
});

getCollectionConfig()

Get sync configuration for a collection.

getCollectionConfig(collection: string): CollectionSyncConfig | undefined

applyPreset()

Apply a preset configuration.

async applyPreset(preset: ConfigPreset): Promise<void>

ConfigPreset:

type ConfigPreset = 
  | 'offline-first-all'
  | 'online-first-all'
  | 'realtime-all'
  | 'balanced'
  | 'performance'
  | 'battery-saver';

Example:

await configService.applyPreset('offline-first-all');

exportConfig()

Export current configuration.

exportConfig(): ConfigExport

ConfigExport:

interface ConfigExport {
  version: string;
  timestamp: number;
  collections: Record<string, CollectionSyncConfig>;
}

importConfig()

Import configuration from export.

async importConfig(config: ConfigExport): Promise<void>

Provider Functions

provideMongoDBSync()

Full-featured MongoDB sync provider.

function provideMongoDBSync(
  config: MongoDBProviderConfig
): EnvironmentProviders

MongoDBProviderConfig:

interface MongoDBProviderConfig {
  connectionString: string;
  database: string;
  collections?: Record<string, string>;
  defaultStrategy?: SyncStrategy;
  enableRealtime?: boolean;
  autoInit?: boolean;
}

Example:

import { provideMongoDBSync } from '@storage-system';

export const appConfig: ApplicationConfig = {
  providers: [
    provideMongoDBSync({
      connectionString: environment.mongoDbConnectionString,
      database: 'my-app',
      collections: {
        userPreferences: 'user-preferences',
        appState: 'app-state'
      },
      defaultStrategy: {
        mode: 'offline-first',
        priority: 'normal',
        conflictResolution: 'last-write-wins'
      },
      enableRealtime: true,
      autoInit: true
    })
  ]
};

provideMongoDBSyncMinimal()

Lightweight MongoDB sync provider.

function provideMongoDBSyncMinimal(
  connectionString: string,
  database: string
): EnvironmentProviders

Example:

export const appConfig: ApplicationConfig = {
  providers: [
    provideMongoDBSyncMinimal(
      environment.mongoDbConnectionString,
      'my-app'
    )
  ]
};

Types & Interfaces

SyncPriority

type SyncPriority = 'critical' | 'high' | 'normal' | 'low';

Usage:

• critical: Sync immediately (user-facing changes)
• high: Sync within 5 minutes (important data)
• normal: Sync within 30 minutes (regular updates)
• low: Sync manually or during idle time (analytics, logs)

ConflictResolution

type ConflictResolution = 'last-write-wins' | 'merge' | 'manual';

Strategies:

• last-write-wins: Most recent timestamp wins
• merge: Field-level merge (keeps both changes)
• manual: Emit event for user resolution

SyncStrategy

interface SyncStrategy {
  mode: 'offline-first' | 'online-first' | 'realtime';
  priority: SyncPriority;
  interval?: number;
  batchSize?: number;
  conflictResolution: ConflictResolution;
}

ChangeStreamFilter

interface ChangeStreamFilter {
  operationType?: 'insert' | 'update' | 'delete' | 'replace';
  'fullDocument.field'?: any;
  'updateDescription.updatedFields.field'?: any;
}

Example:

const filter: ChangeStreamFilter = {
  operationType: 'insert',
  'fullDocument.userId': '123',
  'fullDocument.status': { $ne: 'deleted' }
};

Subscription

interface Subscription {
  collection: string;
  unsubscribe: () => void;
  isActive: () => boolean;
}

Error Handling

All async methods can throw the following errors:

ConnectionError

class ConnectionError extends Error {
  code: 'CONN_FAILED' | 'CONN_TIMEOUT' | 'AUTH_FAILED';
  retryable: boolean;
}

SyncError

class SyncError extends Error {
  code: 'SYNC_FAILED' | 'CONFLICT' | 'QUOTA_EXCEEDED';
  operation: SyncOperation;
  retryable: boolean;
}

ValidationError

class ValidationError extends Error {
  code: 'INVALID_DATA' | 'SCHEMA_MISMATCH';
  details: ValidationDetail[];
}

Example Error Handling:

try {
  await adapter.push('tasks', taskData);
} catch (error) {
  if (error instanceof SyncError && error.retryable) {
    console.log('Sync failed, will retry:', error.message);
    await syncEngine.enqueue(error.operation);
  } else {
    console.error('Fatal sync error:', error);
  }
}

Performance Considerations

Connection Pooling

The adapter automatically manages connection pooling. Default pool size is 10 connections.

const adapter = new MongoDBSyncAdapter({
  connectionString: process.env['MONGODB_CONNECTION_STRING'],
  // Connection pool is managed automatically
});

Batch Operations

Always prefer bulkPush() over multiple push() calls:

// ❌ Bad - Multiple network calls
for (const doc of documents) {
  await adapter.push('collection', doc);
}

// ✅ Good - Single batched call
await adapter.bulkPush('collection', documents);

Query Optimization

Use projections to reduce data transfer:

// ❌ Bad - Fetches all fields
const users = await adapter.find('users', { active: true });

// ✅ Good - Fetches only needed fields
const users = await adapter.find('users', { active: true }, {
  projection: { name: 1, email: 1, _id: 1 }
});

Next Steps

• 🎯 Advanced Topics - Custom strategies and optimization
• 🚀 Getting Started - Quick setup guide
• 📊 Overview - Architecture and design

Need Help? Check the troubleshooting guide or open an issue.