feat(bin): storage accesors [WIP]

bin/delete-objects/ObjectStorage.ts

@@ -1,6 +1,12 @@
 import AWS from 'aws-sdk';
 import { existsSync, createReadStream } from 'fs';
 import readline from 'readline';
+import { createHash } from 'crypto';
+
+import { ShardsRepository } from '../../lib/core/shards/Repository';
+import { Shard } from '../../lib/core/shards/Shard';
+import { MongoDBCollections, TempShardDocument } from './temp-shard.model';
+import { ObjectId } from 'mongodb';
 
 export interface StorageObject {
   Key: string;
@@ -131,3 +137,82 @@ export class S3ObjectStorageReader implements ObjectStorageReader {
   }
 }
 
+export interface ShardsReader {
+  list(pageSize: number): AsyncGenerator<Shard>;
+  isV1(s: Shard): boolean;
+}
+
+interface TempShardsReader {
+  list(pageSize: number): AsyncGenerator<TempShardDocument>;
+}
+
+interface TempShardsWriter {
+  write(shard: Shard): Promise<void>;
+}
+
+function isValidShardHash(hash: string) {
+  return !!hash.match(/^[a-f0-9]{40}$/);
+}
+
+export function ripemd160(content: string) {
+  if (!isValidShardHash(content)) {
+    throw Error('Invalid hex string');
+  }
+
+  return createHash('ripemd160').update(Buffer.from(content, 'hex')).digest('hex');
+}
+
+export class DatabaseShardsReader implements ShardsReader {  
+  constructor(private readonly shardsRepository: ShardsRepository) {}
+
+  isV1(s: Shard): boolean {
+    const doesNotHaveUuid = !s.uuid;
+
+    return isValidShardHash(s.hash) && doesNotHaveUuid;
+  }
+
+  async* list(pageSize = 1000): AsyncGenerator<Shard> {
+    let offset = 0;
+    do {
+      const shards = await this.shardsRepository.findWithNoUuid(
+        pageSize,
+        offset,
+      );
+      for (const shard of shards) {
+        yield shard;
+      }
+      offset += shards.length;
+    } while (offset % pageSize === 0);
+  }
+}
+
+export class DatabaseTempShardsWriter implements TempShardsWriter {
+  constructor(private readonly tempShards: MongoDBCollections['tempShards']) {}
+
+  async write(shard: Shard): Promise<void> {
+    await this.tempShards.insertOne({
+      hash: shard.hash,
+      objectStorageHash: ripemd160(shard.hash),
+      shardId: new ObjectId(shard.id),
+      size: shard.size,
+    });
+  }
+}
+
+export class DatabaseTempShardsReader implements TempShardsReader {
+  constructor(private readonly tempShards: MongoDBCollections['tempShards']) {}
+
+  async* list(pageSize = 1000): AsyncGenerator<TempShardDocument> {
+    let offset = 0;
+    do {
+      const tempShards = await this.tempShards.find(
+        {},
+        { limit: pageSize, skip: offset },
+      ).toArray();
+      for (const tempShard of tempShards) {
+        yield tempShard;
+      }
+      offset += tempShards.length;
+    } while (offset % pageSize === 0);
+  }
+}

bin/delete-objects/temp-shard.model.ts

@@ -0,0 +1,54 @@
+import { ObjectId, Document, Collection, Db, MongoClient } from 'mongodb';
+
+export interface MongoDBCollections {
+  tempShards: Collection<TempShardDocument>;
+}
+
+interface TempShard extends Document {
+  hash: string;
+  objectStorageHash: string;
+  shardId: string;
+  size: number;
+}
+
+export interface TempShardDocument extends Omit<TempShard, 'shardId'> {
+  shardId: ObjectId;
+}
+
+export class MongoDB {
+  private uri: string;
+  private db: Db | null;
+  private client: MongoClient;
+
+  constructor(uri: string) {
+    this.uri = uri;
+    this.db = null;
+    this.client = new MongoClient(this.uri);
+  }
+
+  get URI() {
+    return this.uri;
+  }
+
+  async connect(): Promise<MongoDB> {
+    await this.client.connect();
+
+    this.db = this.client.db('__inxt-network');
+
+    return this;
+  }
+
+  getCollections(): MongoDBCollections {
+    if (!this.db) {
+      throw new Error('Database not initialized');
+    }
+
+    return {
+      tempShards: this.db.collection<TempShardDocument>('tempshards'),
+    };
+  }
+
+  disconnect(): Promise<void> {
+    return this.client.close();
+  }
+}