Workflow 003: Data Sync Pipeline

Fresh Last Updated: January 2025

Field	Value
Workflow ID	WF-003
Version	1.0
Complexity	High
Estimated Setup	4-8 hours

Overview

Build a bidirectional sync pipeline between Notion and external systems (CRM, project management tools, databases). Handle conflict resolution, incremental updates, and data transformation.

Use Cases

Sync Notion contacts with Salesforce/HubSpot CRM
Mirror GitHub issues to Notion database
Sync Notion tasks with Jira/Asana
Replicate Notion data to PostgreSQL for analytics
Two-way sync between Notion and Airtable

Architecture

Sync Strategy Comparison

Strategy	Complexity	Use When
One-Way	Low	Single source of truth
Two-Way	High	Both systems need edits
Hub & Spoke	Medium	Multiple systems, central control

Prerequisites

[ ] Completed SOP 002: Authentication
[ ] External system API credentials
[ ] Persistent storage for sync state (Redis, SQLite, etc.)
[ ] Understanding of both data models

Implementation

Step 1: Project Setup

bash

mkdir notion-sync-pipeline
cd notion-sync-pipeline
npm init -y
npm install @notionhq/client dotenv better-sqlite3 axios

Create .env:

env

NOTION_API_KEY=secret_xxxxx
NOTION_DATABASE_ID=xxxxx
EXTERNAL_API_KEY=xxxxx
EXTERNAL_API_URL=https://api.example.com
SYNC_DB_PATH=./sync-state.db

Step 2: Sync State Database

javascript

// src/sync-state.js
const Database = require('better-sqlite3');

class SyncState {
  constructor(dbPath) {
    this.db = new Database(dbPath);
    this.init();
  }

  init() {
    this.db.exec(`
      CREATE TABLE IF NOT EXISTS sync_records (
        id TEXT PRIMARY KEY,
        notion_id TEXT,
        external_id TEXT,
        notion_updated_at TEXT,
        external_updated_at TEXT,
        sync_status TEXT DEFAULT 'synced',
        last_synced_at TEXT,
        checksum TEXT,
        metadata TEXT
      );

      CREATE INDEX IF NOT EXISTS idx_notion_id ON sync_records(notion_id);
      CREATE INDEX IF NOT EXISTS idx_external_id ON sync_records(external_id);
      CREATE INDEX IF NOT EXISTS idx_sync_status ON sync_records(sync_status);

      CREATE TABLE IF NOT EXISTS sync_logs (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        timestamp TEXT DEFAULT CURRENT_TIMESTAMP,
        operation TEXT,
        direction TEXT,
        record_id TEXT,
        status TEXT,
        details TEXT
      );
    `);
  }

  getByNotionId(notionId) {
    return this.db.prepare(
      'SELECT * FROM sync_records WHERE notion_id = ?'
    ).get(notionId);
  }

  getByExternalId(externalId) {
    return this.db.prepare(
      'SELECT * FROM sync_records WHERE external_id = ?'
    ).get(externalId);
  }

  getAllPending() {
    return this.db.prepare(
      'SELECT * FROM sync_records WHERE sync_status = ?'
    ).all('pending');
  }

  upsert(record) {
    const existing = this.db.prepare(
      'SELECT id FROM sync_records WHERE notion_id = ? OR external_id = ?'
    ).get(record.notion_id, record.external_id);

    if (existing) {
      this.db.prepare(`
        UPDATE sync_records SET
          notion_id = ?,
          external_id = ?,
          notion_updated_at = ?,
          external_updated_at = ?,
          sync_status = ?,
          last_synced_at = ?,
          checksum = ?,
          metadata = ?
        WHERE id = ?
      `).run(
        record.notion_id,
        record.external_id,
        record.notion_updated_at,
        record.external_updated_at,
        record.sync_status || 'synced',
        new Date().toISOString(),
        record.checksum,
        JSON.stringify(record.metadata || {}),
        existing.id
      );
    } else {
      const id = `sync_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
      this.db.prepare(`
        INSERT INTO sync_records (
          id, notion_id, external_id, notion_updated_at,
          external_updated_at, sync_status, last_synced_at, checksum, metadata
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
      `).run(
        id,
        record.notion_id,
        record.external_id,
        record.notion_updated_at,
        record.external_updated_at,
        record.sync_status || 'synced',
        new Date().toISOString(),
        record.checksum,
        JSON.stringify(record.metadata || {})
      );
    }
  }

  log(operation, direction, recordId, status, details = '') {
    this.db.prepare(`
      INSERT INTO sync_logs (operation, direction, record_id, status, details)
      VALUES (?, ?, ?, ?, ?)
    `).run(operation, direction, recordId, status, details);
  }

  getRecentLogs(limit = 100) {
    return this.db.prepare(
      'SELECT * FROM sync_logs ORDER BY timestamp DESC LIMIT ?'
    ).all(limit);
  }
}

module.exports = SyncState;

Step 3: Data Transformers

javascript

// src/transformers.js
const crypto = require('crypto');

class DataTransformer {
  // Notion -> External
  static notionToExternal(notionPage) {
    const props = notionPage.properties;

    return {
      id: notionPage.id,
      name: props.Name?.title?.[0]?.plain_text || '',
      email: props.Email?.email || '',
      company: props.Company?.rich_text?.[0]?.plain_text || '',
      status: props.Status?.select?.name || null,
      phone: props.Phone?.phone_number || '',
      notes: props.Notes?.rich_text?.[0]?.plain_text || '',
      tags: props.Tags?.multi_select?.map(t => t.name) || [],
      createdAt: notionPage.created_time,
      updatedAt: notionPage.last_edited_time,
      source: 'notion'
    };
  }

  // External -> Notion
  static externalToNotion(externalRecord) {
    return {
      Name: {
        title: [{ text: { content: externalRecord.name || '' } }]
      },
      Email: {
        email: externalRecord.email || null
      },
      Company: {
        rich_text: [{ text: { content: externalRecord.company || '' } }]
      },
      Status: externalRecord.status
        ? { select: { name: externalRecord.status } }
        : { select: null },
      Phone: {
        phone_number: externalRecord.phone || null
      },
      Notes: {
        rich_text: [{ text: { content: externalRecord.notes || '' } }]
      },
      Tags: {
        multi_select: (externalRecord.tags || []).map(t => ({ name: t }))
      },
      'External ID': {
        rich_text: [{ text: { content: externalRecord.id || '' } }]
      }
    };
  }

  // Generate checksum for conflict detection
  static generateChecksum(data) {
    const normalized = JSON.stringify(data, Object.keys(data).sort());
    return crypto.createHash('md5').update(normalized).digest('hex');
  }

  // Compare for changes
  static hasChanges(record1, record2) {
    const checksum1 = this.generateChecksum(record1);
    const checksum2 = this.generateChecksum(record2);
    return checksum1 !== checksum2;
  }
}

module.exports = DataTransformer;

Step 4: External API Client

javascript

// src/external-client.js
const axios = require('axios');

class ExternalAPIClient {
  constructor(config) {
    this.client = axios.create({
      baseURL: config.baseUrl,
      headers: {
        'Authorization': `Bearer ${config.apiKey}`,
        'Content-Type': 'application/json'
      },
      timeout: 30000
    });
  }

  async getRecords(params = {}) {
    const response = await this.client.get('/contacts', { params });
    return response.data;
  }

  async getRecord(id) {
    const response = await this.client.get(`/contacts/${id}`);
    return response.data;
  }

  async createRecord(data) {
    const response = await this.client.post('/contacts', data);
    return response.data;
  }

  async updateRecord(id, data) {
    const response = await this.client.patch(`/contacts/${id}`, data);
    return response.data;
  }

  async deleteRecord(id) {
    await this.client.delete(`/contacts/${id}`);
  }

  async getUpdatedSince(timestamp) {
    const response = await this.client.get('/contacts', {
      params: { updated_since: timestamp }
    });
    return response.data;
  }
}

module.exports = ExternalAPIClient;

Step 5: Sync Engine

javascript

// src/sync-engine.js
const { Client } = require('@notionhq/client');
const SyncState = require('./sync-state');
const DataTransformer = require('./transformers');
const ExternalAPIClient = require('./external-client');
require('dotenv').config();

class SyncEngine {
  constructor() {
    this.notion = new Client({ auth: process.env.NOTION_API_KEY });
    this.databaseId = process.env.NOTION_DATABASE_ID;
    this.syncState = new SyncState(process.env.SYNC_DB_PATH);
    this.external = new ExternalAPIClient({
      baseUrl: process.env.EXTERNAL_API_URL,
      apiKey: process.env.EXTERNAL_API_KEY
    });

    this.conflictStrategy = 'latest-wins'; // or 'notion-wins', 'external-wins'
  }

  async fullSync() {
    console.log('🔄 Starting full sync...\n');

    // 1. Fetch all records from both systems
    const notionRecords = await this.fetchAllFromNotion();
    const externalRecords = await this.external.getRecords();

    console.log(`  Notion: ${notionRecords.length} records`);
    console.log(`  External: ${externalRecords.length} records\n`);

    // 2. Build lookup maps
    const notionByExtId = new Map();
    const externalById = new Map();

    for (const record of notionRecords) {
      const extId = record.properties['External ID']?.rich_text?.[0]?.plain_text;
      if (extId) notionByExtId.set(extId, record);
    }

    for (const record of externalRecords) {
      externalById.set(record.id, record);
    }

    // 3. Process each external record
    for (const extRecord of externalRecords) {
      const notionRecord = notionByExtId.get(extRecord.id);

      if (!notionRecord) {
        // Create in Notion
        await this.createInNotion(extRecord);
      } else {
        // Check for updates
        await this.syncPair(notionRecord, extRecord);
      }

      await this.sleep(350);
    }

    // 4. Check for Notion records without external counterpart
    for (const notionRecord of notionRecords) {
      const extId = notionRecord.properties['External ID']?.rich_text?.[0]?.plain_text;

      if (!extId) {
        // Create in external system
        await this.createInExternal(notionRecord);
        await this.sleep(350);
      }
    }

    console.log('\n✅ Full sync complete');
  }

  async incrementalSync() {
    console.log('🔄 Starting incremental sync...\n');

    // Get last sync timestamp
    const lastSync = this.getLastSyncTime();

    // 1. Get changes from Notion
    const notionChanges = await this.getNotionChangesSince(lastSync);
    console.log(`  Notion changes: ${notionChanges.length}`);

    // 2. Get changes from external system
    const externalChanges = await this.external.getUpdatedSince(lastSync);
    console.log(`  External changes: ${externalChanges.length}`);

    // 3. Process Notion -> External
    for (const notionRecord of notionChanges) {
      const syncRecord = this.syncState.getByNotionId(notionRecord.id);

      if (syncRecord?.external_id) {
        await this.pushToExternal(notionRecord, syncRecord.external_id);
      } else {
        await this.createInExternal(notionRecord);
      }

      await this.sleep(350);
    }

    // 4. Process External -> Notion
    for (const extRecord of externalChanges) {
      const syncRecord = this.syncState.getByExternalId(extRecord.id);

      if (syncRecord?.notion_id) {
        await this.pushToNotion(extRecord, syncRecord.notion_id);
      } else {
        await this.createInNotion(extRecord);
      }

      await this.sleep(350);
    }

    console.log('\n✅ Incremental sync complete');
  }

  async syncPair(notionRecord, externalRecord) {
    const syncRecord = this.syncState.getByNotionId(notionRecord.id);

    const notionUpdated = new Date(notionRecord.last_edited_time);
    const externalUpdated = new Date(externalRecord.updated_at);
    const lastSynced = syncRecord ? new Date(syncRecord.last_synced_at) : new Date(0);

    const notionChanged = notionUpdated > lastSynced;
    const externalChanged = externalUpdated > lastSynced;

    if (notionChanged && externalChanged) {
      // Conflict!
      await this.resolveConflict(notionRecord, externalRecord);
    } else if (notionChanged) {
      // Push to external
      await this.pushToExternal(notionRecord, externalRecord.id);
    } else if (externalChanged) {
      // Push to Notion
      await this.pushToNotion(externalRecord, notionRecord.id);
    }
    // else: No changes
  }

  async resolveConflict(notionRecord, externalRecord) {
    console.log(`⚠️ Conflict detected: ${notionRecord.id}`);

    const notionUpdated = new Date(notionRecord.last_edited_time);
    const externalUpdated = new Date(externalRecord.updated_at);

    let winner;

    switch (this.conflictStrategy) {
      case 'latest-wins':
        winner = notionUpdated > externalUpdated ? 'notion' : 'external';
        break;
      case 'notion-wins':
        winner = 'notion';
        break;
      case 'external-wins':
        winner = 'external';
        break;
    }

    console.log(`  Resolution: ${winner} wins`);

    if (winner === 'notion') {
      await this.pushToExternal(notionRecord, externalRecord.id);
    } else {
      await this.pushToNotion(externalRecord, notionRecord.id);
    }

    this.syncState.log(
      'conflict_resolved',
      winner,
      notionRecord.id,
      'success',
      JSON.stringify({ strategy: this.conflictStrategy })
    );
  }

  async createInNotion(externalRecord) {
    console.log(`  Creating in Notion: ${externalRecord.name}`);

    try {
      const properties = DataTransformer.externalToNotion(externalRecord);

      const response = await this.notion.pages.create({
        parent: { database_id: this.databaseId },
        properties
      });

      this.syncState.upsert({
        notion_id: response.id,
        external_id: externalRecord.id,
        notion_updated_at: response.last_edited_time,
        external_updated_at: externalRecord.updated_at,
        checksum: DataTransformer.generateChecksum(externalRecord)
      });

      this.syncState.log('create', 'external->notion', response.id, 'success');
      console.log(`    ✅ Created: ${response.id}`);

    } catch (error) {
      this.syncState.log('create', 'external->notion', externalRecord.id, 'error', error.message);
      console.error(`    ❌ Error: ${error.message}`);
    }
  }

  async createInExternal(notionRecord) {
    console.log(`  Creating in external: ${notionRecord.properties.Name?.title?.[0]?.plain_text}`);

    try {
      const data = DataTransformer.notionToExternal(notionRecord);
      const response = await this.external.createRecord(data);

      // Update Notion with external ID
      await this.notion.pages.update({
        page_id: notionRecord.id,
        properties: {
          'External ID': {
            rich_text: [{ text: { content: response.id } }]
          }
        }
      });

      this.syncState.upsert({
        notion_id: notionRecord.id,
        external_id: response.id,
        notion_updated_at: notionRecord.last_edited_time,
        external_updated_at: response.updated_at,
        checksum: DataTransformer.generateChecksum(data)
      });

      this.syncState.log('create', 'notion->external', notionRecord.id, 'success');
      console.log(`    ✅ Created: ${response.id}`);

    } catch (error) {
      this.syncState.log('create', 'notion->external', notionRecord.id, 'error', error.message);
      console.error(`    ❌ Error: ${error.message}`);
    }
  }

  async pushToExternal(notionRecord, externalId) {
    console.log(`  Updating external: ${externalId}`);

    try {
      const data = DataTransformer.notionToExternal(notionRecord);
      const response = await this.external.updateRecord(externalId, data);

      this.syncState.upsert({
        notion_id: notionRecord.id,
        external_id: externalId,
        notion_updated_at: notionRecord.last_edited_time,
        external_updated_at: response.updated_at,
        checksum: DataTransformer.generateChecksum(data)
      });

      this.syncState.log('update', 'notion->external', notionRecord.id, 'success');

    } catch (error) {
      this.syncState.log('update', 'notion->external', notionRecord.id, 'error', error.message);
      console.error(`    ❌ Error: ${error.message}`);
    }
  }

  async pushToNotion(externalRecord, notionId) {
    console.log(`  Updating Notion: ${notionId}`);

    try {
      const properties = DataTransformer.externalToNotion(externalRecord);

      const response = await this.notion.pages.update({
        page_id: notionId,
        properties
      });

      this.syncState.upsert({
        notion_id: notionId,
        external_id: externalRecord.id,
        notion_updated_at: response.last_edited_time,
        external_updated_at: externalRecord.updated_at,
        checksum: DataTransformer.generateChecksum(externalRecord)
      });

      this.syncState.log('update', 'external->notion', notionId, 'success');

    } catch (error) {
      this.syncState.log('update', 'external->notion', notionId, 'error', error.message);
      console.error(`    ❌ Error: ${error.message}`);
    }
  }

  async fetchAllFromNotion() {
    const pages = [];
    let cursor = undefined;

    do {
      const response = await this.notion.databases.query({
        database_id: this.databaseId,
        start_cursor: cursor,
        page_size: 100
      });

      pages.push(...response.results);
      cursor = response.has_more ? response.next_cursor : undefined;
    } while (cursor);

    return pages;
  }

  async getNotionChangesSince(timestamp) {
    const response = await this.notion.databases.query({
      database_id: this.databaseId,
      filter: {
        timestamp: 'last_edited_time',
        last_edited_time: { after: timestamp }
      }
    });

    return response.results;
  }

  getLastSyncTime() {
    // Get from sync state or default to 24 hours ago
    const logs = this.syncState.getRecentLogs(1);
    if (logs.length > 0) {
      return logs[0].timestamp;
    }
    const yesterday = new Date();
    yesterday.setDate(yesterday.getDate() - 1);
    return yesterday.toISOString();
  }

  sleep(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}

module.exports = SyncEngine;

Step 6: Main Runner

javascript

// src/index.js
const SyncEngine = require('./sync-engine');

const engine = new SyncEngine();
const mode = process.argv[2] || 'incremental';

async function run() {
  console.log(`🚀 Notion Sync Pipeline - ${mode} mode\n`);

  switch (mode) {
    case 'full':
      await engine.fullSync();
      break;
    case 'incremental':
      await engine.incrementalSync();
      break;
    case 'status':
      const logs = engine.syncState.getRecentLogs(20);
      console.log('Recent sync activity:');
      console.table(logs);
      break;
    default:
      console.log('Usage: node src/index.js [full|incremental|status]');
  }
}

run().catch(err => {
  console.error('❌ Sync failed:', err);
  process.exit(1);
});

Conflict Resolution Strategies

Verification Checklist

[ ] Initial full sync completes successfully
[ ] Incremental sync detects changes
[ ] Conflicts resolved per strategy
[ ] Sync state persists between runs
[ ] Rate limits respected
[ ] Errors logged properly
[ ] Data transforms correctly both ways

Troubleshooting

Issue	Cause	Solution
Duplicate records	No ID mapping	Ensure External ID is set
Data loss	Wrong conflict strategy	Change to appropriate strategy
Sync loop	Missing checksum	Track checksums to detect actual changes
Memory issues	Too many records	Implement batching

Workflow 003: Data Sync Pipeline ​

Overview ​

Use Cases ​

Architecture ​

Sync Strategy Comparison ​

Prerequisites ​

Implementation ​

Step 1: Project Setup ​

Step 2: Sync State Database ​

Step 3: Data Transformers ​

Step 4: External API Client ​

Step 5: Sync Engine ​

Step 6: Main Runner ​

Conflict Resolution Strategies ​

Verification Checklist ​

Troubleshooting ​

See Also ​

Workflow 003: Data Sync Pipeline

Overview

Use Cases

Architecture

Sync Strategy Comparison

Prerequisites

Implementation

Step 1: Project Setup

Step 2: Sync State Database

Step 3: Data Transformers

Step 4: External API Client

Step 5: Sync Engine

Step 6: Main Runner

Conflict Resolution Strategies

Verification Checklist

Troubleshooting

See Also