import { PersistenceLayer } from "./PersistenceLayer" ; import { ASYNC_NULL_FUNCTION , NULL_FUNCTION } from "polar-shared/src/util/Functions" ; import { AsyncFunction , AsyncWorkQueue } from 'polar-shared/src/util/AsyncWorkQueue' ; import { DocMetaRef } from "./DocMetaRef" ; import { Datastore , DocMetaSnapshotEvent , DocMetaSnapshotEventListener , SyncDoc , SyncDocMap , SyncDocs } from './Datastore' ; import { UUIDs } from '../metadata/UUIDs' ; import { ProgressListener , ProgressTracker } from 'polar-shared/src/util/ProgressTracker' ; import { DocMetas } from '../metadata/DocMetas' ; import { DefaultPersistenceLayer } from './DefaultPersistenceLayer' ; import { DocMeta } from '../metadata/DocMeta' ; import { isPresent } from 'polar-shared/src/Preconditions' ; import { URLs } from "polar-shared/src/util/URLs" ; import { Logger } from "polar-shared/src/logger/Logger" ; import { BackendFileRefs } from './BackendFileRefs' ; import { IDocMeta } from "polar-shared/src/metadata/IDocMeta" ; import { BackendFileRef } from "polar-shared/src/datastore/BackendFileRef" ; import { Visibility } from "polar-shared/src/datastore/Visibility" ; const log = Logger . create ( ) ; export class PersistenceLayers { /**
     * Change visibility of the given DocMeta including setting the visibility
     * itself on the DocInfo but also setting the visibility for the individual
     * files.
     *
     */ public static async changeVisibility ( store , docMeta , visibility ) { log . info ( "Changing document visibility changed to: " , visibility ) ; const backendFileRefs = BackendFileRefs . toBackendFileRefs ( docMeta ) ; const writeFileOpts = { visibility , updateMeta : true } ; const toWriteFilePromise = async ( backendFileRef ) : Promise < void > => { await store . writeFile ( backendFileRef . backend , backendFileRef , undefined ! , writeFileOpts ) ; } ; const toWriteFilePromises = ( ) : ReadonlyArray < Promise < void > > => { return backendFileRefs . map ( current => toWriteFilePromise ( current ) ) ; } ; const toWriteDocMetaPromise = async ( ) : Promise < void > => { docMeta . docInfo . visibility = visibility ; await store . writeDocMeta ( docMeta ) ; } ; const writeFilePromises = toWriteFilePromises ( ) ; const writeDocMetaPromise = toWriteDocMetaPromise ( ) ; const promises = [ ... writeFilePromises , writeDocMetaPromise ] ; await Promise . all ( promises ) ; log . info ( "Document visibility changed to: " , visibility ) ; } public static toPersistenceLayer ( input ) { return new DefaultPersistenceLayer ( input ) ; } public static async toSyncOrigin ( datastore , ... docMetaRefs ) { const syncDocMap = await PersistenceLayers . toSyncDocMapFromDocs ( datastore , docMetaRefs ) ; return { datastore , syncDocMap } ; } public static async toSyncDocMap ( datastore , progressStateListener = NULL_FUNCTION ) { const docMetaFiles = await datastore . getDocMetaRefs ( ) ; return this . toSyncDocMapFromDocs ( datastore , docMetaFiles , progressStateListener ) ; } public static async toSyncDocMapFromDocs ( datastore , docMetaRefs , progressStateListener = NULL_FUNCTION ) { const syncDocsMap = { } ; const work = [ ] ; const asyncWorkQueue = new AsyncWorkQueue ( work ) ; const init = { total : docMetaRefs . length , id : `datastore:${ datastore . id }#toSyncDocMapFromDocs` } ; const progressTracker = new ProgressTracker ( init ) ; for ( const docMetaRef of docMetaRefs ) { work . push ( async ( ) => { let docMeta = docMetaRef . docMetaProvider ? await docMetaRef . docMetaProvider ( ) : undefined ; if ( ! docMeta ) { const data = await datastore . getDocMeta ( docMetaRef . fingerprint ) ; if ( isPresent ( data ) ) { docMeta = DocMetas . deserialize ( data ! , docMetaRef . fingerprint ) ; } } if ( isPresent ( docMeta ) ) { syncDocsMap [ docMetaRef . fingerprint ] = SyncDocs . fromDocInfo ( docMeta ! . docInfo , 'created' ) ; progressStateListener ( progressTracker . peek ( ) ) ; } else { } } ) ; } await asyncWorkQueue . execute ( ) ; progressStateListener ( progressTracker . terminate ( ) ) ; return syncDocsMap ; } /**
     * Merge both origins so that they contains the same documents. Older
     * documents are upgraded to the latest version and missing documents are
     * copied.  At the end both origins will have the union of both sets.
     */ public static async merge ( syncOrigin0 , syncOrigin1 , listener = ASYNC_NULL_FUNCTION ) { await this . transfer ( syncOrigin0 , syncOrigin1 , listener ) ; await this . transfer ( syncOrigin1 , syncOrigin0 , listener ) ; } /**
     * Make sure the latest version of the files are in both origins.
     */ public static async synchronizeOrigins ( localSyncOrigin , cloudSyncOrigin , listener = ASYNC_NULL_FUNCTION ) { log . notice ( "Transferring from local -> cloud..." ) ; const localToCloud = await PersistenceLayers . transfer ( localSyncOrigin , cloudSyncOrigin , listener , 'local-to-cloud' ) ; log . notice ( "Transferring from local -> cloud...done" , localToCloud ) ; log . notice ( "Transferring from cloud -> local..." ) ; const cloudToLocal = await PersistenceLayers . transfer ( cloudSyncOrigin , localSyncOrigin , listener , 'cloud-to-local' ) ; log . notice ( "Transferring from cloud -> local...done" , cloudToLocal ) ; } /**
     * Synchronize the source with the target so that we know they are both in
     * sync.
     */ public static async transfer ( source , target , listener = ASYNC_NULL_FUNCTION , id = 'none' ) { const result = { docMeta : { total : 0 , writes : 0 } , files : { total : 0 , writes : 0 } } ; async function handleSyncFile ( syncDoc , fileRef ) { ++ result . files . total ; const containsFile = async ( datastore , id ) : Promise < boolean > => { try { return await datastore . containsFile ( fileRef . backend , fileRef ) ; } catch ( e ) { log . error ( `Could not get file ${ fileRef . name } for doc with fingerprint: ${ syncDoc . fingerprint } from ${ id }` , fileRef , e ) ; throw e ; } } ; const targetContainsFile = await containsFile ( target . datastore , 'target' ) ; if ( ! targetContainsFile ) { const sourceContainsFile = await containsFile ( source . datastore , 'source' ) ; if ( sourceContainsFile ) { const sourceFile = source . datastore . getFile ( fileRef . backend , fileRef ) ; const blob = await URLs . toBlob ( sourceFile . url ) ; await target . datastore . writeFile ( sourceFile . backend , fileRef , blob ) ; ++ result . files . writes ; } else { log . warn ( `Both the target and source files are missing in doc ${ syncDoc . fingerprint } (${ syncDoc . title }): ` , fileRef ) ; } } } param sourceSyncDoc param targetSyncDoc async function handleSyncDoc ( sourceSyncDoc , targetSyncDoc ? ) { ++ result . docMeta . total ; /**
             * Return true if we should write doc meta because it's stale.
             */ function computeWriteDocMeta ( ) { if ( ! targetSyncDoc ) { return true ; } const cmp = UUIDs . compare ( targetSyncDoc . uuid , sourceSyncDoc . uuid ) ; return cmp < 0 ; } const targetStale = computeWriteDocMeta ( ) ; if ( targetStale ) { for ( const sourceSyncFile of sourceSyncDoc . files ) { if ( sourceSyncFile . name ) { await handleSyncFile ( sourceSyncDoc , sourceSyncFile ) ; } } const data = await source . datastore . getDocMeta ( sourceSyncDoc . fingerprint ) ; if ( data ) { await target . datastore . write ( sourceSyncDoc . fingerprint , data ! , sourceSyncDoc . docMetaFileRef . docInfo ) ; } else { log . warn ( "handleSyncDoc: No data for fingerprint: " + sourceSyncDoc . fingerprint ) ; } ++ result . docMeta . writes ; } const progress = progressTracker . incr ( ) ; const docMetaSnapshotEvent = { datastore : source . datastore . id , progress , consistency : 'committed' , docMetaMutations : [ ] } ; await listener ( docMetaSnapshotEvent ) ; } const docFileAsyncWorkQueue = new AsyncWorkQueue ( [ ] ) ; const docMetaAsyncWorkQueue = new AsyncWorkQueue ( [ ] ) ; const sourceSyncDocs = Object . values ( source . syncDocMap ) ; const progressID = `transfer:source=${ source . datastore . id },target=${ target . datastore . id }` ; const progressTracker = new ProgressTracker ( { total : sourceSyncDocs . length , id : progressID } ) ; for ( const sourceSyncDoc of sourceSyncDocs ) { const targetSyncDoc = target . syncDocMap [ sourceSyncDoc . fingerprint ] ; const handler = async ( ) => { try { await handleSyncDoc ( sourceSyncDoc , targetSyncDoc ) ; } catch ( e ) { log . error ( "Unable to sync between source and target: " , { sourceSyncDoc , targetSyncDoc } , e ) ; } } ; docMetaAsyncWorkQueue . enqueue ( handler ) ; } const docFileExecutionPromise = docFileAsyncWorkQueue . execute ( ) ; const docMetaExecutionPromise = docMetaAsyncWorkQueue . execute ( ) ; await Promise . all ( [ docFileExecutionPromise , docMetaExecutionPromise ] ) ; await listener ( { datastore : source . datastore . id , progress : progressTracker . terminate ( ) , consistency : 'committed' , docMetaMutations : [ ] } ) ; return result ; } } export interface TransferResult { docMeta : TransferMetrics ; files : TransferMetrics ; } export interface TransferMetrics { total : number ; writes : number ; } export interface SyncOrigin { readonly datastore : Datastore ; readonly syncDocMap : SyncDocMap ; }