import { AbstractDatastore , BinaryFileData , Datastore , DatastoreCapabilities , DatastoreInitOpts , DatastoreOverview , DeleteResult , DocMetaSnapshotEvent , DocMetaSnapshotEventListener , DocMetaSnapshotEvents , DocMetaSnapshotOpts , DocMetaSnapshotResult , ErrorListener , FileSynchronizationEvent , FileSynchronizationEventListener , InitResult , PrefsProvider , SnapshotResult , SyncDocMap , SyncDocMaps , SynchronizationEvent , SynchronizationEventListener , SynchronizingDatastore , WriteFileOpts , WriteOpts } from './Datastore' ; import { DocMetaFileRef , DocMetaRef } from './DocMetaRef' ; import { Backend } from 'polar-shared/src/datastore/Backend' ; import { DocFileMeta } from 'polar-shared/src/datastore/DocFileMeta' ; import { Optional } from 'polar-shared/src/util/ts/Optional' ; import { DatastoreMutation , DefaultDatastoreMutation } from './DatastoreMutation' ; import { UUID } from 'polar-shared/src/metadata/UUID' ; import { Logger } from "polar-shared/src/logger/Logger" ; import { DocMetaComparisonIndex } from './DocMetaComparisonIndex' ; import { PersistenceLayers , SyncOrigin } from './PersistenceLayers' ; import { DocMetaSnapshotEventListeners , EventDeduplicator } from './DocMetaSnapshotEventListeners' ; import { ASYNC_NULL_FUNCTION , NULL_FUNCTION } from 'polar-shared/src/util/Functions' ; import { IEventDispatcher , SimpleReactor } from '../reactor/SimpleReactor' ; import { AsyncFunction } from 'polar-shared/src/util/AsyncWorkQueue' ; import firebase from 'firebase/app' ; import { Dictionaries } from 'polar-shared/src/util/Dictionaries' ; import { Datastores } from './Datastores' ; import { Either } from '../util/Either' ; import { BackendFileRefs } from './BackendFileRefs' ; import { IDocInfo } from "polar-shared/src/metadata/IDocInfo" ; import { FileRef } from "polar-shared/src/datastore/FileRef" ; import { Latch } from "polar-shared/src/util/Latch" ; import { InterceptedPrefsProvider , IPersistentPrefs } from "../util/prefs/Prefs" ; import { GetFileOpts , NetworkLayer } from "polar-shared/src/datastore/IDatastore" ; const log = Logger . create ( ) ; export interface CloudAwareDeleteResult extends DeleteResult { } Deprecated export class CloudAwareDatastore extends AbstractDatastore implements Datastore , SynchronizingDatastore { private static SNAPSHOT_ID = 0 ; public readonly id = 'cloud-aware' ; public readonly local : Datastore ; public readonly cloud : Datastore ; private readonly fileSynchronizationEventDispatcher : IEventDispatcher < FileSynchronizationEvent > = new SimpleReactor ( ) ; private readonly synchronizationEventDispatcher : IEventDispatcher < SynchronizationEvent > = new SimpleReactor ( ) ; private readonly docMetaSnapshotEventDispatcher : IEventDispatcher < DocMetaSnapshotEvent > = new SimpleReactor ( ) ; private readonly docMetaComparisonIndex = new DocMetaComparisonIndex ( ) ; private primarySnapshot ? : SnapshotResult ; public shutdownHook : AsyncFunction = ASYNC_NULL_FUNCTION ; constructor ( local , cloud ) { super ( ) ; this . local = local ; this . cloud = cloud ; } public async init ( errorListener = NULL_FUNCTION , opts = { noInitialSnapshot : false , noSync : false } ) { await this . initDelegates ( errorListener ) ; await this . initSnapshots ( errorListener , opts ) ; return { } ; } private async initDelegates ( errorListener ) { await Promise . all ( [ this . cloud . init ( errorListener , { noInitialSnapshot : true } ) , this . local . init ( errorListener ) ] ) ; } private async initSnapshots ( errorListener , opts ) { const snapshotListener = async ( event ) => this . docMetaSnapshotEventDispatcher . dispatchEvent ( event ) ; if ( ! opts . noSync ) { this . primarySnapshot = await this . snapshot ( snapshotListener , errorListener ) ; } } public async stop ( ) { await this . shutdownHook ( ) ; if ( this . primarySnapshot && this . primarySnapshot . unsubscribe ) { this . primarySnapshot . unsubscribe ( ) ; } await Promise . all ( [ this . cloud . stop ( ) , this . local . stop ( ) ] ) ; } public async contains ( fingerprint ) { return this . local . contains ( fingerprint ) ; } public async getDocMeta ( fingerprint ) { return this . local . getDocMeta ( fingerprint ) ; } public async getDocMetaSnapshot ( opts ) { return this . cloud . getDocMetaSnapshot ( opts ) ; } public async writeFile ( backend , ref , data , opts = { } ) { const datastoreMutation = opts . datastoreMutation || new DefaultDatastoreMutation ( ) ; const result = await this . local . writeFile ( backend , ref , data , opts ) ; datastoreMutation . written . resolve ( true ) ; this . cloud . writeFile ( backend , ref , data , opts ) . then ( ( ) => { datastoreMutation . committed . resolve ( true ) ; } ) . catch ( err => log . error ( "Unable to write file to cloud: " , err ) ) ; return result ; } public getFile ( backend , ref , opts = { } ) { Datastores . assertNetworkLayer ( this , opts . networkLayer ) ; if ( ! opts . networkLayer || opts . networkLayer === 'local' ) { return this . local . getFile ( backend , ref ) ; } else { return this . cloud . getFile ( backend , ref ) ; } } public containsFile ( backend , ref ) { return this . local . containsFile ( backend , ref ) ; } public async deleteFile ( backend , ref ) { await this . cloud . deleteFile ( backend , ref ) ; return this . local . deleteFile ( backend , ref ) ; } public async delete ( docMetaFileRef , datastoreMutation = new DefaultDatastoreMutation ( ) ) { datastoreMutation . written . get ( ) . then ( ( ) => { this . docMetaComparisonIndex . remove ( docMetaFileRef . fingerprint ) ; } ) . catch ( err => log . error ( "Could not handle delete: " , err ) ) ; await this . datastoreMutations . executeBatchedWrite ( datastoreMutation , async ( remoteCoordinator ) => { await this . cloud . delete ( docMetaFileRef , remoteCoordinator ) ; } , async ( localCoordinator ) => { await this . local . delete ( docMetaFileRef , localCoordinator ) ; } ) ; return { } ; } public async write ( fingerprint , data , docInfo , opts = { } ) { const datastoreMutation = opts . datastoreMutation || new DefaultDatastoreMutation ( ) ; const writeFileDatastoreMutation = new DefaultDatastoreMutation < boolean > ( ) ; if ( opts . writeFile ) { await this . writeFile ( opts . writeFile . backend , opts . writeFile , opts . writeFile . data , { datastoreMutation : writeFileDatastoreMutation } ) ; } else { writeFileDatastoreMutation . written . resolve ( true ) ; writeFileDatastoreMutation . committed . resolve ( true ) ; } datastoreMutation . written . get ( ) . then ( ( ) => { this . docMetaComparisonIndex . updateUsingDocInfo ( docInfo ) ; } ) . catch ( err => log . error ( "Could not handle delete: " , err ) ) ; await this . datastoreMutations . executeBatchedWrite ( datastoreMutation , async ( remoteCoordinator ) => { await writeFileDatastoreMutation . committed . get ( ) ; await this . cloud . write ( fingerprint , data , docInfo , { datastoreMutation : remoteCoordinator } ) ; } , async ( localCoordinator ) => { await this . local . write ( fingerprint , data , docInfo , { datastoreMutation : localCoordinator } ) ; } ) ; } public async getDocMetaRefs ( ) { return this . local . getDocMetaRefs ( ) ; } public async synchronizeDocs ( ... docMetaRefs ) { log . info ( "CloudAwareDatastore: synchronizeDocs: " , docMetaRefs ) ; const clearDocMeta = ( ... docMetaRefs ) : DocMetaRef [ ] => { return docMetaRefs . map ( current => { return { fingerprint : current . fingerprint } ; } ) ; } ; const cloudSyncOrigin = await PersistenceLayers . toSyncOrigin ( this . cloud , ... clearDocMeta ( ... docMetaRefs ) ) ; const localSyncOrigin = await PersistenceLayers . toSyncOrigin ( this . local , ... docMetaRefs ) ; await PersistenceLayers . synchronizeOrigins ( localSyncOrigin , cloudSyncOrigin , ASYNC_NULL_FUNCTION ) ; } public async snapshot ( docMetaSnapshotEventListener , errorListener = NULL_FUNCTION ) { const isPrimarySnapshot = this . primarySnapshot === undefined ; const snapshotID = CloudAwareDatastore . SNAPSHOT_ID ++ ; const deduplicatedListener = DocMetaSnapshotEventListeners . createDeduplicatedListener ( async ( docMetaSnapshotEvent ) => { await docMetaSnapshotEventListener ( docMetaSnapshotEvent ) ; } ) ; class InitialSnapshotLatch { public readonly syncDocMap : SyncDocMap = { } ; public readonly latch = new Latch < boolean > ( ) ; public readonly id : CloudDatastoreID ; private hasInitialTerminatedBatch : boolean = false ; private pending : number = 0 ; constructor ( id ) { this . id = id ; } private async handleSnapshot ( docMetaSnapshotEvent ) { try { if ( this . hasInitialTerminatedBatch ) { return ; } if ( ! docMetaSnapshotEvent . batch || docMetaSnapshotEvent . batch . id !== 0 ) { return ; } ++ this . pending ; const syncDocs = await DocMetaSnapshotEvents . toSyncDocs ( docMetaSnapshotEvent ) ; SyncDocMaps . putAll ( this . syncDocMap , syncDocs ) ; if ( docMetaSnapshotEvent . consistency === 'committed' && docMetaSnapshotEvent . batch ! . terminated ) { const nrDocs = Dictionaries . size ( this . syncDocMap ) ; this . hasInitialTerminatedBatch = true ; } } finally { -- this . pending ; if ( this . hasInitialTerminatedBatch && this . pending === 0 ) { this . latch . resolve ( true ) ; } } } public createSnapshot ( datastore ) { return datastore . snapshot ( async ( docMetaSnapshotEvent ) => { if ( ! initialSyncCompleted ) { await this . handleSnapshot ( docMetaSnapshotEvent ) ; } await synchronizingListener ( docMetaSnapshotEvent ) ; } , errorListener ) ; } } let initialSyncCompleted = false ; const localInitialSnapshotLatch = new InitialSnapshotLatch ( 'local' ) ; const cloudInitialSnapshotLatch = new InitialSnapshotLatch ( 'cloud' ) ; const synchronizingEventDeduplicator = DocMetaSnapshotEventListeners . createDeduplicatedListener ( async ( docMetaSnapshotEvent ) => { const handleEvent = async ( ) => { try { if ( initialSyncCompleted && isPrimarySnapshot ) { await this . handleSnapshotSynchronization ( docMetaSnapshotEvent , deduplicatedListener . listener ) ; } } finally { await docMetaSnapshotEventListener ( docMetaSnapshotEvent ) ; } } ; handleEvent ( ) . catch ( err => { log . error ( `Unable to handle synchronizing snapshot ${ snapshotID }` , err ) ; errorListener ( err ) ; } ) ; } , this . docMetaComparisonIndex ) ; const synchronizingListener = synchronizingEventDeduplicator . listener ; log . info ( "Local snapshot..." ) ; console . time ( "localSnapshot" ) ; const localSnapshotResultPromise = localInitialSnapshotLatch . createSnapshot ( this . local ) ; await localInitialSnapshotLatch . latch . get ( ) ; console . timeEnd ( "localSnapshot" ) ; log . info ( "Local snapshot...done" ) ; log . info ( "Cloud snapshot..." ) ; const cloudSnapshotResultPromise = cloudInitialSnapshotLatch . createSnapshot ( this . cloud ) ; await cloudInitialSnapshotLatch . latch . get ( ) ; log . info ( "Cloud snapshot...done" ) ; const localSyncOrigin = { datastore : this . local , syncDocMap : localInitialSnapshotLatch . syncDocMap } ; const cloudSyncOrigin = { datastore : this . cloud , syncDocMap : cloudInitialSnapshotLatch . syncDocMap } ; if ( isPrimarySnapshot ) { await PersistenceLayers . synchronizeOrigins ( localSyncOrigin , cloudSyncOrigin , deduplicatedListener . listener ) ; } initialSyncCompleted = true ; await localSnapshotResultPromise ; const cloudSnapshotResult = await cloudSnapshotResultPromise ; log . notice ( "INITIAL SNAPSHOT COMPLETE" ) ; return { unsubscribe : cloudSnapshotResult . unsubscribe } ; } private async handleSnapshotSynchronization ( docMetaSnapshotEvent , listener ) { const toLocalSyncOrigin = async ( ) : Promise < SyncOrigin > => { const docMetaFiles = docMetaSnapshotEvent . docMetaMutations . map ( current => { return { fingerprint : current . fingerprint } ; } ) ; const syncDocMap = await PersistenceLayers . toSyncDocMapFromDocs ( this . local , docMetaFiles ) ; return { datastore : this . local , syncDocMap } ; } ; const toCloudSyncOrigin = async ( ) : Promise < SyncOrigin > => { const syncDocs = await DocMetaSnapshotEvents . toSyncDocs ( docMetaSnapshotEvent ) ; return { datastore : this . cloud , syncDocMap : SyncDocMaps . fromArray ( syncDocs ) } ; } ; if ( docMetaSnapshotEvent . consistency !== 'committed' ) { return ; } for ( const docMetaMutation of docMetaSnapshotEvent . docMetaMutations ) { if ( docMetaMutation . mutationType === 'created' || docMetaMutation . mutationType === 'updated' ) { const cloudSyncOrigin = await toCloudSyncOrigin ( ) ; const localSyncOrigin = await toLocalSyncOrigin ( ) ; log . info ( "Transferring from cloud -> local..." ) ; await PersistenceLayers . transfer ( cloudSyncOrigin , localSyncOrigin , listener , 'cloud-to-local' ) ; log . info ( "Transferring from cloud -> local...done" ) ; } if ( docMetaMutation . mutationType === 'deleted' ) { const docMetaFileRef = await docMetaMutation . docMetaFileRefProvider ( ) ; const fileRefs = BackendFileRefs . toBackendFileRefs ( Either . ofRight ( docMetaFileRef . docInfo ) ) ; for ( const fileRef of fileRefs ) { await this . local . deleteFile ( fileRef . backend , fileRef ) ; } await this . local . delete ( docMetaFileRef ) ; log . info ( "File deleted: " , docMetaFileRef ) ; } } this . synchronizationEventDispatcher . dispatchEvent ( { ... docMetaSnapshotEvent , dest : 'local' } ) ; } public addFileSynchronizationEventListener ( eventListener ) { this . fileSynchronizationEventDispatcher . addEventListener ( eventListener ) ; } public addSynchronizationEventListener ( eventListener ) { this . synchronizationEventDispatcher . addEventListener ( eventListener ) ; } public addDocMetaSnapshotEventListener ( docMetaSnapshotEventListener ) { this . docMetaSnapshotEventDispatcher . addEventListener ( docMetaSnapshotEventListener ) ; } public async deactivate ( ) { await firebase . auth ( ) . signOut ( ) ; } public async overview ( ) { return Optional . first ( await this . local . overview ( ) , await this . cloud . overview ( ) ) . getOrUndefined ( ) ; } public capabilities ( ) { const networkLayers = new Set < NetworkLayer > ( [ 'local' , 'web' ] ) ; return { networkLayers , permission : { mode : 'rw' } , snapshots : true } ; } } /**
 * Represents a doc and its UUID.  The UUID is optional though as older docs
 * may not have a doc but in practice almost all docs will have a UUID.
 */ export interface DocUUID { fingerprint : string ; uuid ? : UUID ; } export type CloudDatastoreID = 'local' | 'cloud' ;