import { Datastore , DocMetaMutation , DocMetaSnapshotBatch , DocMetaSnapshotEventListener , SnapshotResult } from './Datastore' ; import { MemoryDatastore } from './MemoryDatastore' ; import { DiskDatastore } from './DiskDatastore' ; import { Logger } from 'polar-shared/src/logger/Logger' ; import { DocMetaFileRefs , DocMetaRef } from './DocMetaRef' ; import { DocMetas } from '../metadata/DocMetas' ; import { NULL_FUNCTION } from 'polar-shared/src/util/Functions' ; import { Percentages } from 'polar-shared/src/util/Percentages' ; import { ProgressTracker } from 'polar-shared/src/util/ProgressTracker' ; import { AsyncProviders } from 'polar-shared/src/util/Providers' ; import { DefaultPersistenceLayer } from './DefaultPersistenceLayer' ; import deepEqual from 'deep-equal' ; import { Preconditions } from 'polar-shared/src/Preconditions' ; import { AsyncFunction , AsyncWorkQueue } from 'polar-shared/src/util/AsyncWorkQueue' ; import { IDocInfo } from "polar-shared/src/metadata/IDocInfo" ; import { IDocMeta } from "polar-shared/src/metadata/IDocMeta" ; import { NetworkLayer } from "polar-shared/src/datastore/IDatastore" ; const log = Logger . create ( ) ; const ENV_POLAR_DATASTORE = 'POLAR_DATASTORE' ; export class Datastores { public static create ( ) { const name = process . env [ ENV_POLAR_DATASTORE ] ; if ( name === 'MEMORY' ) { log . info ( "Using memory datastore" ) ; return new MemoryDatastore ( ) ; } return new DiskDatastore ( ) ; } public static async getDocMetas ( datastore , listener , docMetaRefs ? ) { if ( ! docMetaRefs ) { docMetaRefs = await datastore . getDocMetaRefs ( ) ; } for ( const docMetaRef of docMetaRefs ) { const docMetaData = await datastore . getDocMeta ( docMetaRef . fingerprint ) ; if ( ! docMetaData ) { throw new Error ( "Could not find docMeta for fingerprint: " + docMetaRef . fingerprint ) ; } const docMeta = DocMetas . deserialize ( docMetaData , docMetaRef . fingerprint ) ; listener ( docMeta ) ; } } Deprecated public static async createCommittedSnapshot ( datastore , listener , batch ? ) { console . time ( "createCommittedSnapshot" ) ; if ( ! batch ) { batch = { id : 0 , terminated : false } ; } console . time ( "getDocMetaRefs" ) ; const docMetaFiles = await datastore . getDocMetaRefs ( ) ; console . timeEnd ( "getDocMetaRefs" ) ; const progressTracker = new ProgressTracker ( { total : docMetaFiles . length , id : `datastore:${ datastore . id }#snapshot` } ) ; const durations = { data : 0 , docMeta : 0 , docInfo : 0 , docMetaFileRef : 0 } ; for ( const docMetaFile of docMetaFiles ) { const dataProvider = AsyncProviders . memoize ( async ( ) => { const before = Date . now ( ) ; try { return await datastore . getDocMeta ( docMetaFile . fingerprint ) ; } finally { durations . data += Date . now ( ) - before ; } } ) ; const docMetaProvider = AsyncProviders . memoize ( async ( ) => { const before = Date . now ( ) ; try { const data = await dataProvider ( ) ; return DocMetas . deserialize ( data ! , docMetaFile . fingerprint ) ; } finally { durations . docMeta += Date . now ( ) - before ; } } ) ; const docInfoProvider = AsyncProviders . memoize ( async ( ) => { const before = Date . now ( ) ; try { return ( await docMetaProvider ( ) ) . docInfo ; } finally { durations . docInfo += Date . now ( ) - before ; } } ) ; const docMetaFileRefProvider = AsyncProviders . memoize ( async ( ) => { const before = Date . now ( ) ; try { return DocMetaFileRefs . createFromDocInfo ( await docInfoProvider ( ) ) ; } finally { durations . docMetaFileRef += Date . now ( ) - before ; } } ) ; const docMetaMutation = { fingerprint : docMetaFile . fingerprint , docMetaFileRefProvider , dataProvider , docMetaProvider , docInfoProvider , mutationType : 'created' , hasPendingWrites : false , fromCache : false } ; await listener ( { datastore : datastore . id , progress : progressTracker . incr ( ) , consistency : 'committed' , docMetaMutations : [ docMetaMutation ] , batch } ) ; } console . log ( "Durations: " , durations ) ; await listener ( { datastore : datastore . id , progress : progressTracker . terminate ( ) , consistency : 'committed' , docMetaMutations : [ ] , batch : { id : batch . id , terminated : true , } } ) ; console . timeEnd ( "createCommittedSnapshot" ) ; return { } ; } /**
     * Remove all the docs in a datastore.  Only do this for testing and for
     * very important use cases.
     */ public static async purge ( datastore , purgeListener = NULL_FUNCTION ) { log . debug ( "Getting doc meta refs..." ) ; const docMetaFiles = await datastore . getDocMetaRefs ( ) ; log . debug ( "Getting doc meta refs...done" ) ; let completed = 0 ; const total = docMetaFiles . length ; const work = [ ] ; const asyncWorkQueue = new AsyncWorkQueue ( work ) ; for ( const docMetaFile of docMetaFiles ) { work . push ( async ( ) => { log . debug ( `Purging file: ${ docMetaFile . fingerprint } in datastore ${ datastore . id }` ) ; const data = await datastore . getDocMeta ( docMetaFile . fingerprint ) ; const docMeta = DocMetas . deserialize ( data ! , docMetaFile . fingerprint ) ; const docMetaFileRef = DocMetaFileRefs . createFromDocInfo ( docMeta . docInfo ) ; await datastore . delete ( docMetaFileRef ) ; ++ completed ; const progress = Percentages . calculate ( completed , total ) ; purgeListener ( { completed , total , progress } ) ; } ) ; } await asyncWorkQueue . execute ( ) ; if ( total === 0 ) { purgeListener ( { completed , total , progress : 100 } ) ; } } /**
     * Compare two filesystems and make sure they're consistent.
     *
     */ public static async checkConsistency ( datastore0 , datastore1 ) { const manifest0 = await this . toDocInfoManifest ( datastore0 ) ; const manifest1 = await this . toDocInfoManifest ( datastore1 ) ; const consistent = deepEqual ( manifest0 , manifest1 ) ; return { consistent , manifest0 , manifest1 } ; } public static async toDocInfoManifest ( datastore ) { const persistenceLayer = new DefaultPersistenceLayer ( datastore ) ; const docMetaRefs = await datastore . getDocMetaRefs ( ) ; const docMetaFiles = [ ... docMetaRefs ] . sort ( ( d0 , d1 ) => d0 . fingerprint . localeCompare ( d1 . fingerprint ) ) ; const result = [ ] ; for ( const docMetaFile of docMetaFiles ) { const docMeta = await persistenceLayer . getDocMeta ( docMetaFile . fingerprint ) ; Preconditions . assertPresent ( docMeta , "toDocInfoManifest could not find docMeta for " + docMetaFile . fingerprint ) ; result . push ( docMeta ! . docInfo ) ; } return result ; } /**
     * Assert that the specified network layer is supported by this datastore.
     */ public static assertNetworkLayer ( datastore , networkLayer ? ) { if ( ! networkLayer ) { return ; } const capabilities = datastore . capabilities ( ) ; if ( ! capabilities . networkLayers . has ( networkLayer ) ) { throw new Error ( `Datastore '${ datastore . id }' does not support ${ networkLayer } only ${ capabilities . networkLayers }` ) ; } } } export type DocMetaListener = ( docMeta : IDocMeta ) => void ; export interface PurgeEvent { readonly completed : number ; readonly total : number ; readonly progress : number ; } export type PurgeListener = ( purgeEvent : PurgeEvent ) => void ; export interface DatastoreConsistency { readonly consistent : boolean ; readonly manifest0 : ReadonlyArray < IDocInfo > ; readonly manifest1 : ReadonlyArray < IDocInfo > ; }