import { each , IIterable , IIterator , iterItems , map , StringExt , toArray , toObject } from '@lumino/algorithm' ; import { BPlusTree , LinkedList } from '@lumino/collections' ; import { DisposableDelegate , IDisposable } from '@lumino/disposable' ; import { ConflatableMessage , IMessageHandler , Message , MessageLoop } from '@lumino/messaging' ; import { ISignal , Signal } from '@lumino/signaling' ; import { Record } from './record' ; import { Schema , validateSchema } from './schema' ; import { IServerAdapter } from './serveradapter' ; import { Table } from './table' ; import { createDuplexId } from './utilities' ; /**
 * A multi-user collaborative datastore.
 *
 * #### Notes
 * A store is structured in a maximally flat way using a hierarchy
 * of tables, records, and fields. Internally, the object graph is
 * synchronized among all users via CRDT algorithms.
 *
 * https://en.wikipedia.org/wiki/Conflict-free_replicated_data_type
 * https://hal.inria.fr/file/index/docid/555588/filename/techreport.pdf
 */ export class Datastore implements IDisposable , IIterable < Table < Schema > > , IMessageHandler { param options returns throws static create ( options ) { let { schemas } = options ; Private . validateSchemas ( schemas ) ; let context = { inTransaction : false , transactionId : '' , version : 0 , storeId : options . id , change : { } , patch : { } } ; let tables = new BPlusTree < Table < Schema > > ( Private . recordCmp ) ; if ( options . restoreState ) { let state = JSON . parse ( options . restoreState ) ; tables . assign ( map ( schemas , s => { return Table . recreate ( s , context , state [ s . id ] || [ ] ) ; } ) ) ; } else { tables . assign ( map ( schemas , s => { return Table . create ( s , context ) ; } ) ) ; } return new Datastore ( context , tables , options . adapter ) ; } /**
     * Dispose of the resources held by the datastore.
     */ dispose ( ) { if ( this . _disposed ) { return ; } this . _disposed = true ; Signal . clearData ( this ) ; this . _adapter = null ; } /**
     * Whether the datastore has been disposed.
     */ get isDisposed ( ) : boolean { return this . _disposed ; } /**
     * A signal emitted when changes are made to the store.
     *
     * #### Notes
     * This signal is emitted either at the end of a local mutation,
     * or after a remote mutation has been applied. The storeId can
     * be used to determine its source.
     *
     * The payload represents the set of local changes that were made
     * to bring the store to its current state.
     *
     * #### Complexity
     * `O(1)`
     */ get changed ( ) : ISignal < Datastore , Datastore . IChangedArgs > { return this . _changed ; } /**
     * The unique id of the store.
     *
     * #### Notes
     * The id is unique among all other collaborating peers.
     *
     * #### Complexity
     * `O(1)`
     */ get id ( ) : number { return this . _context . storeId ; } /**
     * Whether a transaction is currently in progress.
     *
     * #### Complexity
     * `O(1)`
     */ get inTransaction ( ) : boolean { return this . _context . inTransaction ; } /**
     * The current version of the datastore.
     *
     * #### Notes
     * This version is automatically increased for each transaction
     * to the store. However, it might not increase linearly (i.e.
     * it might make jumps).
     *
     * #### Complexity
     * `O(1)`
     */ get version ( ) : number { return this . _context . version ; } returns iter ( ) { return this . _tables . iter ( ) ; } param schema returns throws get < S extends Schema > ( schema ) { let t = this . _tables . get ( schema . id , Private . recordIdCmp ) ; if ( t === undefined ) { throw new Error ( `No table found for schema with id: ${ schema . id }` ) ; } return t as Table < S > ; } returns throws beginTransaction ( ) { let newVersion = this . _context . version + 1 ; let id = this . _transactionIdFactory ( newVersion , this . id ) ; this . _initTransaction ( id , newVersion ) ; MessageLoop . postMessage ( this , new ConflatableMessage ( 'transaction-begun' ) ) ; return id ; } /**
     * Completes a transaction.
     *
     * #### Notes
     * This completes a transaction previously started with
     * `beginTransaction`. If a change has occurred, the
     * `changed` signal will be emitted.
     */ endTransaction ( ) { this . _finalizeTransaction ( ) ; let { patch , change , storeId , transactionId , version } = this . _context ; if ( this . _adapter && ! Private . isPatchEmpty ( patch ) ) { this . _adapter . broadcast ( { id : transactionId , storeId , patch , version } ) ; } this . _cemetery [ transactionId ] = 1 ; if ( ! Private . isChangeEmpty ( this . _context . change ) ) { this . _changed . emit ( { storeId , transactionId , type : 'transaction' , change } ) ; } } /**
     * Handle a message.
     */ processMessage ( msg ) { switch ( msg . type ) { case 'transaction-begun' : if ( this . _context . inTransaction ) { console . warn ( `Automatically ending transaction (did you forget to end it?): ${ this . _context . transactionId }` ) ; this . endTransaction ( ) ; } break ; case 'queued-transaction' : this . _processQueue ( ) ; break ; default : break ; } } param transactionId returns throws undo ( transactionId ) { if ( ! this . _adapter ) { throw Error ( 'No server adapter has been set for the datastore' ) ; } if ( this . inTransaction ) { throw Error ( 'Cannot undo during a transaction' ) ; } return this . _adapter . undo ( transactionId ) ; } param transactionId returns throws redo ( transactionId ) { if ( ! this . _adapter ) { throw Error ( 'No server adapter has been set for the datastore' ) ; } if ( this . inTransaction ) { throw Error ( 'Cannot redo during a transaction' ) ; } return this . _adapter . redo ( transactionId ) ; } /**
     * The handler for broadcasting transactions to peers.
     */ get adapter ( ) : IServerAdapter | null { return this . _adapter ; } returns toString ( ) { return JSON . stringify ( toObject ( map ( this , ( table ) : [ string , Record < Schema > [ ] ] => { return [ table . schema . id , toArray ( table ) ] ; } ) ) ) ; } param id param tables private constructor ( context , tables , adapter ? , transactionIdFactory ? ) { this . _context = context ; this . _tables = tables ; this . _adapter = adapter || null ; this . _transactionIdFactory = transactionIdFactory || createDuplexId ; if ( this . _adapter ) { this . _adapter . onRemoteTransaction = this . _onRemoteTransaction . bind ( this ) ; this . _adapter . onUndo = this . _onUndo . bind ( this ) ; this . _adapter . onRedo = this . _onRedo . bind ( this ) ; } } /**
     * Handle a transaction from the server adapter.
     */ private _onRemoteTransaction ( transaction ) { this . _processTransaction ( transaction , 'transaction' ) ; } /**
     * Handle an undo from the server adapter.
     */ private _onUndo ( transaction ) { this . _processTransaction ( transaction , 'undo' ) ; } /**
     * Handle a redo from the server adapter.
     */ private _onRedo ( transaction ) { this . _processTransaction ( transaction , 'redo' ) ; } param transactionApplication throws private _processTransaction ( transaction , type ) { let { storeId , patch } = transaction ; try { this . _initTransaction ( transaction . id , Math . max ( this . _context . version , transaction . version ) ) ; } catch ( e ) { this . _queueTransaction ( transaction , type ) ; return ; } let change = { } ; try { each ( iterItems ( patch ) , ( [ schemaId , tablePatch ] ) => { let table = this . _tables . get ( schemaId , Private . recordIdCmp ) ; if ( table === undefined ) { console . warn ( `Missing table for schema id '${ schemaId }' in transaction '${ transaction . id }'` ) ; this . _finalizeTransaction ( ) ; return ; } if ( type === 'transaction' || type === 'redo' ) { let count = this . _cemetery [ transaction . id ] ; if ( count === undefined ) { this . _cemetery [ transaction . id ] = 1 ; change [ schemaId ] = Table . patch ( table , tablePatch ) ; return ; } this . _cemetery [ transaction . id ] = count + 1 ; if ( this . _cemetery [ transaction . id ] === 1 ) { change [ schemaId ] = Table . patch ( table , tablePatch ) ; return ; } } else { let count = this . _cemetery [ transaction . id ] ; if ( count === undefined ) { this . _cemetery [ transaction . id ] = - 1 ; return ; } this . _cemetery [ transaction . id ] = count - 1 ; if ( this . _cemetery [ transaction . id ] === 0 ) { change [ schemaId ] = Table . unpatch ( table , tablePatch ) ; } } } ) ; } finally { this . _finalizeTransaction ( ) ; } if ( ! Private . isChangeEmpty ( change ) ) { this . _changed . emit ( { storeId , transactionId : transaction . id , type , change } ) ; } } param transaction private _queueTransaction ( transaction , type ) { this . _transactionQueue . addLast ( [ transaction , type ] ) ; MessageLoop . postMessage ( this , new ConflatableMessage ( 'queued-transaction' ) ) ; } /**
     * Process all transactions currently queued.
     */ private _processQueue ( ) { let queue = this . _transactionQueue ; if ( queue . isEmpty ) { return ; } let sentinel = { } ; queue . addLast ( sentinel as any ) ; while ( true ) { let [ transaction , type ] = queue . removeFirst ( ) ! ; if ( transaction === sentinel ) { return ; } this . _processTransaction ( transaction , type ) ; } } param id param newVersion throws private _initTransaction ( id , newVersion ) { let context = this . _context as Private . MutableContext ; if ( context . inTransaction ) { throw new Error ( `Already in a transaction: ${ this . _context . transactionId }` ) ; } context . inTransaction = true ; context . change = { } ; context . patch = { } ; context . transactionId = id ; context . version = newVersion ; } throws private _finalizeTransaction ( ) { let context = this . _context as Private . MutableContext ; if ( ! context . inTransaction ) { throw new Error ( 'No transaction in progress.' ) ; } context . inTransaction = false ; } private _adapter : IServerAdapter | null ; private _cemetery : { [ id ] : number ; } = { } ; private _disposed = false ; private _tables : BPlusTree < Table < Schema > > ; private _context : Datastore . Context ; private _changed = new Signal < Datastore , Datastore . IChangedArgs > ( this ) ; private _transactionIdFactory : Datastore . TransactionIdFactory ; private _transactionQueue = new LinkedList < [ Datastore . Transaction , Datastore . TransactionType ] > ( ) ; } /**
 * The namespace for the `Datastore` class statics.
 */ export namespace Datastore { /**
     * A type alias for kinds of transactions.
     */ export type TransactionType = 'transaction' | 'undo' | 'redo' ; /**
     * An options object for initializing a datastore.
     */ export interface IOptions { /**
         * The unique id of the datastore.
         */ id : number ; /**
         * The table schemas of the datastore.
         */ schemas : ReadonlyArray < Schema > ; /**
         * An optional handler for broadcasting transactions to peers.
         */ adapter ? : IServerAdapter ; /**
         * An optional transaction id factory to override the default.
         */ transactionIdFactory ? : TransactionIdFactory ; /**
         * Initialize the state to a previously serialized one.
         */ restoreState ? : string ; } /**
     * The arguments object for the store `changed` signal.
     */ export interface IChangedArgs { /**
         * Whether the change was generated by transaction, undo, or redo.
         */ readonly type : TransactionType ; /**
         * The transaction id associated with the change.
         */ readonly transactionId : string ; /**
         * The id of the store responsible for the change.
         */ readonly storeId : number ; /**
         * A mapping of schema id to table change set.
         */ readonly change : Change ; } /**
     * A type alias for a store change.
     */ export type Change = { readonly [ schemaId ] : Table . Change < Schema > ; } ; /**
     * A type alias for a store patch.
     */ export type Patch = { readonly [ schemaId ] : Table . Patch < Schema > ; } ; internal export type MutableChange = { [ schemaId ] : Table . MutableChange < Schema > ; } ; internal export type MutablePatch = { [ schemaId ] : Table . MutablePatch < Schema > ; } ; /**
     * An object representing a datastore transaction.
     */ export type Transaction = { /**
         * The id of the transaction.
         */ readonly id : string ; /**
         * The id of the store responsible for the transaction.
         */ readonly storeId : number ; /**
         * The patch data of the transaction.
         */ readonly patch : Patch ; /**
         * The version of the source datastore.
         */ readonly version : number ; } ; internal export type Context = Readonly < Private . MutableContext > ; /**
     * A factory function for generating a unique transaction id.
     */ export type TransactionIdFactory = ( version : number , storeId : number ) => string ; param datastore param update returns export function withTransaction ( datastore , update ) { let id = '' ; if ( ! datastore . inTransaction ) { id = datastore . beginTransaction ( ) ; } try { update ( id ) ; } finally { if ( id ) { datastore . endTransaction ( ) ; } } return id ; } /**
     * A base type for describing the location of data in a datastore,
     * to be consumed by some object. The only requirement is that it
     * has a datastore object. Objects extending from this will, in general,
     * have some combination of table, record, and field locations.
     */ export type DataLocation = { /**
         * The datastore in which the data is contained.
         */ datastore : Datastore ; } ; /**
     * An interface for referring to a specific table in a datastore.
     */ export type TableLocation < S extends Schema > = { /**
         * The schema in question. This schema must exist in the datastore,
         * or an error may result in its usage.
         */ schema : S ; } ; /**
     * An interface for referring to a specific record in a datastore.
     */ export type RecordLocation < S extends Schema > = TableLocation < S > & { /**
         * The record in question.
         */ record : string ; } ; /**
     * An interface for referring to a specific field in a datastore.
     *
     * #### Notes
     * The field must exist in the schema.
     */ export type FieldLocation < S extends Schema , F extends keyof S [ 'fields' ] > = RecordLocation < S > & { /**
         * The field in question.
         */ field : F ; } ; param datastore param loc returns export function getTable < S extends Schema > ( datastore , loc ) { return datastore . get ( loc . schema ) ; } param datastore param loc returns export function getRecord < S extends Schema > ( datastore , loc ) { return datastore . get ( loc . schema ) . get ( loc . record ) ; } param datastore param loc returns export function getField < S extends Schema , F extends keyof S [ 'fields' ] > ( datastore , loc ) { const record = datastore . get ( loc . schema ) . get ( loc . record ) ; if ( ! record ) { throw Error ( `The record ${ loc . record } could not be found` ) ; } return record [ loc . field ] ; } param datastore param loc param update export function updateTable < S extends Schema > ( datastore , loc , update ) { let table = datastore . get ( loc . schema ) ; table . update ( update ) ; } param datastore param loc param update export function updateRecord < S extends Schema > ( datastore , loc , update ) { let table = datastore . get ( loc . schema ) ; table . update ( { [ loc . record ] : update } ) ; } param datastore param loc param update export function updateField < S extends Schema , F extends keyof S [ 'fields' ] > ( datastore , loc , update ) { let table = datastore . get ( loc . schema ) ; table . update ( { [ loc . record ] : { [ loc . field ] : update } as Record . Update < S > } ) ; } param datastore param loc param slot returns export function listenTable < S extends Schema > ( datastore , loc , slot , thisArg ? ) { const wrapper = ( source , args ) => { if ( ! args . change [ loc . schema . id ] ) { return ; } const tc = args . change [ loc . schema . id ] ! as Table . Change < S > ; slot . bind ( thisArg ) ( source , tc ) ; } ; datastore . changed . connect ( wrapper ) ; return new DisposableDelegate ( ( ) => { datastore . changed . disconnect ( wrapper ) ; } ) ; } param datastore param loc param slot returns export function listenRecord < S extends Schema > ( datastore , loc , slot , thisArg ? ) { const wrapper = ( source , args ) => { if ( ! args . change [ loc . schema . id ] || ! args . change [ loc . schema . id ] [ loc . record ] ) { return ; } const tc = args . change [ loc . schema . id ] ! as Table . Change < S > ; slot . bind ( thisArg ) ( source , tc [ loc . record ] ) ; } ; datastore . changed . connect ( wrapper ) ; return new DisposableDelegate ( ( ) => { datastore . changed . disconnect ( wrapper ) ; } ) ; } param datastore param loc param slot returns export function listenField < S extends Schema , F extends keyof S [ 'fields' ] > ( datastore , loc , slot , thisArg ? ) { const wrapper = ( source , args ) => { if ( ! args . change [ loc . schema . id ] || ! args . change [ loc . schema . id ] [ loc . record ] || ! args . change [ loc . schema . id ] [ loc . record ] [ loc . field as string ] ) { return ; } const tc = args . change [ loc . schema . id ] ! as Table . Change < S > ; slot . bind ( thisArg ) ( source , tc [ loc . record ] [ loc . field ] ) ; } ; datastore . changed . connect ( wrapper ) ; return new DisposableDelegate ( ( ) => { datastore . changed . disconnect ( wrapper ) ; } ) ; } } namespace Private { /**
     * Validates all schemas, and throws an error if any are invalid.
     */ export function validateSchemas ( schemas ) { let errors = [ ] ; for ( let s of schemas ) { let err = validateSchema ( s ) ; if ( err . length ) { errors . push ( `Schema '${ s . id }' validation failed: \n${ err . join ( '\n' ) }` ) ; } } if ( errors . length ) { throw new Error ( errors . join ( '\n\n' ) ) ; } } /**
     * A three-way record comparison function.
     */ export function recordCmp < S extends Schema > ( a , b ) { return StringExt . cmp ( a . schema . id , b . schema . id ) ; } /**
     * A three-way record id comparison function.
     */ export function recordIdCmp < S extends Schema > ( table , id ) { return StringExt . cmp ( table . schema . id , id ) ; } export type MutableContext = { /**
         * Whether the datastore currently in a transaction.
         */ inTransaction : boolean ; /**
         * The id of the current transaction.
         */ transactionId : string ; /**
         * The current version of the datastore.
         */ version : number ; /**
         * The unique id of the datastore.
         */ storeId : number ; /**
         * The current change object of the transaction.
         */ change : Datastore . MutableChange ; /**
         * The current patch object of the transaction.
         */ patch : Datastore . MutablePatch ; } ; /**
     * Checks if a patch is empty.
     */ export function isPatchEmpty ( patch ) { return Object . keys ( patch ) . length === 0 ; } /**
     * Checks if a change is empty.
     */ export function isChangeEmpty ( change ) { return Object . keys ( change ) . length === 0 ; } }