new: wire up remote leveldb with multileveldown and stream new explored chunks

package.json

@@ -4,7 +4,7 @@
   "description": "",
   "main": "index.js",
   "scripts": {
-    "start": "vite --port 8083"
+    "start": "vite --port 8080"
   },
   "repository": {
     "type": "git",
@@ -28,15 +28,16 @@
     "events": "^3.2.0",
     "http-browserify": "^1.7.0",
     "https-browserify": "^1.0.0",
-    "idb": "^5.0.7",
     "json-stable-stringify": "^1.0.1",
     "lodash": "^4.17.20",
+    "simple-websocket": "^9.0.0",
     "stream-browserify": "^3.0.0",
     "url": "^0.11.0",
     "util": "^0.12.3"
   },
   "devDependencies": {
     "@types/events": "^3.0.0",
+    "@types/levelup": "^4.3.0",
     "@types/node": "^14.14.3",
     "vite": "^1.0.0-rc.8"
   }

src/LocalStorageManager.ts

@@ -5,35 +5,22 @@ import {
   Location,
   WorldCoords,
 } from './GlobalTypes';
-import {
-  EthAddress,
-  contractAddress,
-} from './Contract';
-import { openDB, IDBPDatabase } from 'idb';
 import _, { Cancelable } from 'lodash';
-import stringify from 'json-stable-stringify';
+import type { LevelUp } from 'levelup'
 
 const MAX_CHUNK_SIZE = 256;
 
-enum ObjectStore {
-  DEFAULT = 'default',
-  BOARD = 'knownBoard',
-  UNCONFIRMED_ETH_TXS = 'unminedEthTxs',
-}
-
 enum DBActionType {
-  UPDATE,
-  DELETE,
+  UPDATE = 'put',
+  DELETE = 'del',
 }
 
 interface DBAction {
   type: DBActionType;
-  dbKey: string;
-  dbValue?: ExploredChunkData;
+  key: string;
+  value?: ExploredChunkData;
 }
 
-type DBTx = DBAction[];
-
 export interface ChunkStore {
   hasMinedChunk: (chunkFootprint: ChunkFootprint) => boolean;
 }
@@ -227,112 +214,66 @@ export const addToChunkMap = (
 
 
 export class LocalStorageManager implements ChunkStore {
-  private db: IDBPDatabase;
-  private cached: DBTx[];
+  private db: LevelUp;
+  private cached: DBAction[];
   private throttledSaveChunkCacheToDisk: (() => Promise<void>) & Cancelable;
   private nUpdatesLastTwoMins = 0; // we save every 5s, unless this goes above 50
   private chunkMap: Map<string, ExploredChunkData>;
-  private confirmedTxHashes: Set<string>;
-  private account: EthAddress;
 
-  constructor(db: IDBPDatabase, account: EthAddress) {
+  constructor(db: LevelUp) {
     this.db = db;
     this.cached = [];
-    this.confirmedTxHashes = new Set<string>();
     this.throttledSaveChunkCacheToDisk = _.throttle(
       this.saveChunkCacheToDisk,
       2000 // TODO
     );
     this.chunkMap = new Map<string, ExploredChunkData>();
-    this.account = account;
   }
 
   destroy(): void {
     // no-op; we don't actually destroy the instance, we leave the db connection open in case we need it in the future
   }
 
-  static async create(account: EthAddress): Promise<LocalStorageManager> {
-    const db = await openDB(`darkforest-${contractAddress}-${account}`, 1, {
-      upgrade(db) {
-        db.createObjectStore(ObjectStore.DEFAULT);
-        db.createObjectStore(ObjectStore.BOARD);
-        db.createObjectStore(ObjectStore.UNCONFIRMED_ETH_TXS);
-      },
+  private async bulkSetKeyInCollection(updateChunkTxs: DBAction[]): Promise<void> {
+    const chunks = updateChunkTxs.map((chunk) => {
+      if (chunk.value) {
+        return { ...chunk, value: toLSMChunk(chunk.value) }
+      } else {
+        return chunk;
+      }
     });
-    const localStorageManager = new LocalStorageManager(db, account);
-
-    await localStorageManager.loadIntoMemory();
-
-    return localStorageManager;
-  }
-
-  private async getKey(key: string): Promise<string | null | undefined> {
-    return await this.db.get(
-      ObjectStore.DEFAULT,
-      `${contractAddress}-${this.account}-${key}`
-    );
-  }
-
-  private async setKey(key: string, value: string): Promise<void> {
-    await this.db.put(
-      ObjectStore.DEFAULT,
-      value,
-      `${contractAddress}-${this.account}-${key}`
-    );
-  }
-
-  private async bulkSetKeyInCollection(
-    updateChunkTxs: DBTx[],
-    collection: ObjectStore
-  ): Promise<void> {
-    const tx = this.db.transaction(collection, 'readwrite');
-    updateChunkTxs.forEach((updateChunkTx) => {
-      updateChunkTx.forEach(({ type, dbKey: key, dbValue: value }) => {
-        if (type === DBActionType.UPDATE) {
-          tx.store.put(toLSMChunk(value as ExploredChunkData), key);
-        } else if (type === DBActionType.DELETE) {
-          tx.store.delete(key);
+    await new Promise((resolve, reject) => {
+      this.db.batch(chunks as any, (err) => {
+        if (err) {
+          return reject(err);
         }
+
+        resolve()
       });
     });
-    await tx.done;
   }
 
-  private async loadIntoMemory(): Promise<void> {
-    // we can't bulk get all chunks, since idb will crash/hang
-    // we also can't assign random non-primary keys and query on ranges
-    // so we append a random alphanumeric character to the front of keys
-    // and then bulk query for keys starting with 0, then 1, then 2, etc.
-    const borders = '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ~';
-    for (let idx = 0; idx < borders.length - 1; idx += 1) {
-      (
-        await this.db.getAll(
-          ObjectStore.BOARD,
-          IDBKeyRange.bound(borders[idx], borders[idx + 1], false, true)
-        )
-      ).forEach((chunk: LSMChunkData) => {
-        this.updateChunk(toExploredChunk(chunk), true);
-      });
-    }
+  async loadIntoMemory(): Promise<void> {
+    return new Promise((resolve, reject) => {
+      this.db.createReadStream()
+        .on('data', ({ key, value: chunk }) => {
+          this.updateChunk(toExploredChunk(chunk), true);
+        })
+        .on('error', (err) => {
+          console.error('error occurred sinking map', err);
+          reject(err);
+        })
+        .on('end', (data) => {
+          console.log('Initial map has been sunk', data);
+          resolve();
+        });
+    });
   }
 
   private async saveChunkCacheToDisk() {
     const toSave = [...this.cached]; // make a copy
     this.cached = [];
-    await this.bulkSetKeyInCollection(toSave, ObjectStore.BOARD);
-  }
-
-  public async getHomeCoords(): Promise<WorldCoords | null> {
-    const homeCoords = await this.getKey('homeCoords');
-    if (homeCoords) {
-      const parsed = JSON.parse(homeCoords) as { x: number; y: number };
-      return parsed;
-    }
-    return null;
-  }
-
-  public async setHomeCoords(coords: WorldCoords): Promise<void> {
-    await this.setKey('homeCoords', stringify(coords));
+    await this.bulkSetKeyInCollection(toSave);
   }
 
   public hasMinedChunk(chunkLoc: ChunkFootprint): boolean {
@@ -361,14 +302,14 @@ export class LocalStorageManager implements ChunkStore {
     if (this.hasMinedChunk(e.chunkFootprint)) {
       return;
     }
-    const tx: DBTx = [];
+    const tx: DBAction[] = [];
 
     // if this is a mega-chunk, delete all smaller chunks inside of it
     const minedSubChunks = this.getMinedSubChunks(e);
     for (const subChunk of minedSubChunks) {
       tx.push({
         type: DBActionType.DELETE,
-        dbKey: getChunkKey(subChunk.chunkFootprint),
+        key: getChunkKey(subChunk.chunkFootprint),
       });
     }
 
@@ -379,25 +320,25 @@ export class LocalStorageManager implements ChunkStore {
       (chunk) => {
         tx.push({
           type: DBActionType.UPDATE,
-          dbKey: getChunkKey(chunk.chunkFootprint),
-          dbValue: chunk,
+          key: getChunkKey(chunk.chunkFootprint),
+          value: chunk,
         });
       },
       (chunk) => {
         tx.push({
           type: DBActionType.DELETE,
-          dbKey: getChunkKey(chunk.chunkFootprint),
+          key: getChunkKey(chunk.chunkFootprint),
         });
       },
       MAX_CHUNK_SIZE
     );
 
     // modify in-memory store
     for (const action of tx) {
-      if (action.type === DBActionType.UPDATE && action.dbValue) {
-        this.chunkMap.set(action.dbKey, action.dbValue);
+      if (action.type === DBActionType.UPDATE && action.value) {
+        this.chunkMap.set(action.key, action.value);
       } else if (action.type === DBActionType.DELETE) {
-        this.chunkMap.delete(action.dbKey);
+        this.chunkMap.delete(action.key);
       }
     }
 
@@ -406,7 +347,7 @@ export class LocalStorageManager implements ChunkStore {
       return;
     }
 
-    this.cached.push(tx);
+    this.cached = [...this.cached, ...tx];
 
     // save chunks every 5s if we're just starting up, or 30s once we're moving
     this.recomputeSaveThrottleAfterUpdate();

src/index.ts

@@ -1,12 +1,15 @@
 import { CanvasRenderer } from './CanvasRenderer';
-import { emptyAddress, address, Contract } from './Contract';
-import { LocalStorageManager } from './LocalStorageManager';
+import { Contract } from './Contract';
+import { LocalStorageManager, toExploredChunk } from './LocalStorageManager';
 import {
   PlanetHelper,
   VoyageContractData,
   PlanetVoyageIdMap,
 } from './PlanetHelper';
 import { Viewport } from './Viewport';
+import multileveldown from '../vendor/multileveldown-browser';
+import LevelRangeEmitter from '../vendor/level-range-emitter-browser';
+import WebSocket from 'simple-websocket/simplewebsocket.min';
 
 async function start() {
   const canvas = document.querySelector('canvas');
@@ -19,31 +22,54 @@ async function start() {
   const homeCoords = { x: 0, y: 0 };
   const widthInWorldUnits = 250;
   const endTimeSeconds = 1609372800;
-  // TODO: Use mine for testing
-  const myAddress = emptyAddress;
 
-  const chunkStore = await LocalStorageManager.create(myAddress);
+  const db = multileveldown.client({ valueEncoding: 'json', retry: true });
+  const websocketStream = new WebSocket('ws://localhost:8082');
+  const lre = LevelRangeEmitter.client(db);
+  lre.session(db.connect(), websocketStream);
 
-  // initialize dependencies according to a DAG
+  const chunkStore = new LocalStorageManager(db);
+
+  lre.emitter.subscribe((key, type) => {
+    console.log('updated', key, type);
+    if (type === 'put') {
+      db.get(key, (err, value) => {
+        if (err) {
+          console.error('Failed to store chunk in memory:', key);
+          console.error(err);
+          return;
+        }
+
+        console.log('Storing chunk:', key, value);
+        chunkStore.updateChunk(toExploredChunk(value), true);
+      });
+    }
+  });
 
-  // first we initialize the ContractsAPI and get the user's eth account, and load contract constants + state
   const contractsAPI = await Contract.create();
 
-  // get data from the contract
-  const contractConstants = await contractsAPI.getConstants();
+  const [
+    _mapLoaded,
+    contractConstants,
+    worldRadius,
+    allArrivals,
+    planets,
+  ] = await Promise.all([
+    chunkStore.loadIntoMemory(),
+    contractsAPI.getConstants(),
+    contractsAPI.getWorldRadius(),
+    contractsAPI.getAllArrivals(),
+    contractsAPI.getPlanets(),
+  ]);
+
   const perlinThresholds = [
     contractConstants.PERLIN_THRESHOLD_1,
     contractConstants.PERLIN_THRESHOLD_2,
   ];
-  // const players = await contractsAPI.getPlayers();
-  const worldRadius = await contractsAPI.getWorldRadius();
 
   const arrivals: VoyageContractData = {};
   const planetVoyageIdMap: PlanetVoyageIdMap = {};
-  const allArrivals = await contractsAPI.getAllArrivals();
-  // fetch planets after allArrivals, since an arrival to a new planet might be sent
-  // while we are fetching
-  const planets = await contractsAPI.getPlanets();
+
   planets.forEach((planet, locId) => {
     if (planets.has(locId)) {
       planetVoyageIdMap[locId] = [];