synclib-sync

TypeScript/JavaScript sync client for coordinating database changes with Elixir Phoenix server. Works in tandem with synclib to provide real-time bidirectional sync between your local SQLite database and a Phoenix server.

Features

• 🔄 Bidirectional Sync - Push local changes and receive remote updates
• 🔌 Phoenix Channels - Built on Phoenix WebSocket channels for real-time communication
• 🔁 Automatic Reconnection - Handles disconnections and reconnects automatically
• 📦 Batch Operations - Efficient batching of changes for better performance
• 🔀 Conflict Resolution - Pluggable conflict resolution strategies
• 📊 Schema Migrations - Automatic schema version management and migrations
• 📸 Snapshot Streaming - Efficient initial data loading with incremental updates
• 💬 Real-time Events - Support for livestream, conversation, and job update events
• 🎯 TypeScript - Full type safety and IntelliSense support

Installation

npm install synclib-sync synclib
# or
pnpm add synclib-sync synclib
# or
yarn add synclib-sync synclib

Quick Start

import { SynclibDatabase } from 'synclib';
import { SyncClient } from 'synclib-sync';

// 1. Open your local database
const db = await SynclibDatabase.open({
  databaseName: 'myapp.db'
});

// 2. Create tables
db.exec(`
  CREATE TABLE IF NOT EXISTS users (
    id TEXT PRIMARY KEY,
    name TEXT NOT NULL,
    email TEXT NOT NULL
  )
`);

// 3. Create sync client
const syncClient = new SyncClient({
  db,
  serverUrl: 'wss://api.example.com/socket',
  clientId: 'user-123',
  initialChannels: [
    {
      channelName: 'user',
      channelId: 'user-123',
    },
  ],
  pushInterval: 5000, // Push changes every 5 seconds
});

// 4. Initialize and connect
await syncClient.initialize();
await syncClient.connect('your-auth-token');

// 5. Stream initial data
await syncClient.streamSnapshot(['users']);

// 6. Listen for remote changes
syncClient.onRemoteChange((change) => {
  console.log('Remote change applied:', change);
  // Refresh your UI
});

// 7. Make local changes (they'll be synced automatically)
await db.write({
  tableName: 'users',
  rowId: '456',
  operation: SynclibOperation.INSERT,
  sql: `INSERT INTO users (id, name, email) VALUES (?, ?, ?)`,
  params: ['456', 'Jane Doe', '[email protected]'],
  data: JSON.stringify({ id: '456', name: 'Jane Doe', email: '[email protected]' })
});

Usage with Svelte

Here's a complete example of integrating synclib-sync in a Svelte application:

// src/lib/stores/sync.ts
import { writable } from 'svelte/store';
import { SynclibDatabase } from 'synclib';
import { SyncClient, type SyncClientConfig } from 'synclib-sync';

export const db = writable<SynclibDatabase | null>(null);
export const syncClient = writable<SyncClient | null>(null);
export const isConnected = writable(false);
export const isSyncing = writable(false);

export async function initializeSync(userId: string, token: string) {
  // Open database
  const database = await SynclibDatabase.open({
    databaseName: 'myapp.db'
  });

  // Run migrations
  const currentVersion = database.getSchemaVersion();
  if (currentVersion < 1) {
    database.exec(`
      CREATE TABLE IF NOT EXISTS todos (
        id TEXT PRIMARY KEY,
        title TEXT NOT NULL,
        completed INTEGER DEFAULT 0
      )
    `);
    await database.setSchemaVersion(1);
  }

  // Create sync client
  const client = new SyncClient({
    db: database,
    serverUrl: 'wss://api.example.com/socket',
    clientId: userId,
    initialChannels: [
      {
        channelName: 'user',
        channelId: userId,
      },
    ],
    pushInterval: 5000,
  });

  // Listen for connection state changes
  client.onStateChange((state) => {
    isConnected.set(state === 'connected');
  });

  // Listen for remote changes
  client.onRemoteChange((change) => {
    console.log('Remote change:', change);
    // Trigger UI refresh
  });

  // Initialize and connect
  await client.initialize();
  await client.connect(token);

  // Stream initial data
  isSyncing.set(true);
  await client.streamSnapshot(['todos']);

  // Wait for snapshot to complete
  await new Promise<void>((resolve) => {
    const unsubscribe = client.onSnapshotComplete(() => {
      unsubscribe();
      resolve();
    });
  });
  isSyncing.set(false);

  db.set(database);
  syncClient.set(client);

  return { database, client };
}

<!-- src/routes/+layout.svelte -->
<script lang="ts">
  import { onMount } from 'svelte';
  import { initializeSync, isConnected, isSyncing } from '$lib/stores/sync';

  let userId = 'user-123';
  let token = 'your-auth-token';

  onMount(async () => {
    await initializeSync(userId, token);
  });
</script>

{#if $isSyncing}
  <div class="loading">Syncing data...</div>
{/if}

{#if !$isConnected}
  <div class="offline">Offline - changes will sync when reconnected</div>
{/if}

<slot />

<!-- src/routes/+page.svelte -->
<script lang="ts">
  import { db } from '$lib/stores/sync';
  import { SynclibOperation } from 'synclib';

  let todos: Array<{ id: string; title: string; completed: number }> = [];
  let newTodoTitle = '';

  $: if ($db) {
    loadTodos();
  }

  function loadTodos() {
    if (!$db) return;
    todos = $db.read('SELECT * FROM todos ORDER BY id');
  }

  async function addTodo() {
    if (!$db || !newTodoTitle.trim()) return;

    const id = crypto.randomUUID();

    await $db.write({
      tableName: 'todos',
      rowId: id,
      operation: SynclibOperation.INSERT,
      sql: `INSERT INTO todos (id, title, completed) VALUES (?, ?, ?)`,
      params: [id, newTodoTitle, 0],
      data: JSON.stringify({ id, title: newTodoTitle, completed: 0 })
    });

    newTodoTitle = '';
    loadTodos();
  }
</script>

<div>
  <h1>Todos</h1>

  <input
    bind:value={newTodoTitle}
    on:keypress={(e) => e.key === 'Enter' && addTodo()}
    placeholder="Add a todo"
  />
  <button on:click={addTodo}>Add</button>

  {#each todos as todo}
    <div>
      <input
        type="checkbox"
        checked={todo.completed === 1}
        on:change={() => toggleTodo(todo.id, todo.completed)}
      />
      <span>{todo.title}</span>
    </div>
  {/each}
</div>

API Reference

SyncClient

Main class for managing sync with the server.

Constructor

new SyncClient(config: SyncClientConfig)

Config Options:

interface SyncClientConfig {
  // Required
  db: SynclibDatabase;           // Synclib database instance
  serverUrl: string;              // WebSocket server URL
  clientId: string;               // Unique client identifier
  initialChannels: SyncClientChannel[]; // Channels to join on connect

  // Optional
  pushInterval?: number;          // How often to push changes (ms), null = manual
  pushBatchSize?: number;         // Max changes per push (default: 100)
  pullInterval?: number | null;   // How often to pull changes (ms), null = reactive
  metadata?: Record<string, any>; // Custom metadata for hello message
}

interface SyncClientChannel {
  channelName: string;   // Channel type (e.g., 'user', 'tribe', 'guild')
  channelId: string;     // Specific channel ID
  params?: Record<string, string>; // Additional channel params
}

Methods

async initialize(): Promise<void>

Initialize the sync client. Must be called before connect().

async connect(token: string): Promise<void>

Connect to the sync server with authentication token.

async disconnect(): Promise<void>

Disconnect from the sync server.

async sync(): Promise<void>

Manually trigger a sync cycle (push local changes, pull remote changes).

async joinChannel(channel: SyncClientChannel): Promise<void>

Join an additional channel after initial connection.

async leaveChannel(channel: SyncClientChannel): Promise<void>

Leave a channel.

isChannelJoined(channel: SyncClientChannel): boolean

Check if a channel is currently joined.

getJoinedChannels(): string[]

Get all currently joined channel topics.

async streamSnapshot(tables: string[], options?: { incremental?: boolean; channelTopic?: string }): Promise<void>

Stream a snapshot of tables from the server.

• tables: Array of table names to stream
• options.incremental: If true, only stream changes since last sync
• options.channelTopic: Specific channel to use for the request

async fetchRow(table: string, rowId: string): Promise<Record<string, any>>

Fetch a single row from the server (including JSONB fields).

async sendMessage(event: string, payload: Record<string, any>, channelTopic?: string): Promise<Record<string, any>>

Send a custom message to the server and wait for reply.

async sendConversationPresence(options): Promise<void>

Send a conversation presence event (user_joined or user_left).

await syncClient.sendConversationPresence({
  conversationId: 'tribe_123',
  userId: 'user_456',
  event: 'conversation:user_joined',
});

setConflictResolver(resolver: ConflictResolver): void

Set a custom conflict resolution function.

syncClient.setConflictResolver(async (local, remote) => {
  // Return the change to apply, or null to skip
  // Example: last-write-wins
  return remote.timestamp! > local.timestamp! ? remote : local;
});

Event Listeners

All event listeners return an unsubscribe function.

onRemoteChange(listener: (change: ChangeMessage) => void): () => void

Called when a remote change is applied to the local database.

onSnapshotComplete(listener: (streamId: string) => void): () => void

Called when a snapshot stream completes.

onJobUpdate(listener: (update: JobUpdateMessage) => void): () => void

Called when a job update is received (e.g., from ECS tasks).

onLivestream(listener: (message: LivestreamMessage) => void): () => void

Called when a livestream event is received (started/stopped).

onConversation(listener: (message: ConversationMessage) => void): () => void

Called when a conversation event is received (user presence, messages).

onSyncReadyStateChange(listener: (state: SyncReadyState) => void): () => void

Called when the sync ready state changes.

States:

• WAITING_FOR_HELLO: Waiting for initial hello reply
• APPLYING_MIGRATIONS: Applying schema migrations
• READY: Ready to stream snapshots and sync data

onStateChange(listener: (state: ConnectionState) => void): () => void

Called when the WebSocket connection state changes.

States:

• DISCONNECTED: Not connected
• CONNECTING: Connecting to server
• CONNECTED: Connected and ready
• RECONNECTING: Attempting to reconnect
• FAILED: Connection failed

Properties

connectionState: ConnectionState (readonly)

Current connection state.

isReady: boolean (readonly)

Whether the client is ready to stream snapshots (true when ready state is READY).

currentReadyState: SyncReadyState (readonly)

Current sync ready state.

Disposal

async dispose(): Promise<void>

Clean up resources and disconnect.

Advanced Usage

Conflict Resolution

Handle conflicts when local and remote changes affect the same row:

syncClient.setConflictResolver(async (local, remote) => {
  // Last-write-wins strategy
  if (remote.timestamp! > local.timestamp!) {
    return remote;
  }
  return local;

  // Or custom merge logic
  // return {
  //   ...remote,
  //   data: {
  //     ...local.data,
  //     ...remote.data,
  //   }
  // };

  // Or skip the change
  // return null;
});

Incremental Snapshots

For large datasets, use incremental snapshots to only fetch new data:

// First sync - get all data
await syncClient.streamSnapshot(['users', 'posts']);

// Later syncs - only get changes
await syncClient.streamSnapshot(['users', 'posts'], {
  incremental: true
});

Custom Channel Messages

Send custom messages to your Phoenix channels:

const response = await syncClient.sendMessage('custom_event', {
  key: 'value',
  data: { foo: 'bar' }
});

console.log('Server response:', response);

Multiple Channels

Join multiple channels for different data scopes:

const syncClient = new SyncClient({
  // ...
  initialChannels: [
    { channelName: 'user', channelId: userId },
    { channelName: 'tribe', channelId: tribeId },
    { channelName: 'guild', channelId: guildId },
  ],
});

// Or join dynamically
await syncClient.joinChannel({
  channelName: 'party',
  channelId: partyId,
});

Server-Side Integration

This library is designed to work with an Elixir Phoenix server using Phoenix Channels. Your server should implement:

1. Channel authentication - Verify tokens on channel join
2. Change broadcasting - Broadcast changes to relevant channels
3. Snapshot streaming - Stream table data in batches
4. Schema migrations - Manage and distribute schema versions

Example Phoenix channel (Elixir):

defmodule MyAppWeb.SyncChannel do
  use Phoenix.Channel

  def join("sync:user:" <> user_id, _params, socket) do
    # Authenticate and authorize
    {:ok, socket}
  end

  def handle_in("hello", %{"schema_version" => version}, socket) do
    # Check schema version and send migrations if needed
    {:reply, {:ok, %{status: "up_to_date"}}, socket}
  end

  def handle_in("changes_batch", %{"changes" => changes}, socket) do
    # Process and broadcast changes
    Enum.each(changes, fn change ->
      broadcast!(socket, "change", change)
    end)
    {:reply, {:ok, %{}}, socket}
  end

  def handle_in("stream_snapshot", %{"tables" => tables}, socket) do
    # Stream table data in batches
    Task.start(fn ->
      Enum.each(tables, fn table ->
        stream_table_data(socket, table)
      end)
    end)
    {:reply, {:ok, %{}}, socket}
  end
end

Troubleshooting

Connection Issues

If you're having trouble connecting:

1. Check WebSocket URL is correct (should start with wss:// or ws://)
2. Verify authentication token is valid
3. Check server logs for connection errors
4. Ensure CORS is properly configured on your server

Changes Not Syncing

If changes aren't syncing:

1. Check connection state: syncClient.connectionState
2. Verify channels are joined: syncClient.getJoinedChannels()
3. Check for errors in browser console
4. Verify pushInterval is set (or call sync() manually)

Schema Mismatch

If you get schema version errors:

1. The client will automatically apply migrations from the server
2. Listen to onSyncReadyStateChange to track migration progress
3. Ensure your server is sending proper migration SQL

License

MIT