Merkle-CRDT sync

This package provides a synchronization layer for MerkleCRDT instances (from the merkledag package) over a dart_libp2p network. It facilitates the replication of Merkle-CRDTs among multiple peers by:

• Announcing new CRDT heads (root CIDs) using libp2p GossipSub.
• Fetching required Merkle-DAG nodes directly from peers via libp2p streams.
• Utilizing a Kademlia DHT (if available) for discovering peers that can provide specific DAG nodes.

It’s designed to be used in conjunction with the merkledag package for the core Merkle-CRDT logic and dart_libp2p for the underlying peer-to-peer networking.

Features

• Merkle-CRDT State Synchronization: Keeps Merkle-CRDT instances eventually consistent across multiple libp2p peers.
• GossipSub for Head Announcements: Efficiently disseminates new CRDT state heads (root CIDs) using dart_libp2p_pubsub.
• Direct P2P Stream-based DAG Syncing: Fetches only the necessary Merkle-DAG nodes directly from peers, minimizing redundant data transfer.
• DHT Provider Discovery: Can leverage a Kademlia DHT (e.g., dart_libp2p_kad_dht) to find peers holding specific DAG nodes.
• Local Block Storage: Includes an in-memory block store (InMemoryBlockStore) and an interface (BlockStore) for custom persistent storage solutions.
• Centralized Management: Provides a MerkleCrdtGossipManager class to orchestrate the CRDT instance, DAG syncer, and broadcaster.
• Generic: Works with any CRDT payload type P that implements CRDTPayload<V> from the merkledag package.

Core Components

• MerkleCrdtGossipManager<V, P>: The main class an application interacts with. It manages a MerkleCRDT instance and coordinates synchronization.
• P2PStreamDagFetcher<P, V>: Implements the DAGSyncer<P> interface from merkledag. Responsible for fetching and storing MerkleNode<P> objects. Uses libp2p streams for direct transfer and can query a DHT for providers.
• GossipSubAnnouncer: Implements the Broadcaster interface from merkledag. Uses dart_libp2p_pubsub to broadcast new head CIDs and subscribe to remote head announcements.
• BlockStore<P, V>: Interface for storing MerkleNode<P> objects locally.
  • InMemoryBlockStore<P, V>: A simple in-memory implementation.

Getting Started

Prerequisites

• A working Dart environment.
• A configured libp2p Host instance from dart_libp2p.
• A PubSub instance from dart_libp2p_pubsub (typically using GossipSubRouter).
• Optionally, an IpfsDHT instance from dart_libp2p_kad_dht if DHT provider discovery is desired.

Add Dependencies

Add this package and its peer dependencies to your pubspec.yaml. Since these are likely local/path dependencies during development:

dependencies:
  # This package
  dart_libp2p_merkle_crdt:
    path: [path_to_this_package] # Replace with actual path or version if published

  # Core Merkle-CRDT logic
  merkledag:
    path: [path_to_merkledag_package] # e.g., /Users/stephanfeb/IdeaProjects/bsv_apps/merkledag

  # Libp2p stack
  dart_libp2p:
    path: [path_to_dart_libp2p_package] # e.g., /Users/stephanfeb/IdeaProjects/dart-libp2p
  dart_libp2p_pubsub:
    path: [path_to_dart_libp2p_pubsub_package] # e.g., /Users/stephanfeb/IdeaProjects/bsv_apps/dart-libp2p-pubsub
  dart_libp2p_kad_dht: # Optional, for DHT support in P2PStreamDagFetcher
    path: [path_to_dart_libp2p_kat_dht_package] # e.g., /Users/stephanfeb/IdeaProjects/bsv_apps/dart-libp2p-kat-dht

  # Other common dependencies
  dart_cid:
    path: [path_to_dart_cid_package] # e.g., /Users/stephanfeb/IdeaProjects/bsv_apps/dart-cid
  logging: ^1.2.0 # Or your preferred logging package version

Run dart pub get.

Usage

Here’s an example of how to set up and use the MerkleCrdtGossipManager with a GSet<String> CRDT.

import 'dart:async';
import 'dart:typed_data';

import 'package:dart_libp2p/core/host/host.dart' show Host;
import 'package:dart_libp2p/core/peer/peer_id.dart' show PeerId;
// Assuming you have set up your libp2p Host, PubSub, and optionally DHT
// import your_libp2p_setup_file.dart';

import 'package:merkledag/merkledag.dart' show GSet, CRDTPayload; // For GSet and CRDTPayload
import 'package:dart_libp2p_merkle_crdt/dart_libp2p_merkle_crdt.dart';
import 'package:logging/logging.dart';

// Example Payload Factory for GSet<String>
// P (Payload Type) is GSet<String>
// V (Value Type) is Set<String>
GSet<String> gsetPayloadFactory(Uint8List payloadBytes) {
  // This is a simplified example. GSet would need a proper fromBytes constructor.
  // For now, assuming it can be reconstructed or this factory is more complex.
  // If GSet.fromCanonicalBytes() existed: return GSet.fromCanonicalBytes(payloadBytes);
  // For this example, let's assume it's empty if bytes are unrecognized.
  // In a real scenario, GSet would need a proper deserialization method.
  _logger.warning('gsetPayloadFactory: Deserialization from Uint8List is a placeholder.');
  return GSet<String>(); // Placeholder: returns an empty GSet
}

// Placeholder for your libp2p Host, PubSub, and DHT instances
late Host myHost;
late PubSub myPubSub;
IpfsDHT? myDht; // Optional

final _logger = Logger('MerkleCrdtUsageExample');

void main() async {
  // --- Setup Phase (Illustrative - replace with your actual libp2p setup) ---
  // 1. Initialize Logger (optional)
  Logger.root.level = Level.INFO;
  Logger.root.onRecord.listen((record) {
    print('${record.level.name}: ${record.time}: ${record.loggerName}: ${record.message}');
    if (record.error != null) print('ERROR: ${record.error}, ${record.stackTrace}');
  });

  // 2. Initialize your libp2p Host, PubSub, and DHT (myHost, myPubSub, myDht)
  // This part is highly dependent on your application's libp2p setup.
  // For example, using `createLibp2pNode` from `test/real_net_stack.dart` in this package
  // or a similar utility in your main application.
  // myHost = await setupMyLibp2pHost();
  // myPubSub = PubSub(myHost, GossipSubRouter()); // Example
  // await myPubSub.start();
  // myDht = await setupMyDht(myHost); // Example
  // await myDht?.bootstrap();
  // For this example to run, these would need to be actual initialized instances.
  // This example will not run as-is without a concrete libp2p stack.
  print('Placeholder: Initialize myHost, myPubSub, and optionally myDht here.');
  // return; // Uncomment if you want to stop before manager setup without real instances.


  // --- MerkleCrdtGossipManager Setup ---

  // 3. Create a unique ID for this CRDT instance (e.g., a document ID)
  // This ID is used to scope GossipSub topics.
  const crdtInstanceId = 'my-shared-document-123';

  // 4. Create the BlockStore
  final blockStore = InMemoryBlockStore<GSet<String>, Set<String>>();

  // 5. Create the P2PStreamDagFetcher (DAGSyncer)
  // The protocol ID can be any unique string for your DAG sync protocol.
  final dagFetcher = P2PStreamDagFetcher<GSet<String>, Set<String>>(
    myHost,
    blockStore,
    '/my-app/merkle-crdt-dag-sync/1.0.0',
    gsetPayloadFactory, // Provide the factory for your payload type P
    // cidFunction was removed from P2PStreamDagFetcher constructor
    dht: myDht, // Pass the DHT instance if you have one
  );

  // 6. Create the GossipSubAnnouncer (Broadcaster)
  final announcer = GossipSubAnnouncer(
    myHost,
    myPubSub,
    crdtInstanceId, // Ensures announcements are scoped to this CRDT
  );

  // 7. Create the MerkleCrdtGossipManager
  // V = Set<String>, P = GSet<String>
  final manager = MerkleCrdtGossipManager<Set<String>, GSet<String>>(
    dagFetcher: dagFetcher,
    announcer: announcer,
  );

  // --- Using the Manager ---

  // 8. Listen to state changes
  manager.onStateChanged.listen((Set<String>? newState) {
    _logger.info('CRDT state changed (logical value): $newState');
    // Note: onStateChanged emits V?, which is Set<String>? in this case.
  });

  // 9. Apply a local update
  _logger.info('Applying first local update...');
  final update1 = GSet<String>();
  update1.add('apple');
  update1.add('banana');
  await manager.applyLocalUpdate(update1);
  // Current state (logical value V) will be {'apple', 'banana'}
  // This will also trigger a broadcast of the new head CID.

  // Simulate some time for gossip and potential remote updates
  await Future.delayed(Duration(seconds: 2));

  // 10. Apply another local update
  _logger.info('Applying second local update...');
  final update2 = GSet<String>();
  update2.add('cherry');
  // Note: GSet merge logic means 'apple' and 'banana' are preserved.
  await manager.applyLocalUpdate(update2);
  // Current state (logical value V) will be {'apple', 'banana', 'cherry'}

  // 11. Get the current state directly
  Set<String>? currentState = await manager.getState();
  _logger.info('Directly fetched current state (logical value): $currentState');

  // 12. Refresh state (useful if MerkleCRDT doesn't have its own state stream for remote changes)
  // This explicitly fetches the latest state from the underlying MerkleCRDT,
  // which would include any remote changes processed internally by MerkleCRDT.
  await manager.refreshState();
  _logger.info('State after explicit refresh (logical value): ${manager.currentState}');


  // --- Cleanup ---
  // Depending on your application, you might want to close the manager
  // when it's no longer needed. This will stop listening for remote heads.
  // The underlying Host, PubSub, DHT, etc., should be managed separately.
  // await manager.close();
  // await myPubSub.stop();
  // await myDht?.close();
  // await myHost.close();
}

Architecture Notes

• MerkleCRDT (from merkledag package): This is the core data structure. It manages the Merkle-DAG of CRDT payloads and their Merkle clock. It uses a DAGSyncer to fetch/store nodes and a Broadcaster to announce/receive heads.
• P2PStreamDagFetcher (DAGSyncer):
  • When MerkleCRDT needs a node (get(CID)), the fetcher first checks its local BlockStore.
  • If not found, and if a DHT is provided, it queries the DHT for peers (AddrInfo) providing that CID.
  • It then attempts to connect to these peers (or the peer from which a head was announced) and requests the node over a direct libp2p stream using a custom protocol.
  • When MerkleCRDT creates a new node (put(MerkleNode)), the fetcher stores it in the BlockStore and (if a DHT is provided) announces itself as a provider for that node’s CID on the DHT.
• GossipSubAnnouncer (Broadcaster):
  • When MerkleCRDT has a new head (e.g., after add(payload)), it calls broadcast(cidString) on the announcer. The announcer then publishes a small JSON message { "cid": "...", "senderPeerId": "..." } to a CRDT-instance-specific GossipSub topic.
  • The MerkleCRDT (via the announcer passed to its constructor) also calls subscribe() on the announcer. The MerkleCRDT itself listens to this stream of incoming CID strings and internally processes them (fetching data via the DAGSyncer and merging).
• MerkleCrdtGossipManager:
  • Initializes and holds the MerkleCRDT, P2PStreamDagFetcher, and GossipSubAnnouncer.
  • Provides a simplified API (applyLocalUpdate, getState, onStateChanged) to the application, dealing with the logical value type V.
  • Converts between the logical value V and the CRDT payload P where necessary (e.g., P.value gives V).

Important Considerations

• MerkleCRDT’s Internal Head Processing: This library assumes that the MerkleCRDT class from the merkledag package, when given a Broadcaster, will internally call broadcaster.subscribe() and process the incoming CIDs to trigger its merge logic (which in turn uses the DAGSyncer). If this is not the case, the MerkleCrdtGossipManager would need to be adjusted to explicitly handle CIDs from the announcer’s stream and call an appropriate method on MerkleCRDT to incorporate them.
• Payload Deserialization (payloadFactory): The P2PStreamDagFetcher requires a payloadFactory function (P Function(Uint8List payloadBytes)) to deserialize MerkleNode payloads received over the network. You must provide a correct factory for your specific CRDTPayload type P.
• Serialization/Deserialization of MerkleNode: The current implementation in P2PStreamDagFetcher uses a simplified JSON-based serialization for MerkleNode objects for transmission over P2P streams. For production, consider a more efficient and robust binary format (e.g., Protobuf).
• Error Handling & Resilience: The current error handling is basic. Production systems would require more sophisticated retry mechanisms, connection management, and error reporting.

ActivityPub-Inspired P2P Social Primitives (“activity-pubsub”)

This library also provides a suite of P2P social primitives inspired by ActivityPub, built on top of the core Merkle-CRDT synchronization layer. These primitives, developed under the “activity-pubsub” initiative, enable developers to easily integrate decentralized social features into their applications. They leverage DIDs for identity, IPLD for data structures, Merkle-CRDTs for state management, and libp2p (GossipSub and Bitswap) for communication.

Core Concepts

• Actors: Represent users or entities in the social graph, identified by Decentralized Identifiers (DIDs), typically did:key. Each actor has a Profile and manages their own data.
• Content Objects (IPLD): Social objects like posts (Note) are structured using IPLD, allowing them to be content-addressed (referenced by CID).
• Activities (IPLD): Actions performed by actors (e.g., creating a note, following another actor, liking content) are also modeled as IPLD objects. These activities are signed by the actor’s private key.
• CRDTs for State: Mutable collections and actor states (e.g., profiles, lists of posts in an outbox, following lists, liked items) are managed using Merkle-CRDTs. This ensures eventual consistency across peers without central servers.
• Replication & Exchange:
  • Updates to CRDT heads (signifying new state) are announced and replicated using the dart-libp2p-merkle-crdt mechanisms (i.e., GossipSub for head announcements and direct P2P stream-based DAG syncing for CRDT nodes).
  • IPLD content objects and activities are fetched by their CIDs using a Bitswap-compatible mechanism (facilitated by P2PStreamDagFetcher or a dedicated Bitswap client).

Key Features & Usage

The following features correspond to Phases 1-3 of the activity-pubsub.md development plan.

1. Actor Identity & Profile

• Actor Identity (lib/src/p2p/identity/actor.dart):
  • Actors generate and manage their cryptographic key pairs, from which DIDs (e.g., did:key) are derived.
• Actor Profile CRDT (ProfileCrdt - lib/src/p2p/crdt/profile_crdt.dart):
  • A Merkle-CRDT (e.g., LWW-Map or JSON-CRDT) storing profile information.
  • Required fields: id (DID), type (“Person”, “Service”, etc.), publicKeyJwk.
  • Recommended fields: name, preferredUsername, summary, icon (CID to image), image (CID to image).
  • Usage:
    • An actor creates and updates their ProfileCrdt.
    • The head of this CRDT is gossiped (e.g., on a topic like profiles/<actor_did>).
    • Clients discover and sync an actor’s ProfileCrdt to view their profile.

2. Content Publishing

• Content Object (Note - lib/src/p2p/ipld/note.dart):
  • An IPLD schema for simple content (e.g., {"@context": "...", "type": "Note", "content": "...", "published": "timestamp"}).
  • Actors create Note IPLD blocks, resulting in a CID.
• Create Activity (IPLD - schema defined in lib/src/p2p/ipld/activity.dart context):
  • An IPLD schema (e.g., {"@context": "...", "type": "Create", "actor": "actor_did", "object": "note_cid", "published": "timestamp"}).
  • Activities are signed by the actor’s private key.
  • Actors create Create activity IPLD blocks, resulting in a CID.
• outbox CRDT (OutboxCrdt - lib/src/p2p/crdt/outbox_crdt.dart):
  • An append-only log CRDT storing CIDs of an actor’s activities.
  • Usage:
    • To publish: An actor appends the CID of a new activity (e.g., a Create activity for a Note) to their OutboxCrdt.
    • The OutboxCrdt head is gossiped (e.g., on topic outboxes/<actor_did>).
    • Clients sync an actor’s OutboxCrdt to retrieve their activities, then fetch the actual IPLD objects (Activities, Notes) by CID via Bitswap.
    • Actors should make their own content and activities available via Bitswap (e.g., by “pinning” them in their local BlockStore).

3. Basic Social Interactions

• Follow Activity & following CRDT:
  • FollowActivity (lib/src/p2p/ipld/follow_activity.dart): IPLD schema (e.g., {"type": "Follow", "actor": "follower_did", "object": "followed_actor_did"}). Signed and added to the follower’s outbox.
  • FollowingCrdt (lib/src/p2p/crdt/following_crdt.dart): An OR-Set CRDT storing DIDs of actors being followed.
  • Usage: To follow someone, an actor adds a FollowActivity to their outbox and updates their FollowingCrdt. Clients sync this CRDT to know who the user follows.
• Like Activity & liked CRDT:
  • LikeActivity (lib/src/p2p/ipld/like_activity.dart): IPLD schema (e.g., {"type": "Like", "actor": "liker_did", "object": "liked_content_cid"}). Signed and added to the liker’s outbox.
  • LikedCrdt (lib/src/p2p/crdt/liked_crdt.dart): An OR-Set CRDT storing CIDs of liked content.
  • Usage: To like content, an actor adds a LikeActivity to their outbox and updates their LikedCrdt.

4. Enhanced Interactions & Content Management

• Replies (CreateActivity with inReplyTo):
  • A reply is a Note object (IPLD). The Create activity for the reply includes an inReplyTo: "original_content_cid" property.
  • The reply activity is signed and added to the replier’s outbox.
  • Usage: Clients reconstruct conversation threads by linking inReplyTo properties. Reply discovery can be client-side (e.g., scanning outboxes of followed users).
• Announce (Boost/Share) Activity (AnnounceActivity - lib/src/p2p/ipld/announce_activity.dart):
  • IPLD schema (e.g., {"type": "Announce", "actor": "announcer_did", "object": "announced_content_cid"}).
  • Signed and added to the announcer’s outbox.
  • Usage: Client displays announced content, attributing the original author and the announcer.
• Delete Activity (Tombstoning) (DeleteActivity - lib/src/p2p/ipld/delete_activity.dart):
  • IPLD schema (e.g., {"type": "Delete", "actor": "deleter_did", "object": "content_to_delete_cid"}).
  • Signed and added to the actor’s outbox.
  • Usage:
    • Clients should hide content associated with a Delete activity.
    • The actor’s client should unpin (stop serving via Bitswap) their “deleted” content. Other clients may optionally unpin it from their local cache.
• Actor Mentions:
  • Implemented by convention using Mention tags within Note content (e.g., {"type": "Mention", "href": "mentioned_actor_did"}).
  • Usage: Initial mention discovery is typically client-side, scanning content from followed users.

Integrating “activity-pubsub” Features into an Application

1. Core Setup:
   • Initialize your libp2p Host, PubSub (GossipSub), and optionally IpfsDHT as described in the “Getting Started” section of this README.
2. Manage CRDT Instances:
   • For each actor and each type of social CRDT (Profile, Outbox, Following, Liked), you will typically instantiate a MerkleCrdtGossipManager.
   • The crdtInstanceId for each manager should be unique and discoverable, often incorporating the actor’s DID and the CRDT type (e.g., profile-<actor_did>, outbox-<actor_did>).
   • Provide the appropriate payloadFactory for each CRDT type (e.g., ProfileCrdt.fromBytes, OutboxCrdt.fromBytes).
3. Actor Management:
   • Implement logic for creating/loading actor identities (key pairs, DIDs).
4. Implement Social Actions:
   • Creating Content:
     1. Construct the Note IPLD object. Store it locally (e.g., via P2PStreamDagFetcher’s blockStore.put()) to get its CID and make it available via Bitswap.
     2. Construct the CreateActivity IPLD object, embedding the Note’s CID. Sign it. Store it and get its CID.
     3. Update the actor’s OutboxCrdt by adding the CreateActivity’s CID, then call manager.applyLocalUpdate() on the outbox manager.
   • Other Activities (Follow, Like, Announce, Delete): Follow a similar pattern: create the activity IPLD object, sign it, store it, get its CID, and add this CID to the actor’s OutboxCrdt. For Follow and Like, also update the respective FollowingCrdt or LikedCrdt.
5. Data Fetching & Synchronization:
   • To view a profile, sync the ProfileCrdt for the target actor’s DID.
   • To build a feed or view posts, sync the OutboxCrdt of relevant actors. For each activity CID in an outbox, fetch the activity IPLD object, and if it’s a Create or Announce of a Note, fetch the Note IPLD object. This is done via the P2PStreamDagFetcher (which uses Bitswap-like mechanisms).
6. Client-Side Logic:
   • Develop the UI/UX for presenting social information.
   • Implement client-side aggregation (e.g., a chronological feed from multiple followed outboxes).
   • Handle threading of replies by processing inReplyTo fields.
   • Respect Delete activities by hiding or removing content.

The “activity-pubsub” primitives provide a flexible foundation. Application developers are responsible for combining these building blocks to create rich user experiences. For more details on the specific CRDTs and IPLD models, refer to the source files in lib/src/p2p/.

Future development (Phase 4 and beyond) aims to explore advanced discovery, storage strategies, moderation primitives, and more.

Additional Information

• merkledag package: [Link to your merkledag package or its documentation]
• dart_libp2p suite: [Link to dart_libp2p and related packages]

To contribute, please file an issue or submit a pull request.