import { AttachmentIngestEvent , Event , EventID , EventType , } from "@withorbit/core" ; import { OrbitStore } from "@withorbit/store-shared" ; import { OrbitStoreSyncAdapter } from "./orbitStoreSyncAdapter" ; import { SyncAdapter } from "./syncAdapter" ; function log ( ... args ) { const [ first , ... rest ] = args ; console . log ( `[Sync] ${ Date . now ( ) / 1000.0 } ${ first }` , ... rest ) ; } export interface SyncCache { sourceEventIDs : Set < EventID > ; destinationEventIDs : Set < EventID > ; } export async function syncOrbitStore ( { source , destination , sendBatchSize = 200 , receiveBatchSize = 1000 , } ) { const syncCache = { sourceEventIDs : new Set ( ) , destinationEventIDs : new Set ( ) , } ; const sourceSyncInterface = new OrbitStoreSyncAdapter ( source , "local" ) ; const latestSentEventIDKey = `__sync_${ destination . id }_latestEventIDSent` ; const latestReceivedEventIDKey = `__sync_${ destination . id }_latestEventIDReceived` ; const latestEventIDs = await source . database . getMetadataValues ( [ latestSentEventIDKey , latestReceivedEventIDKey , ] ) ; log ( "*** Sending new events" ) ; const sendArgs = { source : sourceSyncInterface , destination : destination , latestSentEventID : ( latestEventIDs . get ( latestSentEventIDKey ) as EventID ) ?? null , setLatestSentEventID : ( eventID ) => source . database . setMetadataValues ( new Map ( [ [ latestSentEventIDKey , eventID ] ] ) ) , batchSize : sendBatchSize , syncCache , } ; const { sentEventCount : sentEventCount1 , latestSentEventID } = await sendNewEvents ( sendArgs ) ; log ( "*** Receiving new events" ) ; const { sentEventCount : receivedEventCount } = await sendNewEvents ( { source : destination , destination : sourceSyncInterface , latestSentEventID : ( latestEventIDs . get ( latestReceivedEventIDKey ) as EventID ) ?? null , setLatestSentEventID : ( eventID ) => source . database . setMetadataValues ( new Map ( [ [ latestReceivedEventIDKey , eventID ] ] ) ) , batchSize : receiveBatchSize , syncCache : { sourceEventIDs : syncCache . destinationEventIDs , destinationEventIDs : syncCache . sourceEventIDs , } , } ) ; log ( "*** Reconciling received events" ) ; const { sentEventCount : sentEventCount2 } = await sendNewEvents ( { ... sendArgs , latestSentEventID , } ) ; return { sentEventCount : sentEventCount1 + sentEventCount2 , receivedEventCount , } ; } async function sendNewEvents ( { source , destination , latestSentEventID , setLatestSentEventID , batchSize , syncCache , } ) { let afterEventID = latestSentEventID ; let sentEventCount = 0 ; while ( true ) { log ( "Requesting next batch of events" ) ; const allEvents = await source . listEvents ( afterEventID , batchSize ) ; allEvents . forEach ( ( event ) => syncCache . sourceEventIDs . add ( event . id ) ) ; const newEvents = allEvents . filter ( ( { id } ) => ! syncCache . destinationEventIDs . has ( id ) ) ; log ( `Received ${ allEvents . length } events (${ newEvents . length } new)` ) ; if ( newEvents . length > 0 ) { const attachmentIngestEvents = newEvents . filter ( ( event ) : event is AttachmentIngestEvent => event . type === EventType . AttachmentIngest ) ; if ( attachmentIngestEvents . length > 0 ) { log ( `Transferring ${ attachmentIngestEvents . length } attachments` ) ; } for ( const { entityID , mimeType } of attachmentIngestEvents ) { const contents = await source . getAttachmentContents ( entityID ) ; await destination . putAttachment ( contents , entityID , mimeType ) ; } if ( attachmentIngestEvents . length > 0 ) { log ( `Finished transferring attachments` ) ; } log ( "Putting events at destination" ) ; await destination . putEvents ( newEvents ) ; sentEventCount += newEvents . length ; log ( `Done. ${ sentEventCount } total events transferred.` ) ; newEvents . forEach ( ( event ) => syncCache . destinationEventIDs . add ( event . id ) ) ; } if ( allEvents . length > 0 ) { afterEventID = allEvents [ allEvents . length - 1 ] . id ; await setLatestSentEventID ( afterEventID ) ; } else { break ; } } return { sentEventCount , latestSentEventID : afterEventID } ; }