import { Ipfs } from "ipfs" ; import { IdentityProvider } from "orbit-db-identity-provider" ; import { InsertOptions , InsertOneOptions , DocumentInterface , FindOptionsInterface , FindOneAndUpdateOptionsInterface , FindOneAndDeleteOptionsInterface , UpdateOptionsInterface , UpdateOneOptionsInterface , UpdateManyOptionsInterface , DeleteOneOptionsInterface , DeleteManyOptionsInterface , ImportOptionsInterface , ImportStreamOptionsInterface , ExportOptionsInterface , IStoreOptions , } from "./interfaces" ; import { LogEntry } from "ipfs-log" ; const OrbitdbStore = require ( "orbit-db-store" ) ; const ObjectId = require ( "bson-objectid" ) ; const CID = require ( "cids" ) ; const CollectionIndex = require ( "./CollectionIndex" ) ; const DagCbor = require ( "ipld-dag-cbor" ) ; class Collection extends OrbitdbStore { constructor ( ipfs , id , dbname , options ) { const opts = Object . assign ( { } , { Index : CollectionIndex } ) ; Object . assign ( opts , options ) ; super ( ipfs , id , dbname , opts ) ; this . _type = "aviondb.collection" ; this . events . on ( "write" , ( address , entry , heads ) => { this . _index . handleEntry ( entry ) ; } ) ; this . events . on ( "replicate.progress" , ( address , hash , entry , progress , total ) => { this . _index . handleEntry ( entry ) ; } ) ; } param { Array < DocumentInterface > } docs param { InsertOptions } options param { Function } callback returns { Promise < string > } insert ( docs , options ? , callback ? ) { for ( const doc of docs ) { if ( ! doc . _id ) { doc . _id = ObjectId . generate ( ) ; } } return this . _addOperation ( { op : "INSERT" , value : docs , } ) ; } param { DocumentInterface } doc param { InsertOneOptions } options param { Function } callback returns { Promise < string > } async insertOne ( doc , options ? , callback ? ) { if ( typeof doc !== "object" ) throw new Error ( "Object documents are only supported" ) ; return await this . insert ( [ doc ] ) ; } param { object } query param { object | string } projection param { FindOptionsInterface } options param { Function } callback returns { Promise < Array < DocumentInterface > > } find ( query , projection ? , options ? , callback ? ) { return this . _index . find ( query , projection , options , callback ) ; } param { JSON Object param { object | string } projection param { object } options param { Function } callback returns { Promise < DocumentInterface > } findOne ( query , projection ? , options ? , callback ? ) { return this . _index . findOne ( query ) ; } param { object } query param { object } modification param { FindOneAndUpdateOptionsInterface } options param { Function } callback returns { Promise < DocumentInterface > } async findOneAndUpdate ( filter = { } , modification , options ? , callback ? ) { const doc = await this . findOne ( filter ) ; if ( doc ) { await this . _addOperation ( { op : "UPDATE" , value : [ doc . _id ] , modification : modification , } ) ; } return doc ; } param { object } filter param { FindOneAndDeleteOptionsInterface } options param { Function } callback returns { Promise < DocumentInterface > } async findOneAndDelete ( filter = { } , options ? , callback ? ) { const doc = await this . findOne ( filter ) ; if ( doc ) { await this . _addOperation ( { op : "DELETE" , value : [ doc . _id ] , } ) ; } return doc ; } param { object | string | number } _id param { object | string } projection param { object } options param { Function } callback returns { Promise < DocumentInterface > } findById ( _id , projection ? , options ? , callback ? ) { return this . _index . findById ( _id , projection , options , callback ) ; } param { object | string | number } _id param { object } options param { Function } callback returns { Promise < DocumentInterface > } async findByIdAndDelete ( _id , options ? , callback ? ) { const doc = this . _index . findById ( _id ) ; if ( doc ) { await this . _addOperation ( { op : "DELETE" , value : [ doc . _id ] , } ) ; } return doc ; } param { object | string | number } _id param { object } modification param { object } options param { Function } callback returns { Promise < DocumentInterface > } async findByIdAndUpdate ( _id , modification , options = { } , callback ? ) { const doc = await this . _index . findById ( _id ) ; if ( doc ) { await this . _addOperation ( { op : "UPDATE" , value : [ doc . _id ] , modification : modification , options : options , } ) ; } return doc ; } param { object } filter param { object } modification param { UpdateOptionsInterface } options param { Function } callback returns { Promise < Array < DocumentInterface > > } async update ( filter = { } , modification , options = { } , callback ? ) { const ids = [ ] ; const docs = [ ] ; if ( options . multi ) { docs . push ( ... ( await this . find ( filter ) ) ) ; ids . push ( ... docs . map ( ( item ) => item . _id ) ) ; } if ( options . upsert && ids . length === 0 ) { } else if ( Object . keys ( options ) . length === 0 && options . constructor === Object ) { const doc = await this . findOne ( filter ) ; if ( doc ) { docs . push ( doc ) ; ids . push ( ... docs . map ( ( item ) => item . _id ) ) ; } } await this . _addOperation ( { op : "UPDATE" , value : ids , modification : modification , options : options , } ) ; return docs ; } param { object } filter param { object } modification param { UpdateOneOptionsInterface } options param { Function } callback returns { Promise < DocumentInterface > } async updateOne ( filter = { } , modification , options = { } , callback ? ) { const doc = await this . findOne ( filter ) ; if ( doc ) { await this . _addOperation ( { op : "UPDATE" , value : [ doc . _id ] , modification : modification , options : options , } ) ; } return doc ; } param { object } filter param { object } modification param { UpdateManyOptionsInterface } options param { Function } callback returns { Promise < Array < DocumentInterface > > } async updateMany ( filter = { } , modification , options = { } , callback ? ) { const docs = await this . find ( filter ) ; const ids = docs . map ( ( item ) => item . _id ) ; await this . _addOperation ( { op : "UPDATE" , value : ids , modification : modification , options : options , } ) ; return docs ; } param { object } filter param { DeleteOneOptionsInterface } options param { Function } callback returns { Promise < DocumentInterface > } async deleteOne ( filter = { } , options ? , callback ? ) { const doc = await this . findOne ( filter ) ; if ( doc ) { await this . _addOperation ( { op : "DELETE" , value : [ doc . _id ] , } ) ; } return doc ; } param { object } filter param { DeleteManyOptionsInterface } options param { Function } callback returns { Promise < Array < DocumentInterface > > } async deleteMany ( filter = { } , options ? , callback ? ) { const docs = await this . find ( filter ) ; const ids = docs . map ( ( item ) => item . _id ) ; if ( ids . length > 0 ) { await this . _addOperation ( { op : "DELETE" , value : ids , } ) ; } return docs ; } param key param query returns { Promise < Array < DocumentInterface > > } distinct ( key , query ) { return this . _index . distinct ( key , query ) ; } returns { string } async getHeadHash ( ) { try { return await this . _oplog . toMultihash ( ) ; } catch { return null ; } } param { string } hash param { boolean } stopWrites returns { Promise < null > } async syncFromHeadHash ( hash , stopWrites ? ) { if ( new CID ( hash ) . equals ( new CID ( await this . getHeadHash ( ) ) ) ) { return ; } const { value } = await this . _ipfs . dag . get ( hash ) ; if ( value . id !== this . id ) { throw "Head Hash ID does not match store ID." ; } const heads = [ ] ; for ( const hashOfHead of value . heads ) { const val = ( await this . _ipfs . dag . get ( hashOfHead ) ) . value ; val . hash = hashOfHead . toBaseEncodedString ( "base58btc" ) ; heads . push ( val ) ; } if ( stopWrites ) { this . _opqueue . pause ( ) ; } await this . sync ( heads ) ; this . _opqueue . start ( ) ; } param { any } data_in param { ImportOptionsInterface } options param { Function } progressCallback async import ( data_in , options = { } , progressCallback ? ) { if ( ! options . overwrite ) { options . overwrite = false ; } if ( ! options . type ) { options . type = "json_mongo" ; } if ( ! options . batchSize ) { options . batchSize = 25 ; } let deserialized_object = { } ; if ( options . type === "cbor" ) { deserialized_object = DagCbor . util . deserialize ( data_in ) ; } else if ( options . type === "json_mongo" ) { deserialized_object = JSON . parse ( data_in ) ; } else if ( options . type === "raw" ) { deserialized_object = data_in ; } else { throw `Unknown options.type: ${ options . type }` ; } async function * streamGenerator ( ) { for ( const entry of deserialized_object ) { yield entry ; } } const objTotalLength = deserialized_object . length ; await this . importStream ( streamGenerator ( ) , objTotalLength , { batchSize : options . batchSize } , progressCallback ) ; } param { AsyncIterable } stream param { number } length param { ImportStreamOptionsInterface } options param { Function } progressCallback async importStream ( stream , length , options , progressCallback ? ) { const totalLength = length ; let currentLength = 0 ; let queue = [ ] ; for await ( const entry of stream ) { if ( queue . length >= options . batchSize ) { await this . insert ( queue ) ; queue = [ ] ; if ( progressCallback ) { const progressPercent = ( currentLength / totalLength ) * 100 ; progressCallback ( currentLength , totalLength , progressPercent ) ; } } else { if ( typeof entry . _id === "object" ) { entry . _id = entry . _id . $oid ; } queue . push ( entry ) ; currentLength += queue . length ; } } if ( queue . length > 0 ) { await this . insert ( queue ) ; currentLength += queue . length ; if ( progressCallback ) { const progressPercent = ( currentLength / totalLength ) * 100 ; progressCallback ( currentLength , totalLength , progressPercent ) ; } queue = [ ] ; } } param { ExportOptionsInterface } options returns { Promise < string | DocumentInterface [ ] > } async export ( options = { } ) { if ( ! options . cursor ) { options . cursor = { } ; } if ( ! options . type ) { options . type = "json_mongo" ; } if ( ! options . query ) { options . query = { } ; } const results = await this . find ( options . query , null , options . cursor ) ; switch ( options . type ) { case "json_mongo" : { return JSON . stringify ( results ) ; } case "cbor" : { return DagCbor . util . serialize ( results ) ; } case "raw" : { return results ; } default : { throw `Unknown options.type: ${ options . type }` ; } } } async drop ( ) { super . drop ( ) ; } } export default Collection ;