MongoDB Advanced Topics

Dive deep into advanced MongoDB sync features, optimization techniques, and production strategies for the Storage System.

Table of Contents

1. Custom Sync Strategies
2. Advanced Conflict Resolution
3. Performance Optimization
4. Security Best Practices
5. Scaling Strategies
6. Monitoring & Debugging
7. Migration Strategies

Custom Sync Strategies

Creating Custom Sync Strategies

Beyond the built-in strategies, you can create custom sync logic tailored to your application's needs.

Custom Strategy Example

import { 
  MongoDBSyncAdapter,
  SyncStrategy,
  SyncPriority,
  ConflictResolution 
} from '@storage-system';

interface CustomSyncStrategy extends SyncStrategy {
  mode: 'custom-hybrid';
  businessLogic: {
    syncDuringBusinessHours: boolean;
    throttleOnMobileData: boolean;
    prioritizeUserActions: boolean;
  };
}

const customStrategy: CustomSyncStrategy = {
  mode: 'custom-hybrid',
  priority: 'high',
  interval: 15 * 60 * 1000, // 15 minutes
  batchSize: 50,
  conflictResolution: 'merge',
  businessLogic: {
    syncDuringBusinessHours: true,
    throttleOnMobileData: true,
    prioritizeUserActions: true
  }
};

Conditional Sync Logic

Implement sync logic that adapts to network conditions and user context:

class AdaptiveSyncStrategy {
  private readonly adapter: MongoDBSyncAdapter;
  private networkType: 'wifi' | 'cellular' | 'offline' = 'wifi';

  async determineSyncStrategy(collection: string): Promise<SyncStrategy> {
    // Detect network type
    this.networkType = await this.detectNetworkType();

    // User is on mobile data - reduce sync frequency
    if (this.networkType === 'cellular') {
      return {
        mode: 'offline-first',
        priority: 'low',
        interval: 60 * 60 * 1000, // 1 hour
        batchSize: 10,
        conflictResolution: 'last-write-wins'
      };
    }

    // User is on WiFi - aggressive sync
    if (this.networkType === 'wifi') {
      return {
        mode: 'online-first',
        priority: 'high',
        interval: 5 * 60 * 1000, // 5 minutes
        batchSize: 100,
        conflictResolution: 'merge'
      };
    }

    // Offline - queue everything
    return {
      mode: 'offline-first',
      priority: 'critical',
      interval: 0,
      batchSize: 1,
      conflictResolution: 'last-write-wins'
    };
  }

  private async detectNetworkType(): Promise<'wifi' | 'cellular' | 'offline'> {
    if (!navigator.onLine) return 'offline';
    
    // Use Network Information API if available
    const connection = (navigator as any).connection;
    if (connection) {
      return connection.effectiveType === '4g' && !connection.saveData
        ? 'wifi'
        : 'cellular';
    }
    
    return 'wifi'; // Default assumption
  }
}

Time-Based Sync Windows

Sync only during specific time windows:

class TimeWindowSyncStrategy {
  private syncWindows = [
    { start: '09:00', end: '17:00' }, // Business hours
    { start: '20:00', end: '22:00' }  // Evening sync
  ];

  isWithinSyncWindow(): boolean {
    const now = new Date();
    const currentTime = `${now.getHours().toString().padStart(2, '0')}:${now.getMinutes().toString().padStart(2, '0')}`;

    return this.syncWindows.some(window => 
      currentTime >= window.start && currentTime <= window.end
    );
  }

  async conditionalSync(collection: string, data: any): Promise<void> {
    if (this.isWithinSyncWindow()) {
      // Immediate sync
      await this.adapter.push(collection, data);
    } else {
      // Queue for next window
      await this.queueForLaterSync(collection, data);
    }
  }
}

Advanced Conflict Resolution

Custom Merge Strategies

Implement field-level merge logic for complex data structures:

interface MergeStrategy<T = any> {
  resolve(local: T, remote: T, metadata: ConflictMetadata): Promise<T>;
}

interface ConflictMetadata {
  localTimestamp: number;
  remoteTimestamp: number;
  field: string;
  conflictType: 'update' | 'delete' | 'concurrent';
}

class FieldLevelMergeStrategy implements MergeStrategy {
  async resolve(local: any, remote: any, metadata: ConflictMetadata): Promise<any> {
    const merged = { ...local };

    for (const key in remote) {
      // Arrays - merge unique values
      if (Array.isArray(remote[key]) && Array.isArray(local[key])) {
        merged[key] = [...new Set([...local[key], ...remote[key]])];
        continue;
      }

      // Objects - deep merge
      if (typeof remote[key] === 'object' && typeof local[key] === 'object') {
        merged[key] = await this.resolve(local[key], remote[key], metadata);
        continue;
      }

      // Primitive values - use timestamp
      if (local[key] !== remote[key]) {
        merged[key] = metadata.localTimestamp > metadata.remoteTimestamp
          ? local[key]
          : remote[key];
      }
    }

    return merged;
  }
}

Operational Transform for Collaborative Editing

Implement OT for real-time collaborative text editing:

interface Operation {
  type: 'insert' | 'delete' | 'retain';
  position: number;
  content?: string;
  length?: number;
}

class OperationalTransform {
  transform(op1: Operation, op2: Operation): [Operation, Operation] {
    // Insert vs Insert
    if (op1.type === 'insert' && op2.type === 'insert') {
      if (op1.position < op2.position) {
        return [op1, { ...op2, position: op2.position + (op1.content?.length || 0) }];
      } else if (op1.position > op2.position) {
        return [{ ...op1, position: op1.position + (op2.content?.length || 0) }, op2];
      }
      // Same position - deterministic resolution
      return [op1, { ...op2, position: op2.position + (op1.content?.length || 0) }];
    }

    // Insert vs Delete
    if (op1.type === 'insert' && op2.type === 'delete') {
      if (op1.position <= op2.position) {
        return [op1, { ...op2, position: op2.position + (op1.content?.length || 0) }];
      } else {
        const adjustedPos = Math.max(op2.position, op1.position - (op2.length || 0));
        return [{ ...op1, position: adjustedPos }, op2];
      }
    }

    // Delete vs Insert (inverse of above)
    if (op1.type === 'delete' && op2.type === 'insert') {
      const [op2Prime, op1Prime] = this.transform(op2, op1);
      return [op1Prime, op2Prime];
    }

    // Delete vs Delete
    if (op1.type === 'delete' && op2.type === 'delete') {
      // Complex case - ranges may overlap
      return this.transformDeletes(op1, op2);
    }

    return [op1, op2];
  }

  private transformDeletes(op1: Operation, op2: Operation): [Operation, Operation] {
    const op1End = op1.position + (op1.length || 0);
    const op2End = op2.position + (op2.length || 0);

    // No overlap
    if (op1End <= op2.position) {
      return [op1, { ...op2, position: op2.position - (op1.length || 0) }];
    }
    if (op2End <= op1.position) {
      return [{ ...op1, position: op1.position - (op2.length || 0) }, op2];
    }

    // Overlap - complex resolution needed
    // This is simplified; production OT requires more sophisticated logic
    return [op1, op2];
  }
}

Three-Way Merge with Common Ancestor

Use a common ancestor to resolve conflicts:

class ThreeWayMerge {
  async merge<T>(base: T, local: T, remote: T): Promise<T> {
    const result: any = { ...base };

    for (const key in base) {
      const baseValue = (base as any)[key];
      const localValue = (local as any)[key];
      const remoteValue = (remote as any)[key];

      // No conflict - both changed to same value
      if (localValue === remoteValue) {
        result[key] = localValue;
        continue;
      }

      // Only local changed
      if (baseValue === remoteValue && baseValue !== localValue) {
        result[key] = localValue;
        continue;
      }

      // Only remote changed
      if (baseValue === localValue && baseValue !== remoteValue) {
        result[key] = remoteValue;
        continue;
      }

      // Both changed - conflict!
      result[key] = await this.resolveConflict(key, localValue, remoteValue);
    }

    return result;
  }

  private async resolveConflict(key: string, local: any, remote: any): Promise<any> {
    // Strategy 1: Type-based resolution
    if (typeof local === 'number' && typeof remote === 'number') {
      return Math.max(local, remote); // Prefer higher value
    }

    // Strategy 2: Array merging
    if (Array.isArray(local) && Array.isArray(remote)) {
      return [...new Set([...local, ...remote])];
    }

    // Strategy 3: Manual resolution (emit event for UI)
    return await this.requestManualResolution(key, local, remote);
  }

  private async requestManualResolution(key: string, local: any, remote: any): Promise<any> {
    // Emit event for UI to handle
    const resolution = await new Promise<any>(resolve => {
      window.dispatchEvent(new CustomEvent('conflict-resolution-needed', {
        detail: { key, local, remote, resolve }
      }));
    });
    return resolution;
  }
}

Performance Optimization

Connection Pooling Optimization

Fine-tune MongoDB connection pool for your workload:

const optimizedMongoDBConfig = {
  connectionString: process.env['MONGODB_CONNECTION_STRING'],
  options: {
    // Connection Pool Settings
    maxPoolSize: 20,        // Max concurrent connections
    minPoolSize: 5,         // Min idle connections
    maxIdleTimeMS: 60000,   // Close idle after 1 minute
    waitQueueTimeoutMS: 5000, // Max wait for connection

    // Network Settings
    connectTimeoutMS: 10000,  // Connection timeout
    socketTimeoutMS: 45000,   // Socket timeout
    serverSelectionTimeoutMS: 30000,

    // Retry Settings
    retryWrites: true,
    retryReads: true,

    // Compression
    compressors: ['snappy', 'zlib'],

    // Write Concern
    w: 'majority',
    journal: true,
    wtimeoutMS: 5000
  }
};

Batch Size Optimization

Determine optimal batch sizes based on document size:

class BatchOptimizer {
  private readonly MAX_BATCH_SIZE_BYTES = 16 * 1024 * 1024; // 16MB (MongoDB limit)
  private readonly TARGET_BATCH_SIZE_BYTES = 4 * 1024 * 1024; // 4MB (optimal)

  calculateOptimalBatchSize(sampleDocuments: any[]): number {
    if (sampleDocuments.length === 0) return 50; // Default

    // Estimate average document size
    const sampleSize = JSON.stringify(sampleDocuments).length;
    const avgDocSize = sampleSize / sampleDocuments.length;

    // Calculate optimal batch count
    const optimalCount = Math.floor(this.TARGET_BATCH_SIZE_BYTES / avgDocSize);

    // Clamp between reasonable bounds
    return Math.max(10, Math.min(1000, optimalCount));
  }

  async autoTuneBatchSize(collection: string): Promise<number> {
    // Sample recent documents
    const samples = await this.adapter.find(collection, {}, { limit: 10 });
    return this.calculateOptimalBatchSize(samples);
  }
}

Index Strategy for Sync Queries

Create indexes to optimize sync queries:

const syncIndexStrategy = {
  // Compound index for timestamp-based queries
  timestampIndex: {
    keys: { updatedAt: 1, _id: 1 },
    options: { name: 'sync_timestamp_idx' }
  },

  // Index for conflict resolution
  versionIndex: {
    keys: { _version: 1, updatedAt: -1 },
    options: { name: 'version_idx' }
  },

  // Partial index for pending syncs
  pendingSyncIndex: {
    keys: { syncStatus: 1, priority: -1 },
    options: { 
      name: 'pending_sync_idx',
      partialFilterExpression: { syncStatus: 'pending' }
    }
  },

  // TTL index for temporary data
  ttlIndex: {
    keys: { expiresAt: 1 },
    options: { 
      name: 'ttl_idx',
      expireAfterSeconds: 0 
    }
  }
};

async function setupSyncIndexes(db: any): Promise<void> {
  for (const collection of ['user-preferences', 'app-state', 'cache']) {
    await db.collection(collection).createIndexes([
      syncIndexStrategy.timestampIndex,
      syncIndexStrategy.versionIndex,
      syncIndexStrategy.pendingSyncIndex,
      syncIndexStrategy.ttlIndex
    ]);
  }
}

Compression Strategy

Implement smart compression for large documents:

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

class SyncCompressor {
  private readonly compressor = new IntelligentCompressor();
  private readonly COMPRESSION_THRESHOLD = 1024; // 1KB

  async compressIfNeeded(data: any): Promise<{ data: any; compressed: boolean }> {
    const serialized = JSON.stringify(data);
    
    if (serialized.length < this.COMPRESSION_THRESHOLD) {
      return { data, compressed: false };
    }

    const compressed = await this.compressor.compress(serialized);
    return { 
      data: compressed.data,
      compressed: true 
    };
  }

  async syncWithCompression(collection: string, document: any): Promise<void> {
    const { data, compressed } = await this.compressIfNeeded(document);
    
    await this.adapter.push(collection, {
      ...data,
      _compressed: compressed,
      _originalSize: compressed ? JSON.stringify(document).length : undefined
    });
  }
}

Security Best Practices

Connection String Security

Never hardcode connection strings - use secure methods:

// ❌ BAD - Hardcoded credentials
const badConfig = {
  connectionString: 'mongodb+srv://user:[email protected]/'
};

// ✅ GOOD - Environment variables
const goodConfig = {
  connectionString: process.env['MONGODB_CONNECTION_STRING']
};

// ✅ BETTER - Secrets management
import { SecretManager } from '@google-cloud/secret-manager';

async function getSecureConnection(): Promise<string> {
  const client = new SecretManager.SecretManagerServiceClient();
  const [version] = await client.accessSecretVersion({
    name: 'projects/my-project/secrets/mongodb-connection/versions/latest'
  });
  return version.payload?.data?.toString() || '';
}

Field-Level Encryption

Encrypt sensitive fields before syncing:

import { webcrypto } from 'crypto';

class FieldEncryption {
  private readonly algorithm = 'AES-GCM';
  private key: CryptoKey | null = null;

  async initialize(): Promise<void> {
    // Derive key from secure source
    const keyMaterial = await webcrypto.subtle.importKey(
      'raw',
      new TextEncoder().encode(process.env['ENCRYPTION_KEY'] || ''),
      { name: 'PBKDF2' },
      false,
      ['deriveKey']
    );

    this.key = await webcrypto.subtle.deriveKey(
      {
        name: 'PBKDF2',
        salt: new TextEncoder().encode('storage-system-salt'),
        iterations: 100000,
        hash: 'SHA-256'
      },
      keyMaterial,
      { name: this.algorithm, length: 256 },
      false,
      ['encrypt', 'decrypt']
    );
  }

  async encryptField(value: string): Promise<string> {
    if (!this.key) throw new Error('Encryption not initialized');

    const iv = webcrypto.getRandomValues(new Uint8Array(12));
    const encrypted = await webcrypto.subtle.encrypt(
      { name: this.algorithm, iv },
      this.key,
      new TextEncoder().encode(value)
    );

    // Combine IV and encrypted data
    const combined = new Uint8Array(iv.length + encrypted.byteLength);
    combined.set(iv);
    combined.set(new Uint8Array(encrypted), iv.length);

    return btoa(String.fromCharCode(...combined));
  }

  async decryptField(encrypted: string): Promise<string> {
    if (!this.key) throw new Error('Encryption not initialized');

    const combined = Uint8Array.from(atob(encrypted), c => c.charCodeAt(0));
    const iv = combined.slice(0, 12);
    const data = combined.slice(12);

    const decrypted = await webcrypto.subtle.decrypt(
      { name: this.algorithm, iv },
      this.key,
      data
    );

    return new TextDecoder().decode(decrypted);
  }
}

// Usage
const encryption = new FieldEncryption();
await encryption.initialize();

const sensitiveDocument = {
  userId: '123',
  email: await encryption.encryptField('[email protected]'),
  ssn: await encryption.encryptField('123-45-6789'),
  publicData: 'this is not encrypted'
};

Role-Based Access Control

Implement RBAC for sync operations:

interface SyncPermissions {
  canRead: boolean;
  canWrite: boolean;
  canDelete: boolean;
  collections: string[];
}

class RBACSync {
  private permissions: Map<string, SyncPermissions> = new Map();

  setUserPermissions(userId: string, permissions: SyncPermissions): void {
    this.permissions.set(userId, permissions);
  }

  async authorizedSync(
    userId: string,
    operation: 'read' | 'write' | 'delete',
    collection: string,
    data: any
  ): Promise<void> {
    const perms = this.permissions.get(userId);
    
    if (!perms) {
      throw new Error('User not authorized');
    }

    if (!perms.collections.includes(collection)) {
      throw new Error(`Access denied to collection: ${collection}`);
    }

    if (operation === 'read' && !perms.canRead) {
      throw new Error('Read permission denied');
    }

    if (operation === 'write' && !perms.canWrite) {
      throw new Error('Write permission denied');
    }

    if (operation === 'delete' && !perms.canDelete) {
      throw new Error('Delete permission denied');
    }

    // Proceed with authorized operation
    await this.performSync(operation, collection, data);
  }
}

Scaling Strategies

Sharding for Large Collections

Configure sharding for horizontal scaling:

const shardingStrategy = {
  // Shard by user ID for user-specific data
  userDataSharding: {
    collection: 'user-preferences',
    key: { userId: 'hashed' },
    unique: false
  },

  // Shard by timestamp for time-series data
  timeSeriesSharding: {
    collection: 'sync-events',
    key: { timestamp: 1, _id: 1 },
    unique: false
  },

  // Compound shard key for multi-tenant
  multiTenantSharding: {
    collection: 'app-state',
    key: { tenantId: 1, _id: 1 },
    unique: false
  }
};

Read/Write Splitting

Optimize by splitting read and write operations:

class ReadWriteSplit {
  private writeAdapter: MongoDBSyncAdapter;
  private readReplicas: MongoDBSyncAdapter[];
  private currentReplicaIndex = 0;

  constructor(
    primary: MongoDBSyncAdapter,
    replicas: MongoDBSyncAdapter[]
  ) {
    this.writeAdapter = primary;
    this.readReplicas = replicas;
  }

  async write(collection: string, data: any): Promise<void> {
    // All writes go to primary
    return this.writeAdapter.push(collection, data);
  }

  async read(collection: string, query: any): Promise<any[]> {
    // Round-robin across read replicas
    const replica = this.readReplicas[this.currentReplicaIndex];
    this.currentReplicaIndex = (this.currentReplicaIndex + 1) % this.readReplicas.length;
    
    return replica.find(collection, query);
  }
}

Monitoring & Debugging

Performance Monitoring

Track sync performance metrics:

import { signal, computed } from '@angular/core';

class SyncPerformanceMonitor {
  private metrics = signal({
    totalSyncs: 0,
    successfulSyncs: 0,
    failedSyncs: 0,
    avgSyncDuration: 0,
    totalDataSynced: 0,
    peakSyncRate: 0
  });

  readonly successRate = computed(() => {
    const m = this.metrics();
    return m.totalSyncs > 0 
      ? (m.successfulSyncs / m.totalSyncs) * 100 
      : 0;
  });

  readonly syncsPerMinute = computed(() => {
    // Calculate based on time window
    return this.calculateSyncRate();
  });

  async trackSync<T>(operation: () => Promise<T>): Promise<T> {
    const startTime = performance.now();
    const startMetrics = this.metrics();

    try {
      const result = await operation();
      const duration = performance.now() - startTime;

      this.metrics.set({
        ...startMetrics,
        totalSyncs: startMetrics.totalSyncs + 1,
        successfulSyncs: startMetrics.successfulSyncs + 1,
        avgSyncDuration: (startMetrics.avgSyncDuration * startMetrics.totalSyncs + duration) / (startMetrics.totalSyncs + 1)
      });

      return result;
    } catch (error) {
      this.metrics.set({
        ...startMetrics,
        totalSyncs: startMetrics.totalSyncs + 1,
        failedSyncs: startMetrics.failedSyncs + 1
      });
      throw error;
    }
  }

  private calculateSyncRate(): number {
    // Implementation for rate calculation
    return 0;
  }
}

Debug Logging

Implement structured logging for debugging:

enum LogLevel {
  DEBUG = 0,
  INFO = 1,
  WARN = 2,
  ERROR = 3
}

class SyncLogger {
  private level: LogLevel = LogLevel.INFO;
  private logs: Array<{ level: LogLevel; message: string; timestamp: number; metadata?: any }> = [];

  setLevel(level: LogLevel): void {
    this.level = level;
  }

  debug(message: string, metadata?: any): void {
    this.log(LogLevel.DEBUG, message, metadata);
  }

  info(message: string, metadata?: any): void {
    this.log(LogLevel.INFO, message, metadata);
  }

  warn(message: string, metadata?: any): void {
    this.log(LogLevel.WARN, message, metadata);
  }

  error(message: string, metadata?: any): void {
    this.log(LogLevel.ERROR, message, metadata);
  }

  private log(level: LogLevel, message: string, metadata?: any): void {
    if (level < this.level) return;

    const entry = {
      level,
      message,
      timestamp: Date.now(),
      metadata
    };

    this.logs.push(entry);
    console.log(`[${LogLevel[level]}] ${message}`, metadata);
  }

  exportLogs(): string {
    return JSON.stringify(this.logs, null, 2);
  }
}

Migration Strategies

Zero-Downtime Migration

Migrate existing data without downtime:

class ZeroDowntimeMigration {
  async migrate(sourceAdapter: any, targetAdapter: MongoDBSyncAdapter): Promise<void> {
    // Phase 1: Dual-write
    await this.enableDualWrite(sourceAdapter, targetAdapter);

    // Phase 2: Backfill historical data
    await this.backfillData(sourceAdapter, targetAdapter);

    // Phase 3: Validate data consistency
    await this.validateConsistency(sourceAdapter, targetAdapter);

    // Phase 4: Switch reads to new system
    await this.switchReads(targetAdapter);

    // Phase 5: Disable dual-write
    await this.disableDualWrite();
  }

  private async enableDualWrite(source: any, target: MongoDBSyncAdapter): Promise<void> {
    // Intercept all writes and send to both systems
    console.log('Dual-write enabled');
  }

  private async backfillData(source: any, target: MongoDBSyncAdapter): Promise<void> {
    const collections = await source.listCollections();
    
    for (const collection of collections) {
      await this.backfillCollection(source, target, collection);
    }
  }

  private async backfillCollection(
    source: any,
    target: MongoDBSyncAdapter,
    collection: string
  ): Promise<void> {
    const batchSize = 100;
    let offset = 0;
    let batch: any[];

    do {
      batch = await source.find(collection, {}, { limit: batchSize, skip: offset });
      
      if (batch.length > 0) {
        await target.bulkPush(collection, batch);
        offset += batch.length;
      }
    } while (batch.length === batchSize);

    console.log(`Backfilled ${offset} documents from ${collection}`);
  }

  private async validateConsistency(source: any, target: MongoDBSyncAdapter): Promise<void> {
    // Sample-based validation
    console.log('Validating data consistency...');
  }

  private async switchReads(target: MongoDBSyncAdapter): Promise<void> {
    // Configure application to read from new system
    console.log('Switched reads to MongoDB');
  }

  private async disableDualWrite(): Promise<void> {
    // Stop writing to old system
    console.log('Dual-write disabled');
  }
}

Next Steps

• 📖 API Reference - Complete API documentation
• 🚀 Getting Started - Quick setup guide
• 📊 Overview - Architecture and design

Questions? Check our troubleshooting guide or visit our GitHub Discussions.