serjunken1990/mongo-sync-service

By serjunken1990

•Updated 11 months ago

Synchronizes data between collections in mongo with mongo replica with a simple configuration file

Image

Databases & storage

211

Overview Tags

serjunken1990/mongo-sync-service repository overview

⁠MongoDB Sync Service

Lightweight service for propagating changes across MongoDB collections using Change Streams with support for nested fields and flexible field mapping.

⁠Features

External Configuration: Define sync rules in a separate JS file
Nested Field Support: Read from and write to nested fields using dot notation
Flexible Field Mapping: Map source fields to different target field names
Lightweight: Node.js with minimal dependencies
Easy to extend: Add new rules by editing config file
Monitoring: Built-in health and metrics endpoints
Graceful shutdown: Handles SIGTERM/SIGINT properly
Docker ready: Includes Dockerfile and compose configuration

⁠Project Structure

project/
├── mongo-sync-service/
│   ├── MongoSyncManager.js    # Core framework
│   ├── index.js                # Entry point
│   ├── package.json
│   ├── Dockerfile
│   └── .dockerignore
├── sync-config.js              # Your configuration (external)
└── docker-compose.yml

⁠Quick Start

⁠1. Create the service directory

mkdir mongo-sync-service
cd mongo-sync-service

⁠2. Copy the framework files

Create the following files:

MongoSyncManager.js (core framework)
index.js (entry point)
package.json
Dockerfile
.dockerignore

⁠3. Create your configuration file

Create sync-config.js in your project root (not inside mongo-sync-service):

module.exports = function(syncManager) {
    // Simple example: propagate user changes to jobs
    syncManager.addRule({
        source: {
            collection: 'USERS_TABLE',
            matchField: '_id',
            fields: {
                '_id': 'USER.USER_ID',
                'NAME': 'USER.USER_NAME',
                'EMAIL': 'USER.USER_EMAIL'
            }
        },
        target: {
            collection: 'JOBS_ENRICHED',
            matchField: 'USER.USER_ID'
        }
    });
};

⁠4. Build the Docker image

docker build -t mongo-sync-service ./mongo-sync-service

⁠5. Add to docker-compose.yml

services:
  mongo-sync-service:
    build: ./mongo-sync-service
    container_name: mongo-sync-service
    depends_on:
      - mongo
    environment:
      - MONGO_URI=mongodb://mongo:27017/?replicaSet=rs0
      - MONGO_DATABASE=pixpro_readmodel
      - CONFIG_FILE=/config/sync-config.js
    volumes:
      - ./sync-config.js:/config/sync-config.js:ro
    restart: unless-stopped
    ports:
      - "8080:8080"

⁠6. Start the service

docker-compose up -d mongo-sync-service

⁠7. Check logs

docker-compose logs -f mongo-sync-service

⁠Configuration Examples

⁠Example 1: Simple - Same field names

module.exports = function(syncManager) {
    syncManager.addRule({
        source: {
            collection: 'USERS_TABLE',
            matchField: '_id',
            fields: ['name', 'email']  // Same names in both collections
        },
        target: {
            collection: 'JOBS_ENRICHED',
            matchField: 'user_id'
        }
    });
};

⁠Example 2: Field mapping (different names)

module.exports = function(syncManager) {
    syncManager.addRule({
        source: {
            collection: 'USERS_TABLE',
            matchField: '_id',
            fields: {
                name: 'user_name',       // USERS.name -> JOBS.user_name
                email: 'user_email'      // USERS.email -> JOBS.user_email
            }
        },
        target: {
            collection: 'JOBS_ENRICHED',
            matchField: 'user_id'
        }
    });
};

⁠Example 3: Nested target fields (flat to nested)

module.exports = function(syncManager) {
    syncManager.addRule({
        source: {
            collection: 'USERS_TABLE',
            matchField: '_id',
            fields: {
                '_id': 'USER.USER_ID',           // Flat to nested
                'NAME': 'USER.USER_NAME',        // Flat to nested
                'EMAIL': 'USER.USER_EMAIL'       // Flat to nested
            }
        },
        target: {
            collection: 'JOBS_ENRICHED',
            matchField: 'USER.USER_ID'           // Match on nested field
        }
    });
};

Result in JOBS_ENRICHED:

{
  "_id": 2,
  "STATUS": "PENDING",
  "USER": {
    "USER_ID": 2,
    "USER_NAME": "John Doe",
    "USER_EMAIL": "[email protected]"
  },
  "_synced_at": "2025-10-24T12:00:00Z"
}

⁠Example 4: Nested source fields (nested to flat)

module.exports = function(syncManager) {
    syncManager.addRule({
        source: {
            collection: 'USERS_TABLE',
            matchField: '_id',
            fields: {
                'profile.name': 'user_name',           // Nested to flat
                'profile.avatar': 'user_avatar',       // Nested to flat
                'contact.email': 'user_email'          // Nested to flat
            }
        },
        target: {
            collection: 'JOBS_ENRICHED',
            matchField: 'user_id'
        }
    });
};

⁠Example 5: Nested to nested

module.exports = function(syncManager) {
    syncManager.addRule({
        source: {
            collection: 'USERS_TABLE',
            matchField: '_id',
            fields: {
                'profile.firstName': 'USER.PROFILE.FIRST_NAME',
                'profile.lastName': 'USER.PROFILE.LAST_NAME',
                'contact.email': 'USER.CONTACT.EMAIL'
            }
        },
        target: {
            collection: 'ORDERS_ENRICHED',
            matchField: 'USER.ID'
        }
    });
};

⁠Example 6: Complex nested structures

module.exports = function(syncManager) {
    syncManager.addRule({
        source: {
            collection: 'PRODUCTS_TABLE',
            matchField: '_id',
            fields: {
                '_id': 'PRODUCT.ID',
                'name': 'PRODUCT.INFO.NAME',
                'description': 'PRODUCT.INFO.DESCRIPTION',
                'pricing.current': 'PRODUCT.PRICING.CURRENT',
                'pricing.original': 'PRODUCT.PRICING.ORIGINAL',
                'images.main': 'PRODUCT.IMAGES.MAIN',
                'images.thumbnail': 'PRODUCT.IMAGES.THUMB',
                'inventory.stock': 'PRODUCT.STOCK.AVAILABLE'
            }
        },
        target: {
            collection: 'ORDER_ITEMS_ENRICHED',
            matchField: 'PRODUCT.ID'
        }
    });
};

Result:

{
  "PRODUCT": {
    "ID": 123,
    "INFO": { "NAME": "...", "DESCRIPTION": "..." },
    "PRICING": { "CURRENT": 99.99, "ORIGINAL": 149.99 },
    "IMAGES": { "MAIN": "...", "THUMB": "..." },
    "STOCK": { "AVAILABLE": 50 }
  }
}

⁠Example 7: Multiple targets for same source

module.exports = function(syncManager) {
    // Users to Jobs
    syncManager.addRule({
        source: {
            collection: 'USERS_TABLE',
            matchField: '_id',
            fields: { name: 'user_name', email: 'user_email' }
        },
        target: {
            collection: 'JOBS_ENRICHED',
            matchField: 'user_id'
        }
    });
    
    // Users to Orders (different field names)
    syncManager.addRule({
        source: {
            collection: 'USERS_TABLE',
            matchField: '_id',
            fields: { name: 'customer_name', email: 'customer_email' }
        },
        target: {
            collection: 'ORDERS_ENRICHED',
            matchField: 'customer_id'
        }
    });
};

⁠Example 8: With filter (only verified users)

module.exports = function(syncManager) {
    syncManager.addRule({
        source: {
            collection: 'USERS_TABLE',
            matchField: '_id',
            fields: { name: 'user_name', email: 'user_email' }
        },
        target: {
            collection: 'JOBS_ENRICHED',
            matchField: 'user_id'
        },
        filter: (change, document) => {
            // Only sync verified and active users
            return document.verified === true && document.status === 'active';
        }
    });
};

⁠Example 9: With transformation

module.exports = function(syncManager) {
    syncManager.addRule({
        source: {
            collection: 'USERS_TABLE',
            matchField: '_id',
            fields: {
                'profile.firstName': 'first_name',
                'profile.lastName': 'last_name',
                'email': 'email'
            }
        },
        target: {
            collection: 'JOBS_ENRICHED',
            matchField: 'user_id'
        },
        transform: (updateDoc, fullDocument) => {
            return {
                ...updateDoc,
                // Add computed fields
                full_name: `${updateDoc.first_name} ${updateDoc.last_name}`,
                email: updateDoc.email?.toLowerCase(),
                // Access full document
                is_premium: fullDocument.subscription?.tier === 'premium',
                display_name: updateDoc.first_name?.toUpperCase()
            };
        }
    });
};

⁠Example 10: Chaining multiple rules

module.exports = function(syncManager) {
    syncManager
        .addRule({
            source: { 
                collection: 'USERS_TABLE', 
                matchField: '_id', 
                fields: ['name', 'email'] 
            },
            target: { collection: 'JOBS_ENRICHED', matchField: 'user_id' }
        })
        .addRule({
            source: { 
                collection: 'USERS_TABLE', 
                matchField: '_id', 
                fields: ['name'] 
            },
            target: { collection: 'ORDERS_ENRICHED', matchField: 'user_id' }
        })
        .addRule({
            source: { 
                collection: 'PRODUCTS_TABLE', 
                matchField: '_id', 
                fields: { name: 'product_name', 'pricing.current': 'price' } 
            },
            target: { collection: 'ORDER_ITEMS', matchField: 'product_id' }
        });
};

⁠API Reference

⁠addRule(config)

Main method to add a synchronization rule.

Parameters:

config.source (Object) - Source configuration
- collection (string) - Source collection name
- matchField (string) - Field to match from (supports dot notation, default: '_id')
- fields (Array|Object) - Fields to extract
  - Array: ['name', 'email'] - same names in target
  - Object: { name: 'user_name', email: 'user_email' } - field mapping
config.target (Object) - Target configuration
- collection (string) - Target collection name
- matchField (string) - Field to match to (supports dot notation)
config.filter (Function) - Optional filter function (change, document) => boolean
config.transform (Function) - Optional transform function (updateDoc, fullDocument) => object

⁠propagate(sourceCollection, targetCollection, sourceMatchField, targetMatchField, fields)

Backward compatible simplified method.

Parameters:

sourceCollection (string) - Source collection name
targetCollection (string) - Target collection name
sourceMatchField (string) - Field in source to match from
targetMatchField (string) - Field in target to match to
fields (Array) - Fields to propagate

⁠propagateUser(userCollection, targets)

Helper method to propagate user changes to multiple collections.

Parameters:

userCollection (string) - User collection name
targets (Array) - Array of target configurations
- collection (string) - Target collection name
- sourceMatchField (string) - Source match field (default: '_id')
- targetMatchField (string) - Target match field (default: 'user_id')
- fields (Array|Object) - Fields to propagate

⁠Monitoring Endpoints

⁠Health Check

curl http://localhost:8080/health

Response:

{
  "status": "healthy",
  "totalChangesDetected": 156,
  "totalDocumentsUpdated": 342,
  "errors": 0,
  "uptimeSeconds": 3600
}

⁠Metrics

curl http://localhost:8080/metrics

⁠Rules

curl http://localhost:8080/rules

Response:

{
  "rules": [
    {
      "source": "USERS_TABLE",
      "target": "JOBS_ENRICHED",
      "matchField": "USER.USER_ID",
      "fields": ["_id->USER.USER_ID", "NAME->USER.USER_NAME"]
    }
  ],
  "total": 1
}

⁠Environment Variables

Variable	Default	Description
`MONGO_URI`	`mongodb://mongo:27017/?replicaSet=rs0`	MongoDB connection URI (must be replica set)
`MONGO_DATABASE`	`pixpro_readmodel`	Database name
`CONFIG_FILE`	`/config/sync-config.js`	Path to config file
`PORT`	`8080`	Health check port

⁠Adding New Tables

Simply edit sync-config.js and add a new rule:

module.exports = function(syncManager) {
    // Existing rule
    syncManager.addRule({
        source: { collection: 'USERS_TABLE', matchField: '_id', fields: ['name', 'email'] },
        target: { collection: 'JOBS_ENRICHED', matchField: 'user_id' }
    });
    
    // New rule
    syncManager.addRule({
        source: { collection: 'USERS_TABLE', matchField: '_id', fields: ['name', 'email'] },
        target: { collection: 'ORDERS_ENRICHED', matchField: 'user_id' }
    });
};

Then restart the service:

docker-compose restart mongo-sync-service

⁠Troubleshooting

⁠Service not starting

Check logs:

docker-compose logs mongo-sync-service

Common issues:

MongoDB not configured as replica set
Configuration file syntax error
MongoDB connection URI incorrect

⁠Configuration not loading

Verify the volume mount:

docker exec mongo-sync-service ls -la /config

Test configuration syntax:

node -c sync-config.js

⁠No changes being synced

MongoDB replica set: Change streams require MongoDB to run as a replica set
```
mongo:
  command: --replSet rs0
```
Verify collection names: Collection names are case-sensitive
```
docker exec mongo mongosh --eval "db.getMongo().getDBNames()"
```

Check field names: Use the exact field names from your documents

docker exec mongo mongosh pixpro_readmodel --eval "db.USERS_TABLE.findOne()"

Review logs: Check for warnings about missing fields
```
docker-compose logs -f mongo-sync-service | grep WARN
```

⁠Nested fields not updating

Ensure you're using the correct dot notation:

Source: 'profile.name' - reads from nested field
Target: 'USER.NAME' - writes to nested field

Test the field path:

// In sync-config.js, add logging in transform
transform: (updateDoc, fullDocument) => {
    console.log('Update doc:', JSON.stringify(updateDoc));
    console.log('Full document:', JSON.stringify(fullDocument));
    return updateDoc;
}

⁠High memory usage

Reduce change stream batch size: Add filters to process fewer events
Limit field mappings: Only sync fields you need

Add time-based filters: Skip old changes

filter: (change) => {
    const hourAgo = Date.now() - (60 * 60 * 1000);
    return change.clusterTime.getHighBits() * 1000 > hourAgo;
}

⁠Development

⁠Local development

cd mongo-sync-service
npm install
MONGO_URI=mongodb://localhost:27017/?replicaSet=rs0 \
MONGO_DATABASE=pixpro_readmodel \
CONFIG_FILE=../sync-config.js \
npm run dev

⁠Testing configuration

Test your configuration without running the full service:

// test-config.js
const MongoSyncManager = require('./MongoSyncManager');
const config = require('./sync-config.js');

const manager = new MongoSyncManager('mongodb://localhost:27017/?replicaSet=rs0', 'test_db');
config(manager);

console.log('Loaded rules:', manager.rules.size);
manager.rules.forEach((rules, collection) => {
    console.log(`\n${collection}:`);
    rules.forEach(rule => {
        console.log(`  -> ${rule.targetCollection}`);
        console.log(`     Fields:`, rule.fieldMapping);
    });
});

Tag summary

Recent tags

Content type

Image

Digest

sha256:0f56e65fa…

Size

47.2 MB

Last updated

11 months ago

docker pull serjunken1990/mongo-sync-service:1.0.0