import { assert , combine , createEntityTooLargeError , exponentialBackOffStrategy , first , last , noop , WorkLoop , workLoop , } from '@syncot/util' ; import { Duplex } from 'readable-stream' ; import { createNotFoundError , isAlreadyExistsError } from './error' ; import { createBaseOperation , Operation } from './operation' ; import { PubSub } from './pubSub' ; import { Schema , SchemaHash } from './schema' ; import { createBaseSnapshot , Snapshot } from './snapshot' ; import { ContentStore } from './store' ; import { OperationStream } from './stream' ; import { ContentType } from './type' ; import { maxOperationSize , maxSchemaSize , maxSnapshotSize , minVersion , } from './limits' ; /**
 * Content and schema management interface.
 */ export interface ContentBackend { param schema registerSchema ( schema ) : Promise < void > ; param hash returns getSchema ( hash ) : Promise < Schema | null > ; param type param id param version returns getSnapshot ( type , id , version ) : Promise < Snapshot > ; param operation submitOperation ( operation ) : Promise < void > ; param type param id param versionStart param versionEnd returns streamOperations ( type , id , versionStart , versionEnd ) : Promise < Duplex > ; } /**
 * The options expected by `createContentBackend`.
 */ export interface CreateContentBackendOptions { /**
     * The ContentStore to use for storing content.
     */ contentStore : ContentStore ; /**
     * The PubSub to use for content-related notifications.
     */ pubSub : PubSub ; /**
     * The content types to support.
     */ contentTypes : { [ key ] : ContentType ; } ; /**
     * Defines the min number of milliseconds for which to cache snapshots and operations in memory.
     * Defaults to 10000.
     */ cacheTTL ? : number ; /**
     * The max number of operations to cache.
     * Defaults to 50.
     */ cacheLimit ? : number ; /**
     * Indicates if the specified snapshot should be stored in `contentStore`.
     * Defaults to: `(snapshot) => snapshot.version % 1000 === 0`.
     */ shouldStoreSnapshot ? : ( snapshot : Snapshot ) => boolean ; /**
     * A function called when a non-critical error occurs.
     * Defaults to: `(error) => console.warn(error)`.
     *
     * Warnings are emitted in the following situations:
     * - saving a snapshot fails.
     */ onWarning ? : ( error : Error ) => void ; } /**
 * Creates a new ContentBackend instance.
 */ export function createContentBackend ( { contentStore , pubSub , contentTypes , cacheTTL = 10000 , cacheLimit = 50 , shouldStoreSnapshot = ( snapshot ) => snapshot . version % 1000 === 0 , onWarning = console . warn . bind ( console ) , } ) { return new Backend ( contentStore , pubSub , contentTypes , cacheTTL , cacheLimit , shouldStoreSnapshot , onWarning ) ; } const { hasOwnProperty } = Object . prototype ; /**
 * The cached snapshot and operations.
 */ interface Cache { /**
     * The cached snapshot.
     */ snapshot : Snapshot ; /**
     * The cached operations which can be applied to the cached snapshot.
     */ operations : Operation [ ] ; /**
     * The time at which this cache item should expire.
     */ expireAt : number ; } class Backend implements ContentBackend { /**
     * The open operation streams, indexed by `combine(type, id)`.
     */ private readonly streams : Map < string , OperationStream [ ] > = new Map ( ) ; /**
     * A set of `combine(type, id)` values indicating which operation streams need updating.
     */ private readonly streamsToUpdate : Set < string > = new Set ( ) ; /**
     * The cache of snapshots and operations, indexed by `combine(type, id)`.
     */ private readonly cache : Map < string , Cache > = new Map ( ) ; /**
     * A Set of `combine(type, id)` values indicating which cache items should expire.
     * The insertion orders matches the order in which the items should expire.
     */ private readonly expiringCacheItems : Set < string > = new Set ( ) ; private triggerStreamsUpdate = noop ; public constructor ( private readonly contentStore , private readonly pubSub , private readonly contentTypes , private readonly cacheTTL , private readonly cacheLimit , private readonly shouldStoreSnapshot , private readonly onWarning ) { assert ( this . contentStore && typeof this . contentStore === 'object' , 'Argument "contentStore" must be a ContentStore instance.' ) ; assert ( this . pubSub && typeof this . pubSub === 'object' , 'Argument "pubSub" must be a PubSub instance.' ) ; assert ( this . contentTypes && typeof this . contentTypes === 'object' , 'Argument "contentTypes" must be an object.' ) ; assert ( Number . isInteger ( cacheTTL ) && cacheTTL >= 0 , 'Argument "cacheTTL" must be a non-negative integer or undefined.' ) ; assert ( Number . isInteger ( cacheLimit ) && cacheLimit >= 0 , 'Argument "cacheLimit" must be a non-negative integer or undefined.' ) ; assert ( typeof shouldStoreSnapshot === 'function' , 'Argument "shouldStoreSnapshot" must be a function or undefined.' ) ; assert ( typeof onWarning === 'function' , 'Argument "onWarning" must be a function or undefined.' ) ; workLoop ( ( triggerStreamsUpdate ) => { this . triggerStreamsUpdate = triggerStreamsUpdate ; return new StreamUpdater ( this . streamsToUpdate , this . getStreams , this . loadOperations , triggerStreamsUpdate ) ; } ) ; } public async registerSchema ( schema ) { this . checkSchemaSize ( schema ) ; const contentType = this . getContentType ( schema . type ) ; contentType . validateSchema ( schema ) ; try { return await this . contentStore . storeSchema ( schema ) ; } catch ( error ) { if ( isAlreadyExistsError ( error ) ) return ; throw error ; } } public async getSchema ( hash ) { return await this . contentStore . loadSchema ( hash ) ; } public async getSnapshot ( type , id , version ) { return this . loadSnapshot ( type , id , version ) ; } public async submitOperation ( operation ) { this . checkOperationSize ( operation ) ; const { type , id , version } = operation ; const key = combine ( type , id ) ; const contentType = this . getContentType ( operation . type ) ; await this . ensureSchema ( contentType , operation . schema ) ; let snapshot = await this . loadSnapshot ( type , id , version - 1 ) ; assert ( version === snapshot . version + 1 , 'operation.version out of sequence.' ) ; snapshot = contentType . apply ( snapshot , operation ) ; this . checkSnapshotSize ( snapshot ) ; try { await this . contentStore . storeOperation ( operation ) ; this . cacheOperations ( key , [ operation ] , contentType ) ; this . pubSub . publish ( combine ( 'operation' , type , id ) , operation ) ; this . storeSnapshotIfNecessary ( snapshot ) ; } catch ( error ) { if ( isAlreadyExistsError ( error ) ) this . updateStreams ( key ) ; throw error ; } } public async streamOperations ( type , id , versionStart , versionEnd ) { const key = combine ( type , id ) ; const stream = new OperationStream ( type , id , versionStart , versionEnd ) ; { const streams = this . streams . get ( key ) ; if ( streams ) { this . streams . set ( key , streams . concat ( stream ) ) ; } else { this . streams . set ( key , [ stream ] ) ; this . pubSub . subscribe ( combine ( 'operation' , type , id ) , this . onOperation ) ; this . touch ( key ) ; } } stream . on ( 'close' , ( ) => { const streams = this . streams . get ( key ) ; if ( ! streams ) return ; if ( streams . length === 1 && streams [ 0 ] === stream ) { this . streams . delete ( key ) ; this . pubSub . unsubscribe ( combine ( 'operation' , type , id ) , this . onOperation ) ; this . touch ( key ) ; } else { this . streams . set ( key , streams . filter ( ( s ) => s !== stream ) ) ; } } ) ; this . updateStreams ( key ) ; return stream ; } private getStreams = ( key ) => { return this . streams . get ( key ) ; } ; private updateStreams ( key ) { this . streamsToUpdate . add ( key ) ; this . triggerStreamsUpdate ( ) ; } private onOperation = ( operation ) : void => { const { type , id , version } = operation ; const key = combine ( type , id ) ; const streams = this . streams . get ( key ) ; if ( ! streams ) return ; for ( const stream of streams ) { if ( stream . versionNext === version ) { stream . pushOperation ( operation ) ; } else { this . updateStreams ( key ) ; } } } ; private getContentType ( type ) { if ( ! hasOwnProperty . call ( this . contentTypes , type ) ) throw new TypeError ( `Unsupported document type: ${ type }.` ) ; return this . contentTypes [ type ] ; } private async ensureSchema ( contentType , schemaHash ) { if ( ! contentType . hasSchema ( schemaHash ) ) { const schema = await this . contentStore . loadSchema ( schemaHash ) ; if ( ! schema ) throw createNotFoundError ( 'Schema' ) ; contentType . registerSchema ( schema ) ; } } private async storeSnapshotIfNecessary ( snapshot ) { if ( ! this . shouldStoreSnapshot ( snapshot ) ) return ; try { await this . contentStore . storeSnapshot ( snapshot ) ; } catch ( error ) { if ( isAlreadyExistsError ( error ) ) return ; queueMicrotask ( ( ) => this . onWarning ( error ) ) ; } } private async loadSnapshot ( type , id , version ) { let snapshot ; const contentType = this . getContentType ( type ) ; const key = combine ( type , id ) ; const cache = this . cache . get ( key ) ; if ( cache && cache . snapshot . version <= version ) { snapshot = cache . snapshot ; for ( const operation of cache . operations ) { if ( operation . version > version ) break ; snapshot = contentType . apply ( snapshot , operation ) ; this . storeSnapshotIfNecessary ( snapshot ) ; } this . touch ( key ) ; } else { snapshot = createBaseSnapshot ( type , id ) ; } if ( snapshot . version === version ) return snapshot ; if ( snapshot . version === 0 ) { snapshot = await this . contentStore . loadSnapshot ( type , id , version ) ; } if ( snapshot . version < version ) { const operations = await this . loadOperations ( type , id , snapshot . version + 1 , version + 1 ) ; for ( const operation of operations ) { snapshot = contentType . apply ( snapshot , operation ) ; this . storeSnapshotIfNecessary ( snapshot ) ; } } if ( snapshot . version < version ) this . cacheSnapshot ( key , snapshot ) ; return snapshot ; } private loadOperations = async ( type , id , versionStart , versionEnd ) : Promise < Operation [ ] > => { const operations = [ ] ; if ( versionEnd <= versionStart ) return operations ; const contentType = this . getContentType ( type ) ; const key = combine ( type , id ) ; const cache = this . cache . get ( key ) ; let versionNext = versionStart ; if ( versionStart === minVersion ) { operations . push ( createBaseOperation ( type , id ) ) ; versionNext ++ ; } if ( cache && cache . operations . length > 0 ) { const cacheVersionStart = first ( cache . operations ) ! . version ; const cacheVersionEnd = last ( cache . operations ) ! . version + 1 ; if ( versionNext >= cacheVersionStart && versionNext < cacheVersionEnd ) { const from = versionNext - cacheVersionStart ; const to = Math . min ( cache . operations . length , versionEnd - cacheVersionStart ) ; for ( let i = from ; i < to ; i ++ ) { operations . push ( cache . operations [ i ] ) ; } versionNext += to - from ; this . touch ( key ) ; } } if ( versionNext < versionEnd ) { const loadedOperations = await this . contentStore . loadOperations ( type , id , versionNext , versionEnd ) ; for ( const loadedOperation of loadedOperations ) { await this . ensureSchema ( contentType , loadedOperation . schema ) ; operations . push ( loadedOperation ) ; } this . cacheOperations ( key , operations , contentType ) ; } return operations ; } ; private touch ( key ) { this . expiringCacheItems . delete ( key ) ; if ( this . streams . has ( key ) ) return ; const cache = this . cache . get ( key ) ; if ( ! cache ) return ; cache . expireAt = Date . now ( ) + this . cacheTTL ; this . expiringCacheItems . add ( key ) ; this . scheduleExpireCache ( ) ; } private expireCacheHandle : NodeJS . Timeout | undefined ; private expireCache = ( ) : void => { const now = Date . now ( ) ; for ( const key of this . expiringCacheItems ) { const cache = this . cache . get ( key ) ; if ( cache && cache . expireAt >= now ) break ; this . cache . delete ( key ) ; this . expiringCacheItems . delete ( key ) ; } if ( this . expiringCacheItems . size === 0 ) this . cancelExpireCache ( ) ; } ; private scheduleExpireCache ( ) { if ( this . expireCacheHandle != null ) return ; this . expireCacheHandle = setInterval ( this . expireCache , 1000 ) ; if ( typeof this . expireCacheHandle . unref === 'function' ) this . expireCacheHandle . unref ( ) ; } private cancelExpireCache ( ) { if ( this . expireCacheHandle == null ) return ; clearInterval ( this . expireCacheHandle ) ; this . expireCacheHandle = undefined ; } private cacheSnapshot ( key , snapshot ) { if ( ! this . cache . has ( key ) ) { this . cache . set ( key , { snapshot , operations : [ ] , expireAt : 0 , } ) ; this . touch ( key ) ; } } private cacheOperations ( key , newOperations , contentType ) { const cache = this . cache . get ( key ) ; if ( ! cache ) return ; const { operations } = cache ; if ( newOperations . length > 0 ) { let versionNext = operations . length > 0 ? last ( operations ) ! . version + 1 : cache . snapshot . version + 1 ; if ( first ( newOperations ) ! . version <= versionNext && versionNext <= last ( newOperations ) ! . version ) { for ( const operation of newOperations ) { if ( operation . version === versionNext ) { operations . push ( operation ) ; versionNext ++ ; } } } } const minCacheTime = Date . now ( ) - this . cacheTTL ; const maxCacheLength = this . cacheLimit ; while ( operations . length > 0 ) { const operation = first ( operations ) ! ; if ( operation . meta == null || operation . meta . time == null || operation . meta . time < minCacheTime || operations . length > maxCacheLength ) { cache . snapshot = contentType . apply ( cache . snapshot , operation ) ; operations . shift ( ) ; } else break ; } this . touch ( key ) ; } private checkSchemaSize ( schema ) { if ( JSON . stringify ( schema ) . length > maxSchemaSize ) throw createEntityTooLargeError ( 'Schema' ) ; } private checkOperationSize ( operation ) { if ( JSON . stringify ( operation ) . length > maxOperationSize ) throw createEntityTooLargeError ( 'Operation' ) ; } private checkSnapshotSize ( snapshot ) { if ( JSON . stringify ( snapshot ) . length > maxSnapshotSize ) throw createEntityTooLargeError ( 'Snapshot' ) ; } } class StreamUpdater implements WorkLoop { private readonly loadLimit = 100 ; public readonly onError = noop ; public readonly retryDelay = exponentialBackOffStrategy ( { minDelay : 1000 , maxDelay : 10000 , delayFactor : 1.5 , } ) ; public constructor ( public jobs , public getStreams , public loadOperations , private notify ) { } public async work ( ) { const promises = Array . from ( this . jobs ) . map ( this . doJob ) ; const results = await Promise . allSettled ( promises ) ; const hasErrors = results . some ( isResultRejected ) ; if ( hasErrors ) throw new Error ( ) ; } private doJob = async ( job ) : Promise < void > => { this . jobs . delete ( job ) ; const maybeStreams = this . getStreams ( job ) ; if ( ! maybeStreams ) return ; const streams = maybeStreams . filter ( needsUpdate ) . sort ( orderStreams ) ; if ( streams . length === 0 ) return ; const { type , id , versionNext } = first ( streams ) ! ; let { versionEnd } = first ( streams ) ! ; for ( let i = 1 ; i < streams . length ; i ++ ) { const stream = streams [ i ] ; if ( stream . versionNext <= versionEnd ) { versionEnd = Math . max ( versionEnd , stream . versionEnd ) ; } } versionEnd = Math . min ( versionEnd , versionNext + this . loadLimit ) ; try { const operations = await this . loadOperations ( type , id , versionNext , versionEnd ) ; for ( const stream of streams ) { if ( needsUpdate ( stream ) && stream . versionNext < versionEnd ) { for ( const operation of operations ) { if ( stream . versionNext === operation . version ) { stream . pushOperation ( operation ) ; } } } } if ( operations . length > 0 && last ( operations ) ! . version === versionEnd - 1 ) { this . jobs . add ( job ) ; this . notify ( ) ; } } catch ( error ) { const stream = first ( streams ) ! ; if ( needsUpdate ( stream ) ) stream . emit ( 'error' , error ) ; this . jobs . add ( job ) ; throw error ; } } ; } function orderStreams ( stream1 , stream2 ) { return ( stream1 . versionNext - stream2 . versionNext || stream2 . versionEnd - stream1 . versionEnd ) ; } function isResultRejected ( result ) { return result . status === 'rejected' ; } function needsUpdate ( stream ) { return ! stream . destroyed && stream . versionNext < stream . versionEnd ; }