[WIP] refactor: daemon modules

.gitignore

@@ -1,5 +1,6 @@
 /data
 /config
+/packages/*/test-out
 
 build
 dist

packages/daemon/src/common/commands/addresses.ts

@@ -0,0 +1,10 @@
+import { Addresses } from '@organicdesign/db-rpc-interfaces'
+import type { ModuleMethod } from '@/interface.js'
+
+const command: ModuleMethod = ({ net, libp2p }) => {
+  net.rpc.addMethod(Addresses.name, async (): Promise<Addresses.Return> => {
+    return libp2p.getMultiaddrs().map(a => a.toString())
+  })
+}
+
+export default command

packages/daemon/src/modules/network/commands/connect.ts → packages/daemon/src/common/commands/connect.ts

@@ -1,14 +1,13 @@
 import { multiaddr } from '@multiformats/multiaddr'
 import { Connect } from '@organicdesign/db-rpc-interfaces'
-import type { Provides, Requires } from '../index.js'
 import type { ModuleMethod } from '@/interface.js'
 
-const command: ModuleMethod<Provides, Requires> = (context, { rpc }) => {
-  rpc.addMethod(Connect.name, async (raw: unknown): Promise<Connect.Return> => {
+const command: ModuleMethod = ({ net, libp2p }) => {
+  net.rpc.addMethod(Connect.name, async (raw: unknown): Promise<Connect.Return> => {
     const params = Connect.Params.parse(raw)
     const address = multiaddr(params.address)
 
-    await context.libp2p.dial(address)
+    await libp2p.dial(address)
 
     return null
   })

packages/daemon/src/common/commands/connections.ts

@@ -0,0 +1,10 @@
+import { Connections } from '@organicdesign/db-rpc-interfaces'
+import type { ModuleMethod } from '@/interface.js'
+
+const command: ModuleMethod = ({ net, libp2p }) => {
+  net.rpc.addMethod(Connections.name, async (): Promise<Connections.Return> => {
+    return libp2p.getConnections().map(c => c.remoteAddr.toString())
+  })
+}
+
+export default command

packages/daemon/src/modules/network/commands/count-peers.ts → packages/daemon/src/common/commands/count-peers.ts

@@ -1,14 +1,13 @@
 import { CountPeers } from '@organicdesign/db-rpc-interfaces'
 import { CID } from 'multiformats/cid'
-import type { Provides, Requires } from '../index.js'
 import type { ModuleMethod } from '@/interface.js'
 
-const command: ModuleMethod<Provides, Requires> = (context, { rpc }) => {
-  rpc.addMethod(CountPeers.name, async (raw: unknown): Promise<CountPeers.Return> => {
+const command: ModuleMethod = ({ net, libp2p }) => {
+  net.rpc.addMethod(CountPeers.name, async (raw: unknown): Promise<CountPeers.Return> => {
     const countPeers = async (cid: CID, options?: { timeout: number }): Promise<number> => {
       let count = 0
 
-      const itr = context.libp2p.contentRouting.findProviders(cid, {
+      const itr = libp2p.contentRouting.findProviders(cid, {
         signal: AbortSignal.timeout(options?.timeout ?? 3000)
       })
 

packages/daemon/src/modules/groups/commands/create-group.ts → packages/daemon/src/common/commands/create-group.ts

@@ -1,23 +1,22 @@
 import { CreateGroup } from '@organicdesign/db-rpc-interfaces'
 import { fromString as uint8ArrayFromString } from 'uint8arrays'
-import type { Provides, Requires } from '../index.js'
 import type { ModuleMethod } from '@/interface.js'
 
-const command: ModuleMethod<Provides, Requires> = (context, { rpc }) => {
-  rpc.addMethod(CreateGroup.name, async (raw: unknown): Promise<CreateGroup.Return> => {
+const command: ModuleMethod = ({ net, welo, groups }) => {
+  net.rpc.addMethod(CreateGroup.name, async (raw: unknown): Promise<CreateGroup.Return> => {
     const params = CreateGroup.Params.parse(raw)
     const peerValues = params.peers.map(p => uint8ArrayFromString(p, 'base58btc'))
 
-    const manifest = await context.welo.determine({
+    const manifest = await welo.determine({
       name: params.name,
       meta: { type: 'group' },
       access: {
         protocol: '/hldb/access/static',
-        config: { write: [context.welo.identity.id, ...peerValues] }
+        config: { write: [welo.identity.id, ...peerValues] }
       }
     })
 
-    await context.groups.add(manifest)
+    await groups.add(manifest)
 
     return manifest.address.cid.toString()
   })

packages/daemon/src/modules/downloader/commands/get-speeds.ts → packages/daemon/src/common/commands/get-speeds.ts

@@ -1,15 +1,14 @@
 import { GetSpeeds } from '@organicdesign/db-rpc-interfaces'
 import { CID } from 'multiformats/cid'
-import type { Provides, Requires } from '../index.js'
 import type { ModuleMethod } from '@/interface.js'
 
-const command: ModuleMethod<Provides, Requires> = (context, { rpc }) => {
-  rpc.addMethod(GetSpeeds.name, async (raw: unknown): Promise<GetSpeeds.Return> => {
+const command: ModuleMethod = ({ net, pinManager }) => {
+  net.rpc.addMethod(GetSpeeds.name, async (raw: unknown): Promise<GetSpeeds.Return> => {
     const params = GetSpeeds.Params.parse(raw)
 
     return Promise.all(params.cids.map(async str => {
       const cid = CID.parse(str)
-      const speed = await context.pinManager.getSpeed(cid, params.range)
+      const speed = await pinManager.getSpeed(cid, params.range)
 
       return {
         cid: str,

packages/daemon/src/modules/downloader/commands/get-status.ts → packages/daemon/src/common/commands/get-status.ts

@@ -1,19 +1,18 @@
 import { GetStatus } from '@organicdesign/db-rpc-interfaces'
 import { CID } from 'multiformats/cid'
-import type { Provides, Requires } from '../index.js'
 import type { ModuleMethod } from '@/interface.js'
 
-const command: ModuleMethod<Provides, Requires> = (context, { rpc }) => {
-  rpc.addMethod(GetStatus.name, async (raw: unknown): Promise<GetStatus.Return> => {
+const command: ModuleMethod = ({ pinManager, net }) => {
+  net.rpc.addMethod(GetStatus.name, async (raw: unknown): Promise<GetStatus.Return> => {
     const params = GetStatus.Params.parse(raw)
 
     return Promise.all(params.cids.map(async str => {
       const cid = CID.parse(str)
 
       const [state, blocks, size] = await Promise.all([
-        context.pinManager.getState(cid),
-        context.pinManager.getBlockCount(cid),
-        context.pinManager.getSize(cid)
+        pinManager.getState(cid),
+        pinManager.getBlockCount(cid),
+        pinManager.getSize(cid)
       ])
 
       return {

packages/daemon/src/common/commands/id.ts

@@ -0,0 +1,11 @@
+import { ID } from '@organicdesign/db-rpc-interfaces'
+import { toString as uint8ArrayToString } from 'uint8arrays'
+import type { ModuleMethod } from '@/interface.js'
+
+const command: ModuleMethod = ({ net, welo }) => {
+  net.rpc.addMethod(ID.name, async (): Promise<ID.Return> => {
+    return uint8ArrayToString(welo.identity.id, 'base58btc')
+  })
+}
+
+export default command

packages/daemon/src/modules/groups/commands/join-group.ts → packages/daemon/src/common/commands/join-group.ts

@@ -1,15 +1,14 @@
 import { JoinGroup } from '@organicdesign/db-rpc-interfaces'
 import { Address } from 'welo'
-import type { Provides, Requires } from '../index.js'
 import type { ModuleMethod } from '@/interface.js'
 
-const command: ModuleMethod<Provides, Requires> = (context, { rpc }) => {
-  rpc.addMethod(JoinGroup.name, async (raw: unknown): Promise<JoinGroup.Return> => {
+const command: ModuleMethod = ({ net, welo, groups }) => {
+  net.rpc.addMethod(JoinGroup.name, async (raw: unknown): Promise<JoinGroup.Return> => {
     const params = JoinGroup.Params.parse(raw)
-    const manifest = await context.welo.fetch(Address.fromString(`/hldb/${params.group}`))
+    const manifest = await welo.fetch(Address.fromString(`/hldb/${params.group}`))
 
     try {
-      await context.groups.add(manifest)
+      await groups.add(manifest)
     } catch (error) {
       if ((error as Error).message.includes('is already open')) {
         throw new Error('group has already been joined')

packages/daemon/src/modules/groups/commands/list-groups.ts → packages/daemon/src/common/commands/list-groups.ts

@@ -1,12 +1,11 @@
 import { ListGroups } from '@organicdesign/db-rpc-interfaces'
-import type { Provides, Requires } from '../index.js'
 import type { ModuleMethod } from '@/interface.js'
 
-const command: ModuleMethod<Provides, Requires> = (context, { rpc }) => {
-  rpc.addMethod(ListGroups.name, async (): Promise<ListGroups.Return> => {
+const command: ModuleMethod = ({ net, groups }) => {
+  net.rpc.addMethod(ListGroups.name, async (): Promise<ListGroups.Return> => {
     const promises: Array<{ group: string, name: string }> = []
 
-    for (const { key: cid, value: database } of context.groups.all()) {
+    for (const { key: cid, value: database } of groups.all()) {
       promises.push({ group: cid, name: database.manifest.name })
     }
 

packages/daemon/src/modules/downloader/commands/set-priority.ts → packages/daemon/src/common/commands/set-priority.ts

@@ -1,21 +1,20 @@
 import Path from 'path'
 import { SetPriority } from '@organicdesign/db-rpc-interfaces'
-import type { Provides, Requires } from '../index.js'
 import type { ModuleMethod } from '@/interface.js'
 
-const command: ModuleMethod<Provides, Requires> = (context, { rpc }) => {
-  rpc.addMethod(SetPriority.name, async (raw: unknown): Promise<SetPriority.Return> => {
+const command: ModuleMethod = ({ net, pinManager }) => {
+  net.rpc.addMethod(SetPriority.name, async (raw: unknown): Promise<SetPriority.Return> => {
     const params = SetPriority.Params.parse(raw)
     const key = Path.join('/', params.group, params.path)
-    const pinInfo = await context.pinManager.get(key)
+    const pinInfo = await pinManager.get(key)
 
     if (pinInfo == null) {
       throw new Error('no such pin')
     }
 
     pinInfo.priority = params.priority
 
-    await context.pinManager.put(key, pinInfo)
+    await pinManager.put(key, pinInfo)
 
     return null
   })

packages/daemon/src/common/commands/sneakernet-receive.ts

@@ -0,0 +1,14 @@
+import { SneakernetReveive } from '@organicdesign/db-rpc-interfaces'
+import type { ModuleMethod } from '@/interface.js'
+
+const command: ModuleMethod = ({ sneakernet, net }) => {
+  net.rpc.addMethod(SneakernetReveive.name, async (raw: unknown): Promise<SneakernetReveive.Return> => {
+    const params = SneakernetReveive.Params.parse(raw)
+
+    await sneakernet.import(params.path)
+
+    return null
+  })
+}
+
+export default command

packages/daemon/src/common/commands/sneakernet-send.ts

@@ -0,0 +1,14 @@
+import { SneakernetSend } from '@organicdesign/db-rpc-interfaces'
+import type { ModuleMethod } from '@/interface.js'
+
+const command: ModuleMethod = ({ sneakernet, net }) => {
+  net.rpc.addMethod(SneakernetSend.name, async (raw: unknown): Promise<SneakernetSend.Return> => {
+    const params = SneakernetSend.Params.parse(raw)
+
+    await sneakernet.export(params.path, params.peers)
+
+    return null
+  })
+}
+
+export default command

packages/daemon/src/modules/groups/commands/sync.ts → packages/daemon/src/common/commands/sync.ts

@@ -2,7 +2,6 @@ import { Sync } from '@organicdesign/db-rpc-interfaces'
 import { HeadsExchange } from 'welo/utils/heads-exchange'
 import { cidstring } from 'welo/utils/index'
 import { getHeads, addHeads } from 'welo/utils/replicator'
-import type { Provides, Requires } from '../index.js'
 import type { ModuleMethod } from '@/interface.js'
 import type { Peer, Libp2p } from '@libp2p/interface'
 import type { Database } from 'welo'
@@ -59,18 +58,18 @@ const sync = async (libp2p: Libp2p, peer: Peer, database: Database, options: Par
   await stream.close()
 }
 
-const command: ModuleMethod<Provides, Requires> = (context, { rpc, network }) => {
-  rpc.addMethod(Sync.name, async (): Promise<Sync.Return> => {
-    const peers = network.libp2p.getPeers()
-    const databases = context.welo.opened.values()
+const command: ModuleMethod = ({ libp2p, welo, net }) => {
+  net.rpc.addMethod(Sync.name, async (): Promise<Sync.Return> => {
+    const peers = libp2p.getPeers()
+    const databases = welo.opened.values()
 
     const promises: Array<Promise<void>> = []
 
     for (const peerId of peers) {
-      const peer = await network.libp2p.peerStore.get(peerId)
+      const peer = await libp2p.peerStore.get(peerId)
 
       for (const database of databases) {
-        promises.push(sync(network.libp2p, peer, database))
+        promises.push(sync(libp2p, peer, database))
       }
     }
 

packages/daemon/src/common/downloader/index.ts

@@ -0,0 +1,122 @@
+import parallel from 'it-parallel'
+import { pipe } from 'it-pipe'
+import { collect } from 'streaming-iterables'
+import { Event, EventTarget } from 'ts-event-target'
+import { linearWeightTranslation } from './utils.js'
+import type { PinManager } from '../pin-manager/index.js'
+import type { Startable } from '@libp2p/interface'
+import type { CID } from 'multiformats/cid'
+
+export class DownloadErrorEvent extends Event<'download:error'> {
+  readonly error: Error
+
+  constructor (error: Error) {
+    super('download:error')
+
+    this.error = error
+  }
+}
+
+export class Downloader implements Startable {
+  private readonly slots: number
+  private readonly pinManager: PinManager
+  private controller: AbortController = new AbortController()
+  private loopPromise: Promise<void> | null = null
+  readonly events = new EventTarget<[DownloadErrorEvent]>()
+
+  constructor (pinManager: PinManager, slots: number) {
+    this.slots = slots
+    this.pinManager = pinManager
+  }
+
+  async start (): Promise<void> {
+    await this.loopPromise
+    this.controller = new AbortController()
+    this.loopPromise = this.loop()
+  }
+
+  async stop (): Promise<void> {
+    this.controller.abort()
+    await this.loopPromise
+  }
+
+  private async loop (): Promise<void> {
+    for (;;) {
+      if (this.isAborted) {
+        return
+      }
+
+      await pipe(
+        () => this.delay(),
+        i => this.getPins(i),
+        i => this.batchDownload(i),
+        i => parallel(i, { concurrency: this.slots, ordered: false }),
+        i => this.catcher(i),
+        async i => collect(i)
+      )
+
+      await new Promise(resolve => setTimeout(resolve, 100))
+    }
+  }
+
+  private async * batchDownload (itr: AsyncIterable<[CID, number]>): AsyncGenerator<() => Promise<{ cid: CID, block: Uint8Array }>, void, undefined> {
+    for await (const [cid, priority] of itr) {
+      if (this.isAborted) {
+        return
+      }
+
+      const weight = Math.floor(linearWeightTranslation(priority / 100) * this.slots) + 1
+      const downloaders = await this.pinManager.download(cid, { limit: weight })
+
+      yield * downloaders
+    }
+  }
+
+  private async * catcher <T = unknown>(itr: AsyncIterable<T>): AsyncIterable<T> {
+    try {
+      yield * itr
+    } catch (e) {
+      const error = e instanceof Error ? e : new Error(JSON.stringify(e))
+
+      this.events.dispatchEvent(new DownloadErrorEvent(error))
+
+      yield * this.catcher(itr)
+    }
+  }
+
+  private async * delay (): AsyncGenerator<undefined> {
+    for (;;) {
+      if (this.isAborted) {
+        return
+      }
+
+      await new Promise(resolve => setTimeout(resolve, 100))
+      yield
+    }
+  }
+
+  async * getPins (loop: AsyncIterable<void>): AsyncGenerator<[CID, number]> {
+    // eslint-disable-next-line @typescript-eslint/no-unused-vars
+    for await (const _ of loop) {
+      for await (const { value } of this.pinManager.getActive()) {
+        if (this.isAborted) {
+          return
+        }
+
+        yield [value.cid, value.priority]
+      }
+    }
+  }
+
+  private get isAborted (): boolean {
+    return this.controller.signal.aborted
+  }
+}
+
+export const createDownloader = async (pinManage: PinManager, slots: number): Promise<Downloader> => {
+  const downloader = new Downloader(pinManage, slots)
+
+  await downloader.start()
+
+  return downloader
+}