import { Observable } from "rxjs" ; import { filter , map } from "rxjs/operators" ; import { AdvertiseEvent , CoatyObject , Controller , CoreTypes , filterOp , ObjectFilter , QueryEvent , RetrieveEvent , Snapshot , Uuid , } from ".." ; import { DbContext } from "./db-context" ; /**
 * A controller which can be used on each Coaty agent to create and/or persist
 * Snapshot objects in time of arbitrary Coaty objects.
 *
 * Available controller options (default value for all options is `false`):
 * - `shouldAdvertiseSnapshots`: if `true` each generated Snapshot object will
 *   be advertised
 * - `shouldPersistLocalSnapshots`: if `true` each Snapshot object generated
 *   will be persisted in a database collection if the `database` option is set
 *   with valid values
 * - `shouldPersistObservedSnapshots`: if `true` each Snapshot object observed
 *   as Advertise event will be persisted in database if the `database` option
 *   is set with valid values; Note: local snapshots will not be observed and
 *   hereby not persisted, if `shouldPersistLocalSnapshots` is not additionally
 *   set to `true`.
 * - `shouldReplyToQueries`: if `true` HistorianController will execute matching
 *   query events on the database if the `database` option is set with valid
 *   values
 * - `database`: with two properties `key` and `collection`; `key` references
 *   the database key as defined in the `databases` option of your
 *   configuration; `collection` is the name of the collection to be used. If
 *   the collection doesn't exist, it will be created in the given database.
 */ export class HistorianController extends Controller { /**
     * Log tag for logging database operations.
     */ private static readonly _LOG_TAG_DB = "db" ; private _dbCtxPromise : Promise < DbContext > ; onInit ( ) { super . onInit ( ) ; } onCommunicationManagerStarting ( ) { super . onCommunicationManagerStarting ( ) ; if ( this . options [ "shouldReplyToQueries" ] ) { this . _observeQueries ( ) ; } if ( this . options [ "shouldPersistObservedSnapshots" ] ) { this . _observeAdvertise ( ) ; } } param obj param tags generateSnapshot ( obj , ... tags ) ; param obj param timestamp param tags generateSnapshot ( obj , timestamp , ... tags ) ; generateSnapshot ( obj , param1 ? , ... param2 ) { if ( typeof param1 === "string" ) { param2 . push ( param1 ) ; } if ( typeof param1 !== "number" ) { param1 = Date . now ( ) ; } const snapshot = { objectId : this . runtime . newUuid ( ) , parentObjectId : obj . objectId , objectType : CoreTypes . OBJECT_TYPE_SNAPSHOT , coreType : "Snapshot" , name : `${ obj . objectType } ${ obj . name }` , creatorId : this . container . identity . objectId , creationTimestamp : param1 , object : CoreTypes . clone ( obj ) , tags : param2 , } ; return this . _processSnaphot ( snapshot ) ; } param parentObjectId querySnapshotsByParentId ( parentObjectId ) { const objectFilter = { conditions : [ "parentObjectId" , filterOp . equals ( parentObjectId ) ] , orderByProperties : [ [ "creationTimestamp" , "Desc" ] ] , } ; return this . communicationManager . publishQuery ( QueryEvent . withCoreTypes ( [ "Snapshot" ] , objectFilter ) ) . pipe ( map ( retrieve => retrieve . data . objects as Snapshot [ ] ) ) ; } param parentObjectId param startTimestamp param endTimestamp findSnapshotsByParentId ( parentObjectId , startTimestamp ? , endTimestamp ? ) { const andConds = [ ] ; if ( parentObjectId ) { andConds . push ( [ "parentObjectId" , filterOp . equals ( parentObjectId ) ] ) ; } if ( startTimestamp ) { andConds . push ( [ "creationTimestamp" , filterOp . greaterThanOrEqual ( startTimestamp ) ] ) ; } if ( endTimestamp ) { andConds . push ( [ "creationTimestamp" , filterOp . lessThanOrEqual ( endTimestamp ) ] ) ; } const objectFilter = { conditions : { and : andConds , } , orderByProperties : [ [ "creationTimestamp" , "Desc" ] ] , } ; return this . findSnapshotsByFilter ( objectFilter ) ; } param startTimestamp param endTimestamp findSnapshotsByTimeFrame ( startTimestamp ? , endTimestamp ? ) { return this . findSnapshotsByParentId ( undefined , startTimestamp , endTimestamp ) ; } param objectFilter findSnapshotsByFilter ( objectFilter ) { let snapshots ; return this . _getDbContext ( ) . then ( dbCtx => dbCtx . findObjects < Snapshot > ( this . options [ "database" ] . collection , objectFilter ) ) . then ( iterator => iterator . forBatch ( results => { snapshots = results ; } ) ) . then ( ( ) => snapshots ) ; } param event protected onQueryReceived ( event ) { } param event param snapshots protected onQueryRetrieved ( event , snapshots ) { } private _observeQueries ( ) { return this . communicationManager . observeQuery ( ) . pipe ( filter ( event => event . data . isCoreTypeCompatible ( "Snapshot" ) ) ) . subscribe ( event => { this . onQueryReceived ( event ) ; this . _getDbContext ( ) . then ( dbCtx => dbCtx . findObjects < Snapshot > ( this . options [ "database" ] . collection , event . data . objectFilter ) ) . then ( iterator => iterator . forBatch ( results => { this . onQueryRetrieved ( event , results ) ; event . retrieve ( RetrieveEvent . withObjects ( results ) ) ; } ) ) . catch ( error => { } ) ; } ) ; } private _observeAdvertise ( ) { return this . communicationManager . observeAdvertiseWithCoreType ( "Snapshot" ) . subscribe ( event => { this . _getDbContext ( ) . then ( dbCtx => dbCtx . insertObjects ( this . options [ "database" ] . collection , event . data . object ) ) . catch ( error => this . logError ( error , "Failed to persist observed snapshot" , HistorianController . _LOG_TAG_DB ) ) ; } ) ; } private _processSnaphot ( snapshot ) { const advertiseIfRequested = ( s ) => { if ( this . options [ "shouldAdvertiseSnapshots" ] ) { this . communicationManager . publishAdvertise ( AdvertiseEvent . withObject ( s ) ) ; } return s ; } ; if ( this . options [ "shouldPersistLocalSnapshots" ] ) { return this . _getDbContext ( ) . then ( dbCtx => dbCtx . insertObjects ( this . options [ "database" ] . collection , snapshot ) , error => this . logError ( error , "Failed to persist new snapshot" , HistorianController . _LOG_TAG_DB ) ) . then ( ( ) => advertiseIfRequested ( snapshot ) ) ; } else { return Promise . resolve ( advertiseIfRequested ( snapshot ) ) ; } } private _getDbContext ( ) { if ( this . _dbCtxPromise === undefined ) { if ( this . options [ "database" ] && this . options [ "database" ] . key && this . options [ "database" ] . collection ) { const dbCtx = new DbContext ( this . runtime . databaseOptions [ this . options [ "database" ] . key ] ) ; this . _dbCtxPromise = dbCtx . addCollection ( this . options [ "database" ] . collection ) . then ( ( ) => dbCtx ) ; } else { this . _dbCtxPromise = Promise . reject ( "Unspecified database options for HistorianController" ) ; } } return this . _dbCtxPromise ; } }