ETL/Sync Job Builder

Build reliable, incremental data synchronization pipelines.

ETL Job Pattern

// jobs/sync-users.ts interface SyncJob { name: string; source: "database" | "api" | "file"; destination: "database" | "warehouse" | "s3"; schedule: string; }

export class ETLJob { constructor(private name: string, private watermarkKey: string) {}

async run() { console.log(🔄 Starting ${this.name}...);

try {
  // 1. Get last watermark
  const lastSync = await this.getWatermark();
  console.log(`  Last sync: ${lastSync}`);

  // 2. Extract data
  const data = await this.extract(lastSync);
  console.log(`  Extracted ${data.length} records`);

  // 3. Transform data
  const transformed = await this.transform(data);

  // 4. Load data
  await this.load(transformed);

  // 5. Update watermark
  await this.updateWatermark(new Date());

  console.log(`✅ ${this.name} complete`);
} catch (error) {
  console.error(`❌ ${this.name} failed:`, error);
  throw error;
}

}

private async extract(since: Date) { // Extract logic return []; }

private async transform(data: any[]) { // Transform logic return data; }

private async load(data: any[]) { // Load logic }

private async getWatermark(): Promise<Date> { const watermark = await prisma.syncWatermark.findUnique({ where: { key: this.watermarkKey }, }); return watermark?.lastSync || new Date(0); }

private async updateWatermark(timestamp: Date) { await prisma.syncWatermark.upsert({ where: { key: this.watermarkKey }, create: { key: this.watermarkKey, lastSync: timestamp }, update: { lastSync: timestamp }, }); } }

Watermark Strategy

// Track sync progress model SyncWatermark { key String @id lastSync DateTime metadata Json?

@@index([lastSync]) }

// Incremental sync using watermark async function syncOrdersIncremental() { // Get last sync time const watermark = await prisma.syncWatermark.findUnique({ where: { key: "orders_sync" }, });

const lastSync = watermark?.lastSync || new Date(0);

// Fetch only new/updated records const newOrders = await sourceDb.order.findMany({ where: { updated_at: { gt: lastSync }, }, orderBy: { updated_at: "asc" }, });

console.log(📦 Syncing ${newOrders.length} orders...);

// Process in batches for (let i = 0; i < newOrders.length; i += 100) { const batch = newOrders.slice(i, i + 100);

await destinationDb.order.createMany({
  data: batch,
  skipDuplicates: true, // Idempotency
});

}

// Update watermark to latest record if (newOrders.length > 0) { const latestTimestamp = newOrders[newOrders.length - 1].updated_at;

await prisma.syncWatermark.upsert({
  where: { key: "orders_sync" },
  create: { key: "orders_sync", lastSync: latestTimestamp },
  update: { lastSync: latestTimestamp },
});

}

console.log(✅ Sync complete); }

Idempotent Upsert Pattern

// Idempotent sync - safe to run multiple times async function syncUsersIdempotent(users: User[]) { for (const user of users) { await prisma.user.upsert({ where: { id: user.id }, create: user, update: { email: user.email, name: user.name, updated_at: user.updated_at, }, }); } }

// Batch upsert for better performance async function syncUsersBatch(users: User[]) { // PostgreSQL: Use ON CONFLICT await prisma.$executeRaw INSERT INTO users (id, email, name, updated_at) SELECT * FROM UNNEST( ${users.map((u) => u.id)}::bigint[], ${users.map((u) => u.email)}::text[], ${users.map((u) => u.name)}::text[], ${users.map((u) => u.updated_at)}::timestamp[] ) ON CONFLICT (id) DO UPDATE SET email = EXCLUDED.email, name = EXCLUDED.name, updated_at = EXCLUDED.updated_at WHERE users.updated_at < EXCLUDED.updated_at ; }

Retry Logic with Exponential Backoff

async function syncWithRetry<T>( operation: () => Promise<T>, maxRetries: number = 3, baseDelay: number = 1000 ): Promise<T> { for (let attempt = 0; attempt <= maxRetries; attempt++) { try { return await operation(); } catch (error) { if (attempt === maxRetries) throw error;

  const delay = baseDelay * Math.pow(2, attempt);
  console.log(`  Retry ${attempt + 1}/${maxRetries} after ${delay}ms`);
  await sleep(delay);
}

}

throw new Error("Max retries exceeded"); }

// Usage await syncWithRetry( async () => { return await syncOrders(); }, 3, 1000 );

Change Data Capture (CDC)

// Listen to database changes import { PrismaClient } from "@prisma/client";

const prisma = new PrismaClient();

// PostgreSQL: Listen to logical replication async function setupCDC() { await prisma.$executeRaw CREATE PUBLICATION orders_publication FOR TABLE orders; ;

// Subscribe to changes (using pg library) const client = await pg.connect();

client.query("LISTEN orders_changed;");

client.on("notification", async (msg) => { const change = JSON.parse(msg.payload);

if (change.operation === "INSERT" || change.operation === "UPDATE") {
  await syncOrder(change.data);
}

}); }

Conflict Resolution

interface ConflictResolution { strategy: "source-wins" | "dest-wins" | "latest-wins" | "merge"; }

async function syncWithConflictResolution( sourceRecord: any, destRecord: any, strategy: ConflictResolution["strategy"] ) { if (strategy === "source-wins") { return sourceRecord; }

if (strategy === "dest-wins") { return destRecord; }

if (strategy === "latest-wins") { return sourceRecord.updated_at > destRecord.updated_at ? sourceRecord : destRecord; }

if (strategy === "merge") { // Merge non-null fields return { ...destRecord, ...Object.fromEntries( Object.entries(sourceRecord).filter(([_, v]) => v != null) ), }; } }

Monitoring & Observability

// Track sync job metrics interface SyncMetrics { jobName: string; startTime: Date; endTime: Date; recordsProcessed: number; recordsInserted: number; recordsUpdated: number; recordsSkipped: number; errors: number; durationMs: number; }

async function logSyncMetrics(metrics: SyncMetrics) { await prisma.syncMetric.create({ data: metrics, });

console.log(📊 Sync Metrics Job: ${metrics.jobName} Records: ${metrics.recordsProcessed} Inserted: ${metrics.recordsInserted} Updated: ${metrics.recordsUpdated} Errors: ${metrics.errors} Duration: ${metrics.durationMs}ms ); }

Full ETL Job Example

// jobs/sync-orders-to-warehouse.ts export class OrdersETLJob extends ETLJob { constructor() { super("orders-etl", "orders_warehouse_sync"); }

async extract(since: Date): Promise<Order[]> { return prisma.order.findMany({ where: { updated_at: { gt: since }, }, include: { items: true, user: true, }, orderBy: { updated_at: "asc" }, }); }

async transform(orders: Order[]): Promise<WarehouseOrder[]> { return orders.map((order) => ({ order_id: order.id, user_email: order.user.email, total_amount: order.total, item_count: order.items.length, status: order.status, order_date: order.created_at, synced_at: new Date(), })); }

async load(data: WarehouseOrder[]): Promise<void> { const batchSize = 100;

for (let i = 0; i < data.length; i += batchSize) {
  const batch = data.slice(i, i + batchSize);

  await warehouseDb.$executeRaw`
    INSERT INTO orders_fact (
      order_id, user_email, total_amount, item_count,
      status, order_date, synced_at
    )
    VALUES ${batch
      .map(
        (o) => `(
      ${o.order_id}, '${o.user_email}', ${o.total_amount},
      ${o.item_count}, '${o.status}', '${o.order_date}',
      '${o.synced_at}'
    )`
      )
      .join(",")}
    ON CONFLICT (order_id) DO UPDATE SET
      total_amount = EXCLUDED.total_amount,
      status = EXCLUDED.status,
      synced_at = EXCLUDED.synced_at
  `;
}

} }

// Run job new OrdersETLJob().run();

Scheduling

// Schedule ETL jobs import cron from "node-cron";

// Run every hour cron.schedule("0 * * * *", async () => { await new OrdersETLJob().run(); });

// Run every 15 minutes cron.schedule("*/15 * * * *", async () => { await syncUsersIncremental(); });

// Run nightly at 2 AM cron.schedule("0 2 * * *", async () => { await fullDataSync(); });

Error Handling & Recovery

async function syncWithErrorHandling() { const checkpoint = await getCheckpoint(); let processedRecords = 0;

try { const records = await fetchRecords(checkpoint);

for (const record of records) {
  try {
    await processRecord(record);
    processedRecords++;

    // Save checkpoint every 100 records
    if (processedRecords % 100 === 0) {
      await saveCheckpoint(record.id);
    }
  } catch (error) {
    // Log error but continue
    console.error(`Failed to process record ${record.id}:`, error);
    await logFailedRecord(record.id, error);
  }
}

await saveCheckpoint("completed");

} catch (error) { // Critical failure - job will retry from checkpoint console.error("Job failed:", error); throw error; } }

Best Practices

• Incremental sync: Use watermarks, don't full-scan

• Idempotent operations: Safe to retry

• Batch processing: Process 100-1000 records at a time

• Checkpointing: Resume from failure point

• Retry with backoff: Handle transient failures

• Monitor metrics: Track job health

• Test thoroughly: Including failure scenarios

Output Checklist

• ETL job class created

• Watermark tracking implemented

• Incremental sync logic

• Idempotent upsert operations

• Retry logic with backoff

• Conflict resolution strategy

• Monitoring and metrics

• Error handling and recovery

• Job scheduling configured

• Testing including failure cases