crdt_grpc_sync 1.0.1

A gRPC network sync layer for crdt, mirrors & inspired by crdt_sync, to get full P2P, local-first sync in any Dart app.

A gRPC network sync layer for crdt, mirrors & inspired by crdt_sync.

• gRPC-Powered: Synchronizes Crdt nodes in real-time via persistent, bidirectional gRPC streams.
• BYO Crdt Backend: Bring your own DB, existing support for sqlite_crdt, hive_crdt, postgres_crdt.
• Multiple Topology Support: Use relayChanges to toggle between hub/chain or mesh mode.
• Resilient: Auto-reconnect with configurable backoff (uses defaultBackoffStrategy from grpc package), catch-up sync for late joiners and message chunking.
• Exntensible Pipeline: Custom handshake data, changeset builders, record validators/transformers, per-table event hooks.

Pair with implementations below to have P2P, local-first sync for your app.

Tested and works with:

• sqlite_crdt
• hive_crdt
• postgres_crdt
• Or create your own custom implementation via CrdtSyncConfig:
  • recordFormat: CrdtRecordFormat.custom
  • mapOutgoingChangeset
  • mapIncomingChangeset

Huge thanks & credits go to cachapa as the author of the crdt engine that powers this.

Warning

• Co-authored with AI. I created architecture, AI drafted implementation, then artisanally hand refactored.
• I'm no Dart expert, so there may be language-specific intricacies on handling async processes that were missed.
• Everything in example/ and test/ was created by AI.

Usage

Minimal server

import 'package:crdt/map_crdt.dart';
import 'package:crdt_grpc_sync/crdt_grpc_sync.dart';

Future<void> main() async {
  final crdt = MapCrdt(['chat']);
  await listen(crdt, 8080, config: CrdtSyncServiceConfig(
    onClientConnected: (crdtSync, data) {
      print('Peer connected: ${crdtSync.peerId}');
    },
  ));
}

Minimal client

import 'package:crdt/map_crdt.dart';
import 'package:crdt_grpc_sync/crdt_grpc_sync.dart';

Future<void> main() async {
  final crdt = MapCrdt(['chat']);
  final client = CrdtSyncClient(crdt, Uri.parse('grpc://localhost:8080'));
  await client.connect();
}

See example/ for working examples.

API

Tip

listen(Crdt crdt, int port, {CrdtSyncServiceConfig config}) — Start gRPC server, returns CrdtSyncServer.

CrdtSyncServer class — Bundles gRPC Server + CrdtSyncService with properties: server, port & method: shutdown().

Customization

All customization via config objects. Pass CrdtSyncConfig / CrdtSyncServiceConfig to listen() or CrdtSyncClient().

// Defaults are as below
// Shared config (applied to both server and client)
final syncConfig = CrdtSyncConfig(
  changesetBuilder: null,          // Custom changeset construction, null = default crdt.getChangeset
  validateRecord: null,            // Reject records before merge, null = no validation
  mapIncomingChangeset: null,      // Transform after decode, null = no transforms after decode
  mapOutgoingChangeset: null,      // Transform before encode, null = no transforms before encode
  recordFormat: null,              // Set record format (sql/noSql/custom, auto-detected if null)
  onChangesetReceived: null,       // Called before merge — notification only, reports peerId + per-table record count, null = passthrough
  onChangesetSent: null,           // Called after send — notification only, reports peerId + per-table record count, null = passthrough
  chunkSize: 1000,                 // Records per gRPC message, required
  relayChanges: true,              // true = hub/chain, false = mesh, required
  verbose: false,                  // Log payloads to console
);

  // Server (example: port 8080)
await listen(crdt, 8080, config: CrdtSyncServiceConfig(
  handshakeDataBuilder: null,      // Build custom handshake response — receives (peerId, peerData) from client, null = no custom data in resp
  onClientConnecting: null,        // Fires on Handshake RPC received — receives ServiceCall (IP, headers), null = no-op
  onClientConnected: null,         // Fires on Sync session active — receives (CrdtSync, decoded peerData), null = no-op
  onClientDisconnected: null,      // Fires on Peer left — receives (peerId, code, reason), null = no-op
  syncConfig: CrdtSyncConfig(),    // Shared config (defaults above)
));

// Client (example: grpc://host:8080)
final client = CrdtSyncClient(
  crdt,
  Uri.parse('grpc://host:8080'),
  handshakeDataBuilder: null,      // Build handshake metadata — returns data sent to server, null = no custom data in req
  onConnecting: null,              // Fires before handshake attempt, null = no-op
  onConnected: null,               // Fires on Handshake done, sync active — receives (serverPeerId, decodedServerData), null = no-op
  onDisconnected: null,            // Fires when Sync session closed — receives (peerId, code, reason), null = no-op
  onError: null,                   // Fires on non-fatal error — receives (error, stackTrace), null = errors silently consumed
  retryDelayBuilder: null,         // Custom backoff fn, null = uses grpc defaultBackoffStrategy (exponential); fn returning null = stop retrying
  maxRetries: null,                // Max reconnect attempts, null = infinite
  channelOptions: null,            // Exposes gRPC ChannelOptions (credentials, user-agent, etc), null = default grpc ChannelOptions class (secure/insecure based on URI scheme)
  syncConfig: CrdtSyncConfig(),    // Shared config (defaults above)
);

Feature Req and Bugs

File an issue or start a discussion — I'll get around to it when I can!