import BaseAdapter , { StorageProxy } from "./adapters/base" ; import IDB from "./adapters/IDB" ; import { waterfall , deepEqual } from "./utils" ; import { v4 as uuid4 } from "uuid" ; import { omitKeys , RE_RECORD_ID } from "./utils" ; import KintoBase from "./KintoBase" ; import { AggregateResponse , Collection as KintoCollection } from "kinto-http" ; import { KintoRepresentation , IdSchema , RemoteTransformer , AvailableHook , Hooks , UndefiendKintoRepresentation , UpdateRepresentation , WithOptional , RecordStatus , SyncResult , KintoError , Conflict , Change , ConflictsChange , Emitter , CollectionSyncOptions , } from "./types" ; const RECORD_FIELDS_TO_CLEAN = [ "_status" ] ; const AVAILABLE_HOOKS = [ "incoming-changes" ] ; const IMPORT_CHUNK_SIZE = 200 ; param { Object } a param { Object } b param { Array } localFields return { boolean } export function recordsEqual ( a , b , localFields = [ ] ) { const fieldsToClean = [ ... RECORD_FIELDS_TO_CLEAN , "last_modified" , ... localFields , ] ; const cleanLocal = ( r ) => omitKeys ( r , fieldsToClean ) ; return deepEqual ( cleanLocal ( a ) , cleanLocal ( b ) ) ; } /**
 * Synchronization result object.
 */ export class SyncResultObject { /**
     * Public constructor.
     */ public lastModified ? : number | null ; private _lists : SyncResult ; private _cached : Partial < SyncResult > ; constructor ( ) { type { Boolean } this . lastModified = null ; this . _lists = { errors : [ ] , created : [ ] , updated : [ ] , deleted : [ ] , published : [ ] , conflicts : [ ] , skipped : [ ] , resolved : [ ] , void : [ ] , } ; this . _cached = { } ; } param { String } type param { Array } entries return { SyncResultObject } add < K extends keyof SyncResult > ( type , entries ) { if ( ! Array . isArray ( this . _lists [ type ] ) ) { console . warn ( `Unknown type "${ type }"` ) ; return this ; } if ( ! Array . isArray ( entries ) ) { entries = [ entries ] ; } this . _lists [ type ] = [ ... this . _lists [ type ] , ... ( entries as SyncResult [ K ] ) ] ; delete this . _cached [ type ] ; return this ; } get ok ( ) { return this . errors . length + this . conflicts . length === 0 ; } get errors ( ) { return this . _lists [ "errors" ] ; } get conflicts ( ) { return this . _lists [ "conflicts" ] ; } get skipped ( ) { return this . _deduplicate ( "skipped" ) ; } get resolved ( ) { return this . _deduplicate ( "resolved" ) ; } get created ( ) { return this . _deduplicate ( "created" ) ; } get updated ( ) { return this . _deduplicate ( "updated" ) ; } get deleted ( ) { return this . _deduplicate ( "deleted" ) ; } get published ( ) { return this . _deduplicate ( "published" ) ; } private _deduplicate < K extends keyof SyncResult > ( list ) { if ( ! ( list in this . _cached ) ) { const recordsWithoutId = new Set ( ) ; const recordsById = new Map ( ) ; this . _lists [ list ] . forEach ( ( record ) => { if ( ! record . id ) { recordsWithoutId . add ( record ) ; } else { recordsById . set ( record . id , record ) ; } } ) ; this . _cached [ list ] = Array . from ( recordsById . values ( ) ) . concat ( Array . from ( recordsWithoutId ) ) ; } return ( this . _cached as NonNullable < SyncResult > ) [ list ] ; } param { String } type return { SyncResultObject } reset ( type ) { this . _lists [ type ] = [ ] ; delete this . _cached [ type ] ; return this ; } private toObject ( ) { return { ok : this . ok , lastModified : this . lastModified , errors : this . errors , created : this . created , updated : this . updated , deleted : this . deleted , skipped : this . skipped , published : this . published , conflicts : this . conflicts , resolved : this . resolved , } ; } } export class ServerWasFlushedError extends Error { public clientTimestamp : number ; public serverTimestamp ? : number ; public message ! : string ; constructor ( clientTimestamp , serverTimestamp , message ) { super ( message ) ; if ( Error . captureStackTrace ) { Error . captureStackTrace ( this , ServerWasFlushedError ) ; } this . clientTimestamp = clientTimestamp ; this . serverTimestamp = serverTimestamp ; } } function createUUIDSchema ( ) { return { generate ( ) { return uuid4 ( ) ; } , validate ( id ) { return typeof id === "string" && RE_RECORD_ID . test ( id ) ; } , } ; } example export function createKeyValueStoreIdSchema ( ) { return { generate ( ) { throw new Error ( "createKeyValueStoreIdSchema() does not generate an id" ) ; } , validate ( ) { return true ; } , } ; } function markStatus < T extends { _status ? : string ; } > ( record , status ) { return { ... record , _status : status } ; } function markDeleted < T extends { _status ? : string ; } > ( record ) { return markStatus ( record , "deleted" ) ; } function markSynced < T extends { _status ? : string ; } > ( record ) { return markStatus ( record , "synced" ) ; } param { IDBTransactionProxy } transaction param { Object } remote param { Array < String > } localFields param { String } strategy The  Collection . strategy .
 *  return { Object } function importChange < T extends { id : string ; last_modified ? : number ; _status ? : RecordStatus ; } > ( transaction , remote , localFields , strategy ) { const local = transaction . get ( remote . id ) ; if ( ! local ) { if ( remote . deleted ) { return { type : "skipped" , data : remote } ; } const synced = markSynced ( remote ) ; transaction . create ( synced ) ; return { type : "created" , data : synced } ; } const synced = { ... local , ... markSynced ( remote ) } ; if ( strategy === Collection . strategy . PULL_ONLY ) { if ( remote . deleted ) { transaction . delete ( remote . id ) ; return { type : "deleted" , data : local } ; } transaction . update ( synced ) ; return { type : "updated" , data : { old : local , new : synced } } ; } const isIdentical = recordsEqual ( local , remote , localFields ) ; if ( local . _status !== "synced" ) { if ( local . _status === "deleted" ) { return { type : "skipped" , data : local } ; } if ( isIdentical ) { transaction . update ( synced ) ; return { type : "updated" , data : { old : local , new : synced } } ; } if ( local . last_modified !== undefined && local . last_modified === remote . last_modified ) { return { type : "void" } ; } return { type : "conflicts" , data : { type : "incoming" , local : local , remote : remote } , } ; } if ( remote . deleted ) { transaction . delete ( remote . id ) ; return { type : "deleted" , data : local } ; } transaction . update ( synced ) ; if ( isIdentical ) { return { type : "void" } ; } return { type : "updated" , data : { old : local , new : synced } } ; } /**
 * Abstracts a collection of records stored in the local database, providing
 * CRUD operations and synchronization helpers.
 */ export default class Collection < B extends { id : string ; last_modified ? : number ; _status ? : RecordStatus ; } = any > { param { String } bucket param { String } name param { KintoBase } kinto param { Object } options private _bucket : string ; private _name : string ; private _lastModified : number | null ; public db : BaseAdapter < B > ; public kinto : KintoBase < B > ; public events ? : Emitter ; public idSchema : IdSchema ; public remoteTransformers : RemoteTransformer [ ] ; public hooks : Hooks < B > ; public localFields : string [ ] ; constructor ( bucket , name , kinto , options = { } ) { this . _bucket = bucket ; this . _name = name ; this . _lastModified = null ; const db = options . adapter ? options . adapter ( `${ bucket }/${ name }` , options . adapterOptions ) : new IDB < B > ( `${ bucket }/${ name }` , options . adapterOptions ) ; if ( ! ( db instanceof BaseAdapter ) ) { throw new Error ( "Unsupported adapter." ) ; } this . db = db ; type { KintoBase } this . kinto = kinto ; type { EventEmitter } this . events = options . events ; type { Object } this . idSchema = this . _validateIdSchema ( options . idSchema ) ; type { Array } this . remoteTransformers = this . _validateRemoteTransformers ( options . remoteTransformers ) ; type { Object } this . hooks = this . _validateHooks ( options . hooks ) ; type { Array } this . localFields = options . localFields || [ ] ; } type { KintoClient } get api ( ) { return this . kinto . api ; } type { String } get name ( ) { return this . _name ; } type { String } get bucket ( ) { return this . _bucket ; } type { Number } get lastModified ( ) { return this . _lastModified ; } type { Object } static get strategy ( ) { return { CLIENT_WINS : "client_wins" , SERVER_WINS : "server_wins" , PULL_ONLY : "pull_only" , MANUAL : "manual" , } ; } param { Object | undefined } idSchema return { Object } private _validateIdSchema ( idSchema ? ) { if ( typeof idSchema === "undefined" ) { return createUUIDSchema ( ) ; } if ( typeof idSchema !== "object" ) { throw new Error ( "idSchema must be an object." ) ; } else if ( typeof idSchema . generate !== "function" ) { throw new Error ( "idSchema must provide a generate function." ) ; } else if ( typeof idSchema . validate !== "function" ) { throw new Error ( "idSchema must provide a validate function." ) ; } return idSchema ; } param { Array | undefined } remoteTransformers return { Array } private _validateRemoteTransformers ( remoteTransformers ? ) { if ( typeof remoteTransformers === "undefined" ) { return [ ] ; } if ( ! Array . isArray ( remoteTransformers ) ) { throw new Error ( "remoteTransformers should be an array." ) ; } return remoteTransformers . map ( ( transformer ) => { if ( typeof transformer !== "object" ) { throw new Error ( "A transformer must be an object." ) ; } else if ( typeof transformer . encode !== "function" ) { throw new Error ( "A transformer must provide an encode function." ) ; } else if ( typeof transformer . decode !== "function" ) { throw new Error ( "A transformer must provide a decode function." ) ; } return transformer ; } ) ; } param { Array | undefined } hook . return { Array } private _validateHook ( hook ) { if ( ! Array . isArray ( hook ) ) { throw new Error ( "A hook definition should be an array of functions." ) ; } return hook . map ( ( fn ) => { if ( typeof fn !== "function" ) { throw new Error ( "A hook definition should be an array of functions." ) ; } return fn ; } ) ; } param { Object | undefined } hooks return { Object } private _validateHooks ( hooks ? ) { if ( typeof hooks === "undefined" ) { return { } ; } if ( Array . isArray ( hooks ) ) { throw new Error ( "hooks should be an object, not an array." ) ; } if ( typeof hooks !== "object" ) { throw new Error ( "hooks should be an object." ) ; } const validatedHooks = { } ; for ( const hook in hooks ) { if ( ! AVAILABLE_HOOKS . includes ( hook as AvailableHook ) ) { throw new Error ( "The hook should be one of " + AVAILABLE_HOOKS . join ( ", " ) ) ; } validatedHooks [ hook as AvailableHook ] = this . _validateHook ( hooks [ hook as AvailableHook ] ! ) ; } return validatedHooks ; } return { Promise } async clear ( ) { await this . db . clear ( ) ; await this . db . saveMetadata ( null ) ; await this . db . saveLastModified ( null ) ; return { data : [ ] , permissions : { } } ; } param { String } type param { Object } record return { Promise } private _encodeRecord ( type , record ) { const transformers = type === "remote" ? this . remoteTransformers : [ ] ; if ( ! transformers . length ) { return Promise . resolve ( record ) ; } return waterfall ( transformers . map ( ( transformer ) => { return ( record ) => transformer . encode ( record ) ; } ) , record ) ; } param { String } type param { Object } record return { Promise } private _decodeRecord ( type , record ) { const transformers = type === "remote" ? this . remoteTransformers : [ ] ; if ( ! transformers . length ) { return Promise . resolve ( record ) ; } return waterfall ( transformers . reverse ( ) . map ( ( transformer ) => { return ( record ) => transformer . decode ( record ) ; } ) , record ) as Promise < B > ; } param { Object } record param { Object } options return { Promise } create ( record , options = { useRecordId : false , synced : false , } ) { const reject = ( msg ) => Promise . reject ( new Error ( msg ) ) ; if ( typeof record !== "object" ) { return reject ( "Record is not an object." ) ; } if ( ( options . synced || options . useRecordId ) && ! Object . prototype . hasOwnProperty . call ( record , "id" ) ) { return reject ( "Missing required Id; synced and useRecordId options require one" ) ; } if ( ! options . synced && ! options . useRecordId && Object . prototype . hasOwnProperty . call ( record , "id" ) ) { return reject ( "Extraneous Id; can't create a record having one set." ) ; } const newRecord = { ... record , id : options . synced || options . useRecordId ? record . id ! : this . idSchema . generate ( record ) , _status : options . synced ? "synced" : "created" , } as B ; if ( ! this . idSchema . validate ( newRecord . id ! ) ) { return reject ( `Invalid Id: ${ newRecord . id }` ) ; } return this . execute ( ( txn ) => txn . create ( newRecord ) , { preloadIds : [ newRecord . id ] , } ) . catch ( ( err ) => { if ( options . useRecordId ) { throw new Error ( "Couldn't create record. It may have been virtually deleted." ) ; } throw err ; } ) ; } /**
     * Like  CollectionTransaction , but wrapped in its own transaction.
     *
     * Options:
     * - {Boolean} synced: Sets record status to "synced" (default: false)
     * - {Boolean} patch:  Extends the existing record instead of overwriting it
     *   (default: false)
     *
     *  param { Object } record param { Object } options return { Promise } update ( record , options = { synced : false , patch : false , } ) { if ( typeof record !== "object" ) { return Promise . reject ( new Error ( "Record is not an object." ) ) ; } if ( ! Object . prototype . hasOwnProperty . call ( record , "id" ) ) { return Promise . reject ( new Error ( "Cannot update a record missing id." ) ) ; } if ( ! this . idSchema . validate ( record . id ) ) { return Promise . reject ( new Error ( `Invalid Id: ${ record . id }` ) ) ; } return this . execute ( ( txn ) => txn . update ( record , { synced : options . synced ?? false , patch : options . patch ?? false , } ) , { preloadIds : [ record . id ] , } ) ; } /**
     * Like  CollectionTransaction , but wrapped in its own transaction.
     *
     *  param { Object } record return { Promise } upsert ( record ) { if ( typeof record !== "object" ) { return Promise . reject ( new Error ( "Record is not an object." ) ) ; } if ( ! Object . prototype . hasOwnProperty . call ( record , "id" ) ) { return Promise . reject ( new Error ( "Cannot update a record missing id." ) ) ; } if ( ! this . idSchema . validate ( record . id ) ) { return Promise . reject ( new Error ( `Invalid Id: ${ record . id }` ) ) ; } return this . execute ( ( txn ) => txn . upsert ( record ) , { preloadIds : [ record . id ] , } ) ; } /**
     * Like  CollectionTransaction , but wrapped in its own transaction.
     *
     * Options:
     * - {Boolean} includeDeleted: Include virtually deleted records.
     *
     *  param { String } id param { Object } options return { Promise } get ( id , options = { includeDeleted : false } ) { return this . execute ( ( txn ) => txn . get ( id , options ) , { preloadIds : [ id ] } ) ; } /**
     * Like  CollectionTransaction , but wrapped in its own transaction.
     *
     *  param { String } id return { Promise } getAny ( id ) { return this . execute ( ( txn ) => txn . getAny ( id ) , { preloadIds : [ id ] } ) ; } /**
     * Same as  Collection , but wrapped in its own transaction.
     *
     * Options:
     * - {Boolean} virtual: When set to `true`, doesn't actually delete the record,
     *   update its `_status` attribute to `deleted` instead (default: true)
     *
     *  param { String } id param { Object } options return { Promise } delete ( id , options = { virtual : true } ) { return this . execute ( ( transaction ) => { return transaction . delete ( id , options ) ; } , { preloadIds : [ id ] } ) ; } /**
     * Same as  Collection , but wrapped in its own transaction, execulding the parameter.
     *
     *  return { Promise } async deleteAll ( ) { const { data } = await this . list ( { } , { includeDeleted : false } ) ; const recordIds = data . map ( ( record ) => record . id ) ; return this . execute ( ( transaction ) => { return transaction . deleteAll ( recordIds ) ; } , { preloadIds : recordIds } ) ; } /**
     * The same as  CollectionTransaction , but wrapped
     * in its own transaction.
     *
     *  param { String } id return { Promise } deleteAny ( id ) { return this . execute ( ( txn ) => txn . deleteAny ( id ) , { preloadIds : [ id ] } ) ; } param { Object } params param { Object } options return { Promise } async list ( params = { } , options = { includeDeleted : false } ) { params = { order : "-last_modified" , filters : { } , ... params } ; const results = await this . db . list ( params ) ; let data = results ; if ( ! options . includeDeleted ) { data = results . filter ( ( record ) => record . _status !== "deleted" ) ; } return { data , permissions : { } } ; } param { SyncResultObject } syncResultObject param { Array } decodedChanges param { String } strategy The  Collection . strategy (default: MANUAL)
     *  return { Promise } async importChanges ( syncResultObject , decodedChanges , strategy = Collection . strategy . MANUAL ) { try { for ( let i = 0 ; i < decodedChanges . length ; i += IMPORT_CHUNK_SIZE ) { const slice = decodedChanges . slice ( i , i + IMPORT_CHUNK_SIZE ) ; const { imports , resolved } = await this . db . execute ( ( transaction ) => { const imports = slice . map ( ( remote ) => { return importChange ( transaction , remote , this . localFields , strategy ) ; } ) ; const conflicts = ( imports . filter ( ( i ) => i . type === "conflicts" ) as ConflictsChange < B > [ ] ) . map ( ( i ) => i . data ) ; const resolved = this . _handleConflicts ( transaction , conflicts , strategy ) ; return { imports , resolved } ; } , { preload : slice . map ( ( record ) => record . id ) } ) ; imports . forEach ( ( { type , data } ) => syncResultObject . add ( type , data ) ) ; if ( resolved . length > 0 ) { syncResultObject . reset ( "conflicts" ) . add ( "resolved" , resolved ) ; } } } catch ( err ) { const data = { type : "incoming" , message : err . message , stack : err . stack , } ; syncResultObject . add ( "errors" , data ) ; } return syncResultObject ; } param { SyncResultObject } syncResultObject param { Array } toApplyLocally param { Array } conflicts param { String } strategy The  Collection . strategy .
     *  return { Promise } private async _applyPushedResults ( syncResultObject , toApplyLocally , conflicts , strategy = Collection . strategy . MANUAL ) { const toDeleteLocally = toApplyLocally . filter ( ( r ) => r . deleted ) ; const toUpdateLocally = toApplyLocally . filter ( ( r ) => ! r . deleted ) ; const { published , resolved } = await this . db . execute ( ( transaction ) => { const updated = toUpdateLocally . map ( ( record ) => { const synced = markSynced ( record ) ; transaction . update ( synced ) ; return synced ; } ) ; const deleted = toDeleteLocally . map ( ( record ) => { transaction . delete ( record . id ) ; return { id : record . id , deleted : true } ; } ) ; const published = updated . concat ( deleted ) ; const resolved = this . _handleConflicts ( transaction , conflicts , strategy ) ; return { published , resolved } ; } ) ; syncResultObject . add ( "published" , published ) ; if ( resolved . length > 0 ) { syncResultObject . reset ( "conflicts" ) . reset ( "resolved" ) . add ( "resolved" , resolved ) ; } return syncResultObject ; } param { SyncResultObject } result param { String } strategy The  Collection . strategy .
     *  return { Promise < Array < Object > > } private _handleConflicts ( transaction , conflicts , strategy ) { if ( strategy === Collection . strategy . MANUAL ) { return [ ] ; } return conflicts . map ( ( conflict ) => { const resolution = strategy === Collection . strategy . CLIENT_WINS ? conflict . local : conflict . remote ; const rejected = strategy === Collection . strategy . CLIENT_WINS ? conflict . remote : conflict . local ; let accepted , status , id ; if ( resolution === null ) { transaction . delete ( conflict . local . id ) ; accepted = null ; status = "synced" ; id = conflict . local . id ; } else { const updated = this . _resolveRaw ( conflict , resolution ) ; transaction . update ( updated ) ; accepted = updated ; status = updated . _status ; id = updated . id ; } return { rejected , accepted , id , _status : status } ; } ) ; } /**
     * Execute a bunch of operations in a transaction.
     *
     * This transaction should be atomic -- either all of its operations
     * will succeed, or none will.
     *
     * The argument to this function is itself a function which will be
     * called with a  CollectionTransaction . Collection methods
     * are available on this transaction, but instead of returning
     * promises, they are synchronous. execute() returns a Promise whose
     * value will be the return value of the provided function.
     *
     * Most operations will require access to the record itself, which
     * must be preloaded by passing its ID in the preloadIds option.
     *
     * Options:
     * - {Array} preloadIds: list of IDs to fetch at the beginning of
     *   the transaction
     *
     *  return { Promise } execute < R > ( doOperations , { preloadIds = [ ] } = { } ) { for ( const id of preloadIds ) { if ( ! this . idSchema . validate ( id ) ) { return Promise . reject ( Error ( `Invalid Id: ${ id }` ) ) ; } } return this . db . execute ( ( transaction ) => { const txn = new CollectionTransaction ( this , transaction ) ; const result = doOperations ( txn ) ; txn . emitEvents ( ) ; return result ; } , { preload : preloadIds } ) ; } /**
     * Resets the local records as if they were never synced; existing records are
     * marked as newly created, deleted records are dropped.
     *
     * A next call to  Collection . sync will thus republish the whole
     * content of the local collection to the server.
     *
     *  return { Promise } async resetSyncStatus ( ) { const unsynced = await this . list ( { filters : { _status : [ "deleted" , "synced" ] } , order : "" } , { includeDeleted : true } ) ; await this . db . execute ( ( transaction ) => { unsynced . data . forEach ( ( record ) => { if ( record . _status === "deleted" ) { transaction . delete ( record . id ) ; } else { transaction . update ( { ... record , last_modified : undefined , _status : "created" , } ) ; } } ) ; } ) ; this . _lastModified = null ; await this . db . saveLastModified ( null ) ; return unsynced . data . length ; } return { Promise } async gatherLocalChanges ( ) { const unsynced = await this . list ( { filters : { _status : [ "created" , "updated" ] } , order : "" , } ) ; const deleted = await this . list ( { filters : { _status : "deleted" } , order : "" } , { includeDeleted : true } ) ; return await Promise . all ( unsynced . data . concat ( deleted . data ) . map ( this . _encodeRecord . bind ( this , "remote" ) ) ) ; } /**
     * Fetch remote changes, import them to the local database, and handle
     * conflicts according to `options.strategy`. Then, updates the passed
     *  SyncResultObject with import results.
     *
     * Options:
     * - {String} strategy: The selected sync strategy.
     * - {String} expectedTimestamp: A timestamp to use as a "cache busting" query parameter.
     * - {Array<String>} exclude: A list of record ids to exclude from pull.
     * - {Object} headers: The HTTP headers to use in the request.
     * - {int} retry: The number of retries to do if the HTTP request fails.
     * - {int} lastModified: The timestamp to use in `?_since` query.
     *
     *  param { KintoClient . Collection } client param { SyncResultObject } syncResultObject param { Object } options return { Promise } async pullChanges ( client , syncResultObject , options = { } ) { if ( ! syncResultObject . ok ) { return syncResultObject ; } const since = this . lastModified ? this . lastModified : await this . db . getLastModified ( ) ; options = { strategy : Collection . strategy . MANUAL , lastModified : since , headers : { } , ... options , } ; let filters ; if ( options . exclude ) { const exclude_id = options . exclude . slice ( 0 , 50 ) . map ( ( r ) => r . id ) . join ( "," ) ; filters = { exclude_id } ; } if ( options . expectedTimestamp ) { filters = { ... filters , _expected : options . expectedTimestamp , } ; } const { data , last_modified } = await client . listRecords ( { since : options . lastModified ? `${ options . lastModified }` : undefined , headers : options . headers , retry : options . retry , pages : Infinity , filters , } ) ; const unquoted = last_modified ? parseInt ( last_modified , 10 ) : undefined ; const localSynced = options . lastModified ; const serverChanged = unquoted && unquoted > options . lastModified ! ; const emptyCollection = data . length === 0 ; if ( ! options . exclude && localSynced && serverChanged && emptyCollection ) { const e = new ServerWasFlushedError ( localSynced , unquoted , "Server has been flushed. Client Side Timestamp: " + localSynced + " Server Side Timestamp: " + unquoted ) ; throw e ; } syncResultObject . lastModified = unquoted ; const decodedChanges = await Promise . all ( data . map ( ( change ) => { return this . _decodeRecord ( "remote" , change ) ; } ) ) ; const payload = { lastModified : unquoted , changes : decodedChanges } ; const afterHooks = await this . applyHook ( "incoming-changes" , payload ) ; if ( afterHooks . changes . length > 0 ) { await this . importChanges ( syncResultObject , afterHooks . changes , options . strategy ) ; } return syncResultObject ; } applyHook ( hookName , payload ) { if ( typeof this . hooks [ hookName ] === "undefined" ) { return Promise . resolve ( payload ) ; } return waterfall ( this . hooks [ hookName ] ! . map ( ( hook ) => { return ( record ) => { const result = hook ( payload , this ) ; const resultThenable = result && typeof result . then === "function" ; const resultChanges = result && Object . prototype . hasOwnProperty . call ( result , "changes" ) ; if ( ! ( resultThenable || resultChanges ) ) { throw new Error ( `Invalid return value for hook: ${ JSON . stringify ( result ) } has no 'then()' or 'changes' properties` ) ; } return result ; } ; } ) , payload ) as Promise < { changes : B [ ] ; } > ; } /**
     * Publish local changes to the remote server and updates the passed
     *  SyncResultObject with publication results.
     *
     * Options:
     * - {String} strategy: The selected sync strategy.
     * - {Object} headers: The HTTP headers to use in the request.
     * - {int} retry: The number of retries to do if the HTTP request fails.
     *
     *  param { KintoClient . Collection } client param { SyncResultObject } syncResultObject param changes param { Array } changes . toDelete param { Array } changes . toSync param { Object } options return { Promise } async pushChanges ( client , changes , syncResultObject , options = { } ) { if ( ! syncResultObject . ok ) { return syncResultObject ; } const safe = ! options . strategy || options . strategy !== undefined ; const toDelete = changes . filter ( ( r ) => r . _status === "deleted" ) ; const toSync = changes . filter ( ( r ) => r . _status != "deleted" ) ; const synced = ( await client . batch ( ( batch ) => { toDelete . forEach ( ( r ) => { if ( r . last_modified ) { batch . deleteRecord ( r ) ; } } ) ; toSync . forEach ( ( r ) => { const published = this . cleanLocalFields ( r ) as B ; if ( r . _status === "created" ) { batch . createRecord ( published ) ; } else { batch . updateRecord ( published ) ; } } ) ; } , { headers : options . headers , retry : options . retry , safe , aggregate : true , } ) ) as AggregateResponse ; syncResultObject . add ( "errors" , synced . errors . map ( ( e ) => ( { ... e , type : "outgoing" } ) ) ) ; const conflicts = [ ] ; for ( const { type , local , remote } of synced . conflicts ) { const safeLocal = ( local && local . data ) || { id : remote . id } ; const realLocal = await this . _decodeRecord ( "remote" , safeLocal ) ; const realRemote = remote && ( await this . _decodeRecord ( "remote" , remote ) ) ; const conflict = { type , local : realLocal , remote : realRemote } ; conflicts . push ( conflict ) ; } syncResultObject . add ( "conflicts" , conflicts ) ; const missingRemotely = synced . skipped . map ( ( r ) => ( { ... r , deleted : true , } ) ) ; const published = synced . published . map ( ( c ) => c . data ) ; const toApplyLocally = published . concat ( missingRemotely ) ; const decoded = await Promise . all ( toApplyLocally . map ( ( record ) => { return this . _decodeRecord ( "remote" , record ) ; } ) ) ; if ( decoded . length > 0 || conflicts . length > 0 ) { await this . _applyPushedResults ( syncResultObject , decoded , conflicts , options . strategy ) ; } return syncResultObject ; } param { Object } record return { Object } cleanLocalFields ( record ) { const localKeys = [ ... RECORD_FIELDS_TO_CLEAN , ... this . localFields ] ; return omitKeys ( record , localKeys ) ; } param { Object } conflict param { Object } resolution return { Promise } resolve ( conflict , resolution ) { return this . db . execute ( ( transaction ) => { const updated = this . _resolveRaw ( conflict , resolution ) ; transaction . update ( updated ) ; return { data : updated , permissions : { } } ; } ) ; } private private _resolveRaw ( conflict , resolution ) { const resolved = { ... resolution , last_modified : conflict . remote && conflict . remote . last_modified , } ; const synced = deepEqual ( resolved , conflict . remote ) ; return markStatus ( resolved , synced ? "synced" : "updated" ) ; } /**
     * Synchronize remote and local data. The promise will resolve with a
     *  SyncResultObject , though will reject:
     *
     * - if the server is currently backed off;
     * - if the server has been detected flushed.
     *
     * Options:
     * - {Object} headers: HTTP headers to attach to outgoing requests.
     * - {String} expectedTimestamp: A timestamp to use as a "cache busting" query parameter.
     * - {Number} retry: Number of retries when server fails to process the request (default: 1).
     * - {Collection.strategy} strategy: See  Collection . strategy .
     * - {Boolean} ignoreBackoff: Force synchronization even if server is currently
     *   backed off.
     * - {String} bucket: The remove bucket id to use (default: null)
     * - {String} collection: The remove collection id to use (default: null)
     * - {String} remote The remote Kinto server endpoint to use (default: null).
     *
     *  param { Object } options return { Promise } throws async sync ( options = { strategy : Collection . strategy . MANUAL , headers : { } , retry : 1 , ignoreBackoff : false , bucket : null , collection : null , remote : null , expectedTimestamp : null , } ) { options = { ... options , bucket : options . bucket || this . bucket , collection : options . collection || this . name , } ; const previousRemote = this . api . remote ; if ( options . remote ) { this . api . remote = options . remote ; } if ( ! options . ignoreBackoff && this . api . backoff > 0 ) { const seconds = Math . ceil ( this . api . backoff / 1000 ) ; return Promise . reject ( new Error ( `Server is asking clients to back off; retry in ${ seconds }s or use the ignoreBackoff option.` ) ) ; } const client = this . api . bucket ( options . bucket ! ) . collection ( options . collection ! ) ; const result = new SyncResultObject ( ) ; try { await this . pullMetadata ( client , options ) ; await this . pullChanges ( client , result , options ) ; const { lastModified } = result ; if ( options . strategy !== Collection . strategy . PULL_ONLY ) { const toSync = await this . gatherLocalChanges ( ) ; await this . pushChanges ( client , toSync , result , options ) ; const resolvedUnsynced = result . resolved . filter ( ( r ) => r . _status !== "synced" ) ; if ( resolvedUnsynced . length > 0 ) { const resolvedEncoded = await Promise . all ( resolvedUnsynced . map ( ( resolution ) => { let record = resolution . accepted ; if ( record === null ) { record = { id : resolution . id , _status : resolution . _status } ; } return this . _encodeRecord ( "remote" , record ) ; } ) ) ; await this . pushChanges ( client , resolvedEncoded , result , options ) ; } if ( result . published . length > 0 ) { const pullOpts = { ... options , lastModified , exclude : result . published , } ; await this . pullChanges ( client , result , pullOpts ) ; } } if ( result . ok ) { this . _lastModified = await this . db . saveLastModified ( result . lastModified ) ; } } catch ( e ) { this . events ?. emit ( "sync:error" , { ... options , error : e } ) ; throw e ; } finally { this . api . remote = previousRemote ; } this . events ?. emit ( "sync:success" , { ... options , result } ) ; return result ; } deprecated Use  importBulk instead.
     *  param { Array } records return { Promise } async loadDump ( records ) { return this . importBulk ( records ) ; } param { Array } records return { Promise } async importBulk ( records ) { if ( ! Array . isArray ( records ) ) { throw new Error ( "Records is not an array." ) ; } for ( const record of records ) { if ( ! Object . prototype . hasOwnProperty . call ( record , "id" ) || ! this . idSchema . validate ( record . id ) ) { throw new Error ( "Record has invalid ID: " + JSON . stringify ( record ) ) ; } if ( ! record . last_modified ) { throw new Error ( "Record has no last_modified value: " + JSON . stringify ( record ) ) ; } } const { data } = await this . list ( { } , { includeDeleted : true } ) ; const existingById = data . reduce < { [ key ] : B ; } > ( ( acc , record ) => { acc [ record . id ! ] = record ; return acc ; } , { } ) ; const newRecords = records . filter ( ( record ) => { const localRecord = existingById [ record . id ] ; const shouldKeep = localRecord === undefined || ( localRecord . _status === "synced" && localRecord . last_modified !== undefined && record . last_modified > localRecord . last_modified ) ; return shouldKeep ; } ) ; return await this . db . importBulk ( newRecords . map ( markSynced ) ) ; } async pullMetadata ( client , options = { } ) { const { expectedTimestamp , headers } = options ; const query = expectedTimestamp ? { query : { _expected : expectedTimestamp . toString ( ) } } : undefined ; const metadata = await client . getData ( { ... query , headers , } ) ; return this . db . saveMetadata ( metadata as any ) ; } async metadata ( ) { return this . db . getMetadata ( ) ; } } /**
 * A Collection-oriented wrapper for an adapter's transaction.
 *
 * This defines the high-level functions available on a collection.
 * The collection itself offers functions of the same name. These will
 * perform just one operation in its own transaction.
 */ export class CollectionTransaction < B extends { id : string ; last_modified ? : number ; _status ? : RecordStatus ; } > { public collection : Collection < B > ; public adapterTransaction : StorageProxy < B > ; private _events : { action : string ; payload : any ; } [ ] ; constructor ( collection , adapterTransaction ) { this . collection = collection ; this . adapterTransaction = adapterTransaction ; this . _events = [ ] ; } private _queueEvent ( action , payload ) { this . _events . push ( { action , payload } ) ; } /**
     * Emit queued events, to be called once every transaction operations have
     * been executed successfully.
     */ emitEvents ( ) { for ( const { action , payload } of this . _events ) { this . collection . events ?. emit ( action , payload ) ; } if ( this . _events . length > 0 ) { const targets = this . _events . map ( ( { action , payload } ) => ( { action , ... payload , } ) ) ; this . collection . events ?. emit ( "change" , { targets } ) ; } this . _events = [ ] ; } param { String } id return { Object } getAny ( id ) { const record = this . adapterTransaction . get ( id ) ; return { data : record , permissions : { } } ; } param { String } id param { Object } options return { Object } get ( id , options = { includeDeleted : false } ) { const res = this . getAny ( id ) ; if ( ! res . data || ( ! options . includeDeleted && res . data . _status === "deleted" ) ) { throw new Error ( `Record with id=${ id } not found.` ) ; } return res as KintoRepresentation < B > ; } param { String } id param { Object } options return { Object } delete ( id , options = { virtual : true } ) { const existing = this . adapterTransaction . get ( id ) ; const alreadyDeleted = existing && existing . _status === "deleted" ; if ( ! existing || ( alreadyDeleted && options . virtual ) ) { throw new Error ( `Record with id=${ id } not found.` ) ; } if ( options . virtual ) { this . adapterTransaction . update ( markDeleted ( existing ) ) ; } else { this . adapterTransaction . delete ( id ) ; } this . _queueEvent ( "delete" , { data : existing } ) ; return { data : existing , permissions : { } } ; } param { Array } ids return { Object } deleteAll ( ids ) { const existingRecords = [ ] ; ids . forEach ( ( id ) => { existingRecords . push ( this . adapterTransaction . get ( id ) ) ; this . delete ( id ) ; } ) ; this . _queueEvent ( "deleteAll" , { data : existingRecords } ) ; return { data : existingRecords , permissions : { } } ; } param { String } id return { Object } deleteAny ( id ) { const existing = this . adapterTransaction . get ( id ) ; if ( existing ) { this . adapterTransaction . update ( markDeleted ( existing ) ) ; this . _queueEvent ( "delete" , { data : existing } ) ; } return { data : { id , ... existing } as B , deleted : ! ! existing , permissions : { } , } ; } param { Object } record return { Object } create ( record ) { if ( typeof record !== "object" ) { throw new Error ( "Record is not an object." ) ; } if ( ! Object . prototype . hasOwnProperty . call ( record , "id" ) ) { throw new Error ( "Cannot create a record missing id" ) ; } if ( ! this . collection . idSchema . validate ( record . id ) ) { throw new Error ( `Invalid Id: ${ record . id }` ) ; } this . adapterTransaction . create ( record ) ; this . _queueEvent ( "create" , { data : record } ) ; return { data : record , permissions : { } } ; } param { Object } record param { Object } options return { Object } update ( record , options = { synced : false , patch : false , } ) { if ( typeof record !== "object" ) { throw new Error ( "Record is not an object." ) ; } if ( ! Object . prototype . hasOwnProperty . call ( record , "id" ) ) { throw new Error ( "Cannot update a record missing id." ) ; } if ( ! this . collection . idSchema . validate ( record . id ) ) { throw new Error ( `Invalid Id: ${ record . id }` ) ; } const oldRecord = this . adapterTransaction . get ( record . id ) ; if ( ! oldRecord ) { throw new Error ( `Record with id=${ record . id } not found.` ) ; } const newRecord = options . patch ? { ... oldRecord , ... record } : record ; const updated = this . _updateRaw ( oldRecord , newRecord , options ) ; this . adapterTransaction . update ( updated ) ; this . _queueEvent ( "update" , { data : updated , oldRecord } ) ; return { data : updated , oldRecord , permissions : { } } ; } param { Object } oldRecord param { Object } newRecord return { Object } private _updateRaw ( oldRecord , newRecord , { synced = false } = { } ) { const updated = { ... newRecord } ; if ( oldRecord && oldRecord . last_modified && ! updated . last_modified ) { updated . last_modified = oldRecord . last_modified ; } const isIdentical = oldRecord && recordsEqual ( oldRecord , updated , this . collection . localFields ) ; const keepSynced = isIdentical && oldRecord . _status === "synced" ; const neverSynced = ! oldRecord || ( oldRecord && oldRecord . _status === "created" ) ; const newStatus = keepSynced || synced ? "synced" : neverSynced ? "created" : "updated" ; return markStatus ( updated , newStatus ) ; } param { Object } record return { Object } upsert ( record ) { if ( typeof record !== "object" ) { throw new Error ( "Record is not an object." ) ; } if ( ! Object . prototype . hasOwnProperty . call ( record , "id" ) ) { throw new Error ( "Cannot update a record missing id." ) ; } if ( ! this . collection . idSchema . validate ( record . id ) ) { throw new Error ( `Invalid Id: ${ record . id }` ) ; } let oldRecord = this . adapterTransaction . get ( record . id ) ; const updated = this . _updateRaw ( oldRecord , record ) ; this . adapterTransaction . update ( updated ) ; if ( oldRecord && oldRecord . _status === "deleted" ) { oldRecord = undefined ; } if ( oldRecord ) { this . _queueEvent ( "update" , { data : updated , oldRecord } ) ; } else { this . _queueEvent ( "create" , { data : updated } ) ; } return { data : updated , oldRecord , permissions : { } } ; } }