save prog

Author: tabcat

src/replicator/interface.ts

@@ -1,13 +1,17 @@
-import type { GossipHelia, ComponentProtocol } from '@/interface.js'
+import type { ComponentProtocol } from '@/interface.js'
 import type { Replica } from '@/replica/index.js'
 import { HLDB_PREFIX } from '@/utils/constants.js'
 import type { Playable } from '@/utils/playable.js'
 import type { Datastore } from 'interface-datastore'
 import type { Blockstore } from 'interface-blockstore'
 import type { Ed25519PeerId } from '@libp2p/interface/peer-id'
+import type { PubSub } from '@libp2p/interface/pubsub'
+import type { DualKadDHT } from '@libp2p/kad-dht'
 
 export interface Config {
-  ipfs: GossipHelia
+  peerId: Ed25519PeerId
+  pubsub: PubSub
+  dht?: DualKadDHT
   replica: Replica
   datastore: Datastore
   blockstore: Blockstore

src/replicator/live/index.ts

@@ -1,9 +1,8 @@
 import all from 'it-all'
 import { start, stop } from '@libp2p/interface/startable'
 import { base32 } from 'multiformats/bases/base32'
-import type { GossipHelia } from '@/interface'
 import type { CID } from 'multiformats/cid'
-import type { SignedMessage, PublishResult } from '@libp2p/interface/pubsub'
+import type { SignedMessage, PublishResult, PubSub } from '@libp2p/interface/pubsub'
 
 import { dagLinks, loadEntry, traverser } from '@/replica/traversal.js'
 import { cidstring, parsedcid } from '@/utils/index.js'
@@ -19,11 +18,13 @@ import type { Config, ReplicatorModule } from '../interface.js'
 import * as Advert from './message.js'
 import protocol from './protocol.js'
 import type { Blockstore } from 'interface-blockstore'
+import type { Ed25519PeerId } from '@libp2p/interface/dist/src/peer-id/index.js'
 
 const getSharedChannelTopic = (manifest: Manifest): string => `${protocol}${cidstring(manifest.address.cid)}`
 
 export class LiveReplicator extends Playable {
-  readonly ipfs: GossipHelia
+  readonly localPeerId: Ed25519PeerId
+  readonly pubsub: PubSub
   readonly blockstore: Blockstore
   readonly manifest: Manifest
   readonly replica: Replica
@@ -40,7 +41,8 @@ export class LiveReplicator extends Playable {
   readonly _onHeadsMessage: typeof onHeadsMessage
 
   constructor ({
-    ipfs,
+    peerId,
+    pubsub,
     replica,
     blockstore
   }: Config) {
@@ -72,7 +74,8 @@ export class LiveReplicator extends Playable {
     }
     super({ starting, stopping })
 
-    this.ipfs = ipfs
+    this.localPeerId = peerId
+    this.pubsub = pubsub
     this.blockstore = blockstore
     this.replica = replica
     this.manifest = replica.manifest
@@ -85,7 +88,7 @@ export class LiveReplicator extends Playable {
     this.#onReplicaHeadsUpdate = onReplicaHeadsUpdate.bind(this)
     this._onHeadsMessage = onHeadsMessage.bind(this)
 
-    this.shared = new Monitor(this.ipfs.libp2p, getSharedChannelTopic(this.manifest))
+    this.shared = new Monitor(this.pubsub, getSharedChannelTopic(this.manifest))
     this.directs = new Map()
   }
 
@@ -142,7 +145,7 @@ function onPeerJoin (
   evt: CustomEvent<PeerStatusChangeData>
 ): void {
   const { peerId: remotePeerId } = evt.detail
-  const direct = new Direct(this.ipfs.libp2p, remotePeerId)
+  const direct = new Direct(this.pubsub, this.localPeerId, remotePeerId)
   direct.addEventListener(
     'peered',
     () => {

test/replicator/replicator.spec.ts test/replicator/live.spec.tstest/replicator/replicator.spec.ts renamed to test/replicator/live.spec.ts

@@ -3,7 +3,7 @@ import { start, stop } from '@libp2p/interface/startable'
 import { Key } from 'interface-datastore'
 import { MemoryDatastore, NamespaceDatastore } from 'datastore-core'
 
-import { LiveReplicator as Replicator } from '@/replicator/live/index.js'
+import { LiveReplicator } from '@/replicator/live/index.js'
 import { Replica } from '@/replica/index.js'
 import { StaticAccess as Access } from '@/access/static/index.js'
 import staticAccessProtocol from '@/access/static/protocol.js'
@@ -14,39 +14,40 @@ import { getTestManifest } from '../test-utils/manifest.js'
 import { getTestIdentities, getTestIdentity } from '../test-utils/identities.js'
 import { basalEntry } from '@/entry/basal/index.js'
 import { basalIdentity } from '@/identity/basal/index.js'
-import type { Multiaddr } from '@multiformats/multiaddr'
-import { getDhtService, getIdentifyService, getPubsubService, type UsedServices } from '../test-utils/libp2p/services.js'
-import type { Helia } from '@helia/interface'
-import { createLibp2p, Libp2pOptions, type Libp2p } from 'libp2p'
-import { getLibp2pDefaults } from '../test-utils/libp2p/defaults.js'
-import { getPeerDiscovery } from '../test-utils/libp2p/peerDiscovery.js'
-import { createHelia } from 'helia'
-import { waitForMultiaddrs } from '../test-utils/network.js'
+import { getTestKeyChain } from 'test/test-utils/keychain.js'
+import { MemoryBlockstore } from 'blockstore-core'
+import type { Ed25519PeerId } from '@libp2p/interface/dist/src/peer-id/index.js'
+import { createEd25519PeerId } from '@libp2p/peer-id-factory'
+import { getTestPubSubNetwork } from 'test/test-mocks/pubsub.js'
+import type { PubSub } from '@libp2p/interface/pubsub'
 
-const testName = 'live-replicator'
-
-type TestServices = UsedServices<'identify' | 'pubsub' | 'dht'>
+const testName = 'replicator/live'
 
 describe(testName, () => {
   let
-    helia1: Helia<Libp2p<TestServices>>,
-    helia2: Helia<Libp2p<TestServices>>,
-    libp2p1: Libp2p<TestServices>,
-    libp2p2: Libp2p<TestServices>,
-    addr2: Multiaddr[],
+    id1: Ed25519PeerId,
+    id2: Ed25519PeerId,
+    pubsub1: PubSub,
+    pubsub2: PubSub,
     replica1: Replica,
     replica2: Replica,
-    replicator1: Replicator,
-    replicator2: Replicator,
+    replicator1: LiveReplicator,
+    replicator2: LiveReplicator,
     testPaths1: TestPaths,
     testPaths2: TestPaths,
     access: Access,
     datastore: MemoryDatastore,
     datastore1: NamespaceDatastore,
-    datastore2: NamespaceDatastore
+    datastore2: NamespaceDatastore,
+    blockstore: MemoryBlockstore
 
   before(async () => {
-    // debug.enable('libp2p:*')
+    id1 = await createEd25519PeerId()
+    id2 = await createEd25519PeerId()
+
+    const { createPubSubPeer } = getTestPubSubNetwork()
+    pubsub1 = createPubSubPeer(id1)
+    pubsub2 = createPubSubPeer(id2)
 
     testPaths1 = getTestPaths(tempPath, testName + '/1')
     testPaths2 = getTestPaths(tempPath, testName + '/2')
@@ -55,43 +56,24 @@ describe(testName, () => {
     datastore1 = new NamespaceDatastore(datastore, new Key(testPaths1.replica))
     datastore2 = new NamespaceDatastore(datastore, new Key(testPaths2.replica))
 
-    const createLibp2pOptions = async (): Promise<Libp2pOptions<TestServices>> => ({
-      ...(await getLibp2pDefaults()),
-      peerDiscovery: await getPeerDiscovery(),
-      services: {
-        identify: getIdentifyService(),
-        pubsub: getPubsubService(),
-        dht: getDhtService(true)
-      }
-    })
-
-    libp2p1 = await createLibp2p(await createLibp2pOptions())
-    libp2p2 = await createLibp2p(await createLibp2pOptions())
-
-    await Promise.all([
-      waitForMultiaddrs(libp2p1),
-      waitForMultiaddrs(libp2p2)
-    ])
-
-    helia1 = await createHelia({ libp2p: libp2p1 })
-    helia2 = await createHelia({ libp2p: libp2p2 })
-
-    addr2 = libp2p2.getMultiaddrs()
-
     const identities1 = await getTestIdentities(testPaths1)
     const identities2 = await getTestIdentities(testPaths2)
+    const keychain1 = getTestKeyChain()
+    const keychain2 = getTestKeyChain()
 
     const identity1 = await getTestIdentity(
       identities1,
-      libp2p1.keychain,
+      keychain1,
       testName
     )
     const identity2 = await getTestIdentity(
       identities2,
-      libp2p2.keychain,
+      keychain2,
       testName
     )
 
+    blockstore = new MemoryBlockstore()
+
     const write = [identity1.id, identity2.id]
     const accessConfig = {
       access: { protocol: staticAccessProtocol, config: { write } }
@@ -104,7 +86,7 @@ describe(testName, () => {
     replica1 = new Replica({
       manifest,
       datastore: datastore1,
-      blockstore: helia1.blockstore,
+      blockstore,
       access,
       identity: identity1,
       components: {
@@ -115,7 +97,7 @@ describe(testName, () => {
     replica2 = new Replica({
       manifest,
       datastore: datastore2,
-      blockstore: helia2.blockstore,
+      blockstore,
       access,
       identity: identity2,
       components: {
@@ -125,35 +107,31 @@ describe(testName, () => {
     })
     await start(replica1, replica2)
 
-    replicator1 = new Replicator({
-      ipfs: helia1,
+    replicator1 = new LiveReplicator({
+      peerId: id1,
+      pubsub: pubsub1,
       replica: replica1,
       datastore: datastore1,
-      blockstore: helia1.blockstore
+      blockstore
     })
-    replicator2 = new Replicator({
-      ipfs: helia2,
+    replicator2 = new LiveReplicator({
+      peerId: id2,
+      pubsub: pubsub2,
       replica: replica2,
       datastore: datastore2,
-      blockstore: helia2.blockstore
+      blockstore
     })
   })
 
   after(async () => {
     await stop(access)
     await stop(replicator1, replicator2)
     await stop(replica1, replica2)
-    await stop(helia1)
-    await stop(helia2)
   })
 
   describe('instance', () => {
     before(async () => {
       await start(replicator1, replicator2)
-      await Promise.all([
-        libp2p1.dial(addr2),
-        new Promise(resolve => { libp2p2.addEventListener('peer:connect', resolve, { once: true }) })
-      ])
     })
 
     it('exposes instance properties', () => {

test/replicator/zzzync.spec.ts

@@ -30,7 +30,7 @@ import { createHelia } from 'helia'
 import { getPeerDiscovery } from '../test-utils/libp2p/peerDiscovery.js'
 import { waitForMultiaddrs } from '../test-utils/network.js'
 
-const testName = 'zzzync-replicator'
+const testName = 'replicator/zzzync'
 const token = process.env.W3_TOKEN as string
 
 const noToken = token == null

test/test-mocks/dht.ts

@@ -0,0 +1,122 @@
+import { EventTypes, type KadDHT, type QueryEvent, type ProviderEvent, type QueryOptions } from '@libp2p/kad-dht'
+import type { Ed25519PeerId, PeerId } from '@libp2p/interface/peer-id'
+import type { CID } from 'multiformats'
+import { base32 } from 'multiformats/dist/types/src/bases/base32'
+import { peerIdFromString } from '@libp2p/peer-id'
+
+type Mode = 'client' | 'server'
+
+const cid2key = (cid: CID): Uint8Array => cid.multihash.bytes
+const key2str = (key: Uint8Array): string => base32.encode(key)
+const cid2str = (cid: CID): string => key2str(cid2key(cid))
+// const str2key = (str: string): Uint8Array => base32.decode(str)
+
+class TestKadDht implements KadDHT {
+  mode: 'client' | 'server'
+
+  constructor (
+    readonly peerId: Ed25519PeerId,
+    readonly peers: Set<string>,
+    readonly providers: Map<string, Set<string>>,
+    mode: Mode
+  ) {
+    this.mode = mode
+  }
+
+  /**
+   * Get a value from the DHT, the final ValueEvent will be the best value
+   */
+  async * get (key: Uint8Array, options?: QueryOptions): AsyncIterable<QueryEvent> {}
+
+  /**
+   * Find providers of a given CID
+   */
+  async * findProviders (key: CID, options?: QueryOptions): AsyncIterable<QueryEvent> {
+    const providers = this.providers.get(cid2str(key))
+
+    if (providers == null) {
+      return
+    }
+
+    const providerEvent: ProviderEvent = {
+      from: this.peerId,
+      type: EventTypes.PROVIDER,
+      name: 'PROVIDER',
+      providers: [...providers].map(peerId => ({ id: peerIdFromString(peerId), multiaddrs: [], protocols: [] }))
+    }
+
+    yield providerEvent
+  }
+
+  /**
+   * Find a peer on the DHT
+   */
+  async * findPeer (id: PeerId, options?: QueryOptions): AsyncIterable<QueryEvent> {}
+
+  /**
+   * Find the closest peers to the passed key
+   */
+  async * getClosestPeers (key: Uint8Array, options?: QueryOptions): AsyncIterable<QueryEvent> {}
+
+  /**
+   * Store provider records for the passed CID on the DHT pointing to us
+   */
+  async * provide (key: CID, options?: QueryOptions): AsyncIterable<QueryEvent> {
+    const providers = this.providers.get(cid2str(key)) ?? new Set()
+
+    const providerEvent: ProviderEvent = {
+      from: this.peerId,
+      type: EventTypes.PROVIDER,
+      name: 'PROVIDER',
+      providers: [...providers].map(provider => ({
+        id: peerIdFromString(provider),
+        multiaddrs: [],
+        protocols: []
+      }))
+    }
+
+    providers.add(this.peerId.toString())
+    this.providers.set(cid2str(key), providers)
+
+    yield providerEvent
+  }
+
+  /**
+   * Store the passed value under the passed key on the DHT
+   */
+  async * put (key: Uint8Array, value: Uint8Array, options?: QueryOptions): AsyncIterable<QueryEvent> {}
+
+  /**
+   * Returns the mode this node is in
+   */
+  async getMode (): Promise<Mode> {
+    return this.mode
+  }
+
+  /**
+   * If 'server' this node will respond to DHT queries, if 'client' this node will not
+   */
+  async setMode (mode: Mode): Promise<void> {
+    this.mode = mode
+  }
+
+  /**
+   * Force a routing table refresh
+   */
+  async refreshRoutingTable (): Promise<void> {}
+}
+
+interface TestDhtNetwork {
+  createDhtPeer: (peerId: Ed25519PeerId, mode: Mode) => TestKadDht
+}
+
+export const getTestDhtNetwork = (): TestDhtNetwork => {
+  const peers: Set<string> = new Set()
+  const providers: Map<string, Set<string>> = new Map()
+
+  return {
+    createDhtPeer (peerId: Ed25519PeerId, mode: Mode) {
+      return new TestKadDht(peerId, peers, providers, mode)
+    }
+  }
+}