feat(store): new implementation

Author: 0x77dev

.gitignore

@@ -37,6 +37,7 @@ npm-debug.log
 yarn-error.log
 testem.log
 /typings
+.eslintcache
 
 # System Files
 .DS_Store

.husky/.gitignore

@@ -0,0 +1 @@
+_

.husky/pre-commit

@@ -0,0 +1,4 @@
+#!/bin/sh
+. "$(dirname "$0")/_/husky.sh"
+
+npx lint-staged

package.json

@@ -39,11 +39,17 @@
     "husky": "^7.0.4",
     "jest": "27.2.3",
     "lerna": "^4.0.0",
+    "lint-staged": ">=10",
     "prettier": "^2.3.1",
     "standard-version": "^9.3.2",
     "ts-jest": "27.0.5",
     "typescript": "~4.4.3",
     "wrtc": "^0.4.7",
-    "yaml-crypt": "^0.7.6"
+    "yaml-crypt": "^0.7.6",
+    "zx": "^4.3.0"
+  },
+  "lint-staged": {
+    "*.{js,ts}": "eslint --cache --fix",
+    "*.{js,css,md}": "prettier --write"
   }
 }

packages/explorer/src/app.tsx

@@ -1,5 +1,5 @@
 import { create } from '@dstack-js/ipfs';
-import { Stack } from '@dstack-js/lib';
+import { Stack, Shard, ShardKind } from '@dstack-js/lib';
 import React, { useState } from 'react';
 import { Dashboard } from './dashboard';
 
@@ -14,10 +14,13 @@ export const App: React.FunctionComponent<{}> = () => {
 
     const ipfs = await create();
     const stack = await Stack.create(namespace, ipfs);
-    stack.debug();
 
     // @ts-expect-error
     window.stack = stack;
+    // @ts-expect-error
+    window.Shard = Shard;
+    // @ts-expect-error
+    window.ShardKind = ShardKind;
 
     setStack(stack);
   };

packages/lib/src/errors.ts

@@ -3,3 +3,15 @@ export class TimeoutError extends Error {
     super('timeout')
   }
 }
+
+export class UnknownShardKindError extends Error {
+  constructor() {
+    super('unknownShardKind')
+  }
+}
+
+export class InvalidShardPathError extends Error {
+  constructor(extensions: any = 'invalidShardPath') {
+    super(extensions)
+  }
+}

packages/lib/src/index.ts

@@ -1,3 +1,4 @@
 export * from './store'
 export * from './stack'
+export * from './shard'
 export * from './errors'

packages/lib/src/pubsub.ts

@@ -9,8 +9,14 @@ export interface Message<T> {
 }
 
 export class PubSub<T = unknown> {
+  public subscribed: string[] = []
+
   constructor(public ipfs: IPFS, public namespace: string) { }
 
+  public create<T = unknown>(namespace: string): PubSub<T> {
+    return new PubSub<T>(this.ipfs, `${this.namespace}/${namespace}`)
+  }
+
   private getTopic(topic: string): string {
     return `${this.namespace}.${topic}`
   }
@@ -27,14 +33,14 @@ export class PubSub<T = unknown> {
     const topic = await this.ipfs.pubsub.ls()
 
     return topic
+      .map((topic) => topic.replace(`${this.namespace}.`, ''))
       .filter((value) => {
-        if (ignoreInternals && value.includes('$')) {
+        if (ignoreInternals && value.startsWith('$$')) {
           return false
         }
 
         return true
       })
-      .map((topic) => topic.replace(`${this.namespace}.`, ''))
   }
 
   public async peers(topic: string): Promise<number> {
@@ -44,6 +50,8 @@ export class PubSub<T = unknown> {
   }
 
   public async subscribe(topic: string, listener: (msg: Message<T>) => void): Promise<void> {
+    this.subscribed.push(topic)
+
     await this.ipfs.pubsub.subscribe(this.getTopic(topic), (message) => {
       listener({
         from: message.from,
@@ -87,6 +95,12 @@ export class PubSub<T = unknown> {
   }
 
   public async unsubscribe(topic: string): Promise<void> {
+    delete this.subscribed[this.subscribed.indexOf(topic)]
+
     await this.ipfs.pubsub.unsubscribe(this.getTopic(topic))
   }
+
+  public async stop() {
+    await Promise.all(this.subscribed.map(this.unsubscribe))
+  }
 }

packages/lib/src/shard.ts

@@ -0,0 +1,83 @@
+import { CID } from 'multiformats/cid'
+import { Buffer } from 'buffer'
+import { UnknownShardKindError } from './errors'
+import { EventEmitter } from 'events'
+import { Stack } from './stack'
+import { InvalidShardPathError } from '.'
+
+export enum ShardKind {
+  // eslint-disable-next-line no-unused-vars
+  Binary = 'binary',
+  // eslint-disable-next-line no-unused-vars
+  Object = 'object'
+}
+
+export class Shard<TData = Buffer | Record<string, any>> {
+  private events = new EventEmitter()
+
+  private constructor(public stack: Stack, public cid: CID, public kind: ShardKind, public data: TData) { }
+
+  public async put(data: TData): Promise<void> {
+    switch (this.kind) {
+      case ShardKind.Binary:
+        this.cid = await this.stack.ipfs.block.put(new Uint8Array(data as unknown as Buffer))
+        break
+      case ShardKind.Object:
+        this.cid = await this.stack.ipfs.dag.put(data)
+        break
+      default:
+        throw new UnknownShardKindError()
+    }
+
+    this.events.emit('update', this)
+  }
+
+  public on(event: 'update', listener: (shard: Shard<TData>) => void): void {
+    this.events.on(event, listener)
+  }
+
+  public static async new<TData = unknown>(stack: Stack, kind: ShardKind, data: TData): Promise<Shard<TData>> {
+    const cid: CID = await stack.ipfs.dag.put({ empty: true })
+
+    const shard = new Shard<TData>(stack, cid, kind, data)
+    shard.put(data)
+
+    return shard
+  }
+
+  public static async create<TData = Buffer | Record<string, any>>(stack: Stack, cid: CID, kind: ShardKind): Promise<Shard<TData>> {
+    let data
+
+    switch (kind) {
+      case ShardKind.Binary:
+        data = Buffer.from(await stack.ipfs.block.get(cid))
+        break
+      case ShardKind.Object:
+        data = (await stack.ipfs.dag.get(cid)).value
+        break
+      default:
+        throw new UnknownShardKindError()
+    }
+
+    return new Shard(stack, cid, kind, data)
+  }
+
+  public static from<TData = Buffer | Record<string, any>>(stack: Stack, path: string): Promise<Shard<TData>> {
+    // eslint-disable-next-line no-unused-vars
+    const [_, shard, cid, kind] = path.split('/')
+
+    if (shard !== 'shard') {
+      throw new InvalidShardPathError()
+    }
+
+    try {
+      return Shard.create(stack, CID.parse(cid), kind as ShardKind)
+    } catch (error) {
+      throw new InvalidShardPathError(error)
+    }
+  }
+
+  public toString(): string {
+    return `/shard/${this.cid.toString()}/${this.kind}`
+  }
+}

packages/lib/src/stack.ts

@@ -1,26 +1,67 @@
-import type { IPFS } from 'ipfs-core'
+import { IPFS } from 'ipfs-core-types'
 import type { libp2p as Libp2p } from 'ipfs-core/src/components/network'
+import { CID } from 'multiformats/cid'
+import { Store } from '.'
 import { PubSub } from './pubsub'
-import { Store } from './store'
 
 export interface Peer {
   id: string
   address: string
 }
 
-export class Stack<TMessage = any> {
+export interface PeerAnnouncement {
+  kind: 'announcement'
+  peer: Peer
+}
+
+export type StackPubSubMessage = PeerAnnouncement
+
+export class Stack {
+  public pubsub: PubSub<StackPubSubMessage>
   public store: Store
-  public pubsub: PubSub<TMessage>
 
-  private constructor(public namespace: string, public ipfs: IPFS, public id: string) {
-    this.pubsub = new PubSub(ipfs, namespace)
-    this.store = new Store(ipfs, namespace, this.pubsub)
+  private announceInterval?: ReturnType<typeof setTimeout>
+  public announce = false
+
+  constructor(public namespace: CID, public ipfs: IPFS) {
+    this.pubsub = new PubSub(ipfs, namespace.toString())
+    this.store = new Store(this)
+  }
+
+  /**
+   * get this peer info
+   */
+  public async id(): Promise<Peer> {
+    const result = await this.ipfs.id()
+
+    return {
+      id: result.id,
+      address: result.addresses[0].toString()
+    }
   }
 
   private get libp2p() {
     return (this.ipfs as any).libp2p as Libp2p
   }
 
+  public async start(): Promise<void> {
+    this.announceInterval = setInterval(async () => {
+      if (!this.announce) return
+      await this.pubsub.publish('announce', {
+        kind: 'announcement',
+        peer: await this.id()
+      })
+    }, 250)
+
+    await this.store.start()
+  }
+
+  public async stop(): Promise<void> {
+    if (this.announceInterval) clearInterval(this.announceInterval)
+    await this.store.stop()
+    await this.pubsub.stop()
+  }
+
   /**
    * Create stack
    *
@@ -29,10 +70,11 @@ export class Stack<TMessage = any> {
    * @returns Stack instance
    */
   public static async create(namespace: string, ipfs: IPFS) {
-    const { id } = await ipfs.id()
-    const stack = new Stack(namespace, ipfs, id)
+    const cid = await ipfs.dag.put({ namespace })
+
+    const stack = new Stack(cid, ipfs)
+    await stack.start()
 
-    await stack.store.start()
     return stack
   }
 
@@ -88,12 +130,4 @@ export class Stack<TMessage = any> {
       })
     })
   }
-
-  /**
-   * Start logging debug events
-   */
-  public debug(): void {
-    this.onPeerConnect((peer) => console.log('New peer connected', peer.id, peer.address))
-    this.onPeerDisconnected((peer) => console.log('Peer disconnected', peer.id, peer.address))
-  }
 }