ZenithDB Sync

Backend-agnostic sync engine that provides powerful offline-first capabilities as an enhanced wrapper around IndexedDB, featuring optimistic updates, conflict resolution, and seamless data synchronization for local-first applications.

Description

@zenithdb/sync provides a comprehensive synchronization solution that extends Zenith's IndexedDB wrapper with robust offline-first capabilities. It handles queued operations when offline, automatic conflict resolution, and provides a type-safe API for real-time data synchronization with any backend through a transport adapter pattern.

Key Features

• 🔄 Backend-Agnostic: Works with REST APIs, GraphQL, or custom protocols as a sync layer over IndexedDB
• 📱 Offline-First: Queue operations when offline, sync when reconnected with full IndexedDB persistence
• ⚡ Optimistic Updates: Immediate UI updates with background synchronization via IndexedDB wrapper
• 🔀 Conflict Resolution: Multiple strategies including custom resolution functions for IndexedDB conflicts
• 📊 Type-Safe Events: Full TypeScript support with typed event system for enhanced IndexedDB operations
• 🔁 Retry Logic: Exponential backoff for failed operations with IndexedDB queue persistence
• 💾 Persistent Queue: IndexedDB-based queue survives page reloads and browser restarts
• 🛡️ Data Integrity: Transaction-safe operations ensuring IndexedDB consistency

Installation

# Install sync package with storage and types
npm install @zenithdb/sync @zenithdb/storage @zenithdb/types

# Or install everything at once (recommended)
npm install @zenithdb/kit

Quick Start

Basic Setup

import { ZenithSyncEngine, createRestTransport } from "@zenithdb/sync";
import { IndexedDBAdapter } from "@zenithdb/storage";

// Configure your backend transport
const transport = createRestTransport({
  baseUrl: "https://api.yourapp.com",
  endpoints: {
    push: "/sync/push",
    pull: "/sync/pull",
  },
  headers: {
    Authorization: "Bearer your-token",
  },
});

// Set up storage adapter
const adapter = new IndexedDBAdapter("myapp");

// Create sync engine
const syncEngine = new ZenithSyncEngine(transport, adapter, {
  conflictResolution: "server-wins",
  autoSyncInterval: 10000, // Sync every 30 seconds
  retry: {
    maxAttempts: 5,
    initialDelay: 1000,
    backoffMultiplier: 2,
    maxDelay: 100000,
  },
});

// Start syncing
await syncEngine.start();

Making Changes

// Add operations to sync queue
await syncEngine.enqueue({
  operation: "create",
  table: "users",
  key: "user-123",
  data: { name: "John Doe", email: "[email protected]" },
  priority: 1,
});

await syncEngine.enqueue({
  operation: "update",
  table: "posts",
  key: "post-456",
  data: { title: "Updated Title" },
  priority: 2,
});

// Operations are automatically synced in the background
// Or manually trigger sync
await syncEngine.push();

Transport Adapters

REST Transport

Perfect for standard HTTP APIs:

const restTransport = createRestTransport({
  baseUrl: "https://api.example.com",
  endpoints: {
    push: "/sync/push",
    pull: "/sync/pull",
  },
  headers: {
    "Content-Type": "application/json",
    Authorization: "Bearer token",
  },
  timeout: 10000, // 10 seconds,
});

GraphQL Transport

Type-safe GraphQL integration:

const graphqlTransport = createGraphQLTransport({
  endpoint: "https://api.example.com/graphql",
  headers: {
    Authorization: "Bearer token",
  },
  mutations: {
    pushData: `
      mutation SyncPush($operations: [SyncOperationInput!]!) {
        syncPush(operations: $operations) {
          success
          conflicts { id table localValue remoteValue }
        }
      }
    `,
  },
  queries: {
    pullData: `
      query SyncPull($lastSync: DateTime) {
        syncPull(since: $lastSync) {
          operations { id table data timestamp }
          cursor
        }
      }
    `,
  },
});

Event Handling

The sync engine emits typed events for monitoring and debugging:

// Sync lifecycle events
syncEngine.on("sync:start", () => {
  console.log("Sync started");
});

syncEngine.on("sync:complete", (stats) => {
  console.log(`Sync completed: ${stats.pushed} pushed, ${stats.pulled} pulled`);
});

syncEngine.on("sync:error", (error) => {
  console.error("Sync failed:", error);
});

// Conflict handling
syncEngine.on("sync:conflict", (conflict) => {
  console.log(`Conflict in ${conflict.table}:`, conflict);
  // Show conflict resolution UI
  showConflictDialog(conflict);
});

// Queue monitoring
syncEngine.on("queue:change", (size) => {
  updateQueueIndicator(size);
});

// Connection status
syncEngine.on("connection:change", (isOnline) => {
  updateConnectionIndicator(isOnline);
});

Conflict Resolution

Built-in Strategies

const syncEngine = new ZenithSyncEngine(transport, adapter, {
  // Server data takes precedence
  conflictResolution: "server-wins",

  // Local data takes precedence
  // conflictResolution: 'client-wins',

  // Most recent timestamp wins
  // conflictResolution: 'last-write-wins'
});

Custom Resolution

const syncEngine = new ZenithSyncEngine(transport, adapter, {
  conflictResolution: "custom",
  customResolver: async (conflict) => {
    const { local, remote, table, key } = conflict;

    if (table === "users") {
      // Merge user preferences
      return {
        ...remote,
        preferences: { ...remote.preferences, ...local.preferences },
      };
    }

    if (table === "documents") {
      // Use operational transforms for documents
      return await operationalTransform(local, remote);
    }

    // Default to server wins
    return remote;
  },
});

Offline Capabilities

The sync engine automatically handles offline scenarios:

// Operations work the same offline or online
await syncEngine.enqueue({
  operation: "create",
  table: "messages",
  key: "msg-789",
  data: { text: "Hello offline world!" },
});

// When connection is restored, queued operations sync automatically
syncEngine.on("connection:change", (isOnline) => {
  if (isOnline) {
    console.log("Back online! Syncing queued operations...");
    // Sync happens automatically
  }
});

Queue Management

Monitor and manage the sync queue:

// Get current queue status
const queue = await syncEngine.getQueue();
console.log(`${queue.length} operations pending`);

// Filter by status
const failedOps = queue.filter((item) => item.status === "failed");
const pendingOps = queue.filter((item) => item.status === "pending");

// Clean up completed operations
await syncEngine.cleanup();

// Queue size monitoring
console.log(`Current queue size: ${syncEngine.queueSize}`);

Backend Integration Examples

Custom Node.js API

// Server-side sync endpoint
app.post("/sync/push", async (req, res) => {
  const { operations } = req.body;
  const conflicts = [];

  for (const op of operations) {
    try {
      // Apply operation to database
      await applyOperation(op);
    } catch (error) {
      if (error.code === "CONFLICT") {
        conflicts.push({
          id: op.id,
          table: op.table,
          localValue: op.data,
          remoteValue: await getCurrentValue(op.table, op.key),
        });
      }
    }
  }

  res.json({ success: true, conflicts });
});

// Client transport
const customTransport = createRestTransport({
  baseUrl: "https://yourapi.com",
  endpoints: {
    push: "/sync/push",
    pull: "/sync/pull",
  },
});

Advanced Configuration

Retry Logic

const syncEngine = new ZenithSyncEngine(transport, adapter, {
  retry: {
    maxAttempts: 3, // Try up to 3 times
    initialDelay: 2000, // Start with 2 second delay
    backoffMultiplier: 1.5, // Increase delay by 1.5x each retry
    maxDelay: 10000, // Cap at 30 seconds
  },
});

Batch Configuration

const syncEngine = new ZenithSyncEngine(transport, adapter, {
  batchSize: 50, // Send 50 operations per request
  autoSyncInterval: 60000, // Auto-sync every 60 seconds
});

Hooks

syncEngine.setHooks({
  beforePush: async (items) => {
    // Transform items before sending
    return items.map((item) => ({
      ...item,
      clientId: getClientId(),
    }));
  },

  afterPush: async (items) => {
    // Analytics or cleanup after successful push
    analytics.track("sync_pushed", { count: items.length });
  },

  onConflict: async (local, remote) => {
    // Custom conflict handling
    return await showConflictResolutionDialog(local, remote);
  },
});