import { stat } from "fs" ; import Loki from "lokijs" ; import StorageErrorFactory from "../errors/StorageErrorFactory" ; import Context from "../generated/Context" ; import { QUEUE_STATUSCODE } from "../utils/constants" ; import { IQueueMetadata , IQueueMetadataStore , MessageModel , MessageUpdateProperties , QueueACL , QueueModel , ServicePropertiesModel } from "./IQueueMetadataStore" ; import QueueReferredExtentsAsyncIterator from "./QueueReferredExtentsAsyncIterator" ; export class export default class LokiQueueMetadataStore implements IQueueMetadataStore { private readonly db : Loki ; private initialized : boolean = false ; private closed : boolean = false ; private readonly SERVICES_COLLECTION = "$SERVICES_COLLECTION$" ; private readonly QUEUES_COLLECTION = "$QUEUES_COLLECTION$" ; private readonly MESSAGES_COLLECTION = "$MESSAGES_COLLECTION$" ; public constructor ( public readonly lokiDBPath ) { this . db = new Loki ( lokiDBPath , { autosave : true , autosaveInterval : 5000 } ) ; } public isInitialized ( ) { return this . initialized ; } public isClosed ( ) { return this . closed ; } public async init ( ) { await new Promise < void > ( ( resolve , reject ) => { stat ( this . lokiDBPath , ( statError , stats ) => { if ( ! statError ) { this . db . loadDatabase ( { } , ( dbError ) => { if ( dbError ) { reject ( dbError ) ; } else { resolve ( ) ; } } ) ; } else { resolve ( ) ; } } ) ; } ) ; let servicePropertiesColl = this . db . getCollection ( this . SERVICES_COLLECTION ) ; if ( servicePropertiesColl === null ) { servicePropertiesColl = this . db . addCollection ( this . SERVICES_COLLECTION , { unique : [ "accountName" ] } ) ; } if ( this . db . getCollection ( this . QUEUES_COLLECTION ) === null ) { this . db . addCollection ( this . QUEUES_COLLECTION , { indices : [ "accountName" , "name" ] } ) ; } if ( this . db . getCollection ( this . MESSAGES_COLLECTION ) === null ) { this . db . addCollection ( this . MESSAGES_COLLECTION , { indices : [ "accountName" , "queueName" , "messageId" , "visibleTime" ] } ) ; } await new Promise < void > ( ( resolve , reject ) => { this . db . saveDatabase ( ( err ) => { if ( err ) { reject ( err ) ; } else { resolve ( ) ; } } ) ; } ) ; this . initialized = true ; this . closed = false ; } returns { Promise < void > } memberof public async close ( ) { await new Promise < void > ( ( resolve , reject ) => { this . db . close ( ( err ) => { if ( err ) { reject ( err ) ; } else { resolve ( ) ; } } ) ; } ) ; this . closed = true ; } param { ServicePropertiesModel } updateProperties returns { Promise < void > } memberof public async updateServiceProperties ( updateProperties ) { const coll = this . db . getCollection ( this . SERVICES_COLLECTION ) ; const doc = coll . by ( "accountName" , updateProperties . accountName ) ; if ( doc ) { doc . cors = updateProperties . cors === undefined ? doc . cors : updateProperties . cors ; doc . hourMetrics = updateProperties . hourMetrics === undefined ? doc . hourMetrics : updateProperties . hourMetrics ; doc . logging = updateProperties . logging === undefined ? doc . logging : updateProperties . logging ; doc . minuteMetrics = updateProperties . minuteMetrics === undefined ? doc . minuteMetrics : updateProperties . minuteMetrics ; coll . update ( doc ) ; } else { coll . insert ( updateProperties ) ; } } template T param { string } account returns { ( Promise < ServicePropertiesModel | undefined > ) } memberof public async getServiceProperties ( account ) { const coll = this . db . getCollection ( this . SERVICES_COLLECTION ) ; const doc = coll . by ( "accountName" , account ) ; return doc ? doc : undefined ; } template T param { string } account param { string } prefix param { number } maxResults param { number } marker returns { ( Promise < [ QueueModel [ ] , number | undefined ] > ) } memberof public async listQueues ( account , prefix = "" , maxResults = 5000 , marker = 0 ) { const coll = this . db . getCollection ( this . QUEUES_COLLECTION ) ; const query = prefix === "" ? { $loki : { $gt : marker } , accountName : account } : { name : { $regex : `^${ this . escapeRegex ( prefix ) }` } , $loki : { $gt : marker } , accountName : account } ; const docs = coll . chain ( ) . find ( query ) . sort ( ( obj1 , obj2 ) => { if ( obj1 . name === obj2 . name ) return 0 ; if ( obj1 . name > obj2 . name ) return 1 ; return - 1 ; } ) . limit ( maxResults + 1 ) . data ( ) ; const queues = [ ] ; for ( let i = 0 ; i < maxResults && i < docs . length ; i ++ ) { queues . push ( this . queueCopy ( docs [ i ] ) ) ; } if ( docs . length <= maxResults ) { return [ queues , undefined ] ; } else { const nextMarker = docs [ docs . length - 1 ] . $loki - 1 ; return [ queues , nextMarker ] ; } } param { string } account param { string } queue param { Context } context returns { Promise < QueueModel > } memberof public async getQueue ( account , queue , context ? ) { const coll = this . db . getCollection ( this . QUEUES_COLLECTION ) ; const doc = coll . findOne ( { accountName : account , name : queue } ) ; if ( ! doc ) { const requestId = context ? context . contextID : undefined ; throw StorageErrorFactory . getQueueNotFound ( requestId ) ; } return doc ; } see https param { QueueModel } queue param { Context } context returns { Promise < QUEUE_STATUSCODE > } memberof public async createQueue ( queue , context ? ) { const coll = this . db . getCollection ( this . QUEUES_COLLECTION ) ; const doc = coll . findOne ( { accountName : queue . accountName , name : queue . name } ) ; if ( doc ) { const docMeta = doc . metadata ; const queueMeta = queue . metadata ; if ( queueMeta === undefined || queueMeta === { } ) { if ( docMeta !== undefined && docMeta . length !== { } ) { throw StorageErrorFactory . getQueueAlreadyExists ( context ? context . contextID : undefined ) ; } else { return 204 ; } } if ( docMeta === undefined || docMeta . length === { } ) { throw StorageErrorFactory . getQueueAlreadyExists ( context ? context . contextID : undefined ) ; } if ( Object . keys ( queueMeta ) . length !== Object . keys ( docMeta ) . length ) { throw StorageErrorFactory . getQueueAlreadyExists ( context ? context . contextID : undefined ) ; } const nameMap = new Map < string , string > ( ) ; for ( const item in queueMeta ) { if ( queueMeta . hasOwnProperty ( item ) ) { nameMap . set ( item . toLowerCase ( ) , item ) ; } } for ( const item in docMeta ) { if ( docMeta . hasOwnProperty ( item ) ) { const queueMetaName = nameMap . get ( item . toLowerCase ( ) ) ; if ( queueMetaName === undefined ) { throw StorageErrorFactory . getQueueAlreadyExists ( context ? context . contextID : undefined ) ; } if ( docMeta [ item ] !== queueMeta [ queueMetaName ] ) { throw StorageErrorFactory . getQueueAlreadyExists ( context ? context . contextID : undefined ) ; } } } return 204 ; } coll . insert ( queue ) ; return 201 ; } param { string } account param { string } queue param { Context } context returns { Promise < void > } memberof public async deleteQueue ( account , queue , context ? ) { const coll = this . db . getCollection ( this . QUEUES_COLLECTION ) ; const doc = coll . findOne ( { accountName : account , name : queue } ) ; if ( ! doc ) { const requestId = context ? context . contextID : undefined ; throw StorageErrorFactory . getQueueNotFound ( requestId ) ; } coll . remove ( doc ) ; const messageColl = this . db . getCollection ( this . MESSAGES_COLLECTION ) ; messageColl . findAndRemove ( { accountName : account , queueName : queue } ) ; } param { string } account param { string } queue param { QueueACL } queueACL param { Context } context returns { Promise < void > } memberof public async setQueueACL ( account , queue , queueACL ? , context ? ) { const coll = this . db . getCollection ( this . QUEUES_COLLECTION ) ; const doc = coll . findOne ( { accountName : account , name : queue } ) ; if ( ! doc ) { const requestId = context ? context . contextID : undefined ; throw StorageErrorFactory . getQueueNotFound ( requestId ) ; } doc . queueAcl = queueACL ; coll . update ( doc ) ; } param { string } account param { string } queue param { QueueMetadata } metadata param { string } requestId returns { Promise < void > } memberof public async setQueueMetadata ( account , queue , metadata ? , context ? ) { const coll = this . db . getCollection ( this . QUEUES_COLLECTION ) ; const doc = coll . findOne ( { accountName : account , name : queue } ) ; if ( ! doc ) { const requestId = context ? context . contextID : undefined ; throw StorageErrorFactory . getQueueNotFound ( requestId ) ; } doc . metadata = metadata ; coll . update ( doc ) ; } param { string } account param { string } queue param { string } requestId returns { number } memberof public async getMessagesCount ( account , queue , context ? ) { const queueColl = this . db . getCollection ( this . QUEUES_COLLECTION ) ; const doc = queueColl . findOne ( { accountName : account , name : queue } ) ; if ( ! doc ) { const requestId = context ? context . contextID : undefined ; throw StorageErrorFactory . getQueueNotFound ( requestId ) ; } const coll = this . db . getCollection ( this . MESSAGES_COLLECTION ) ; const query = { accountName : account , queueName : queue } ; const numberOfMessages = coll . count ( query ) ; return numberOfMessages ; } param { MessageModel } message param { Context } context returns { Promise < void > } memberof public async insertMessage ( message , context ? ) { this . checkQueueExist ( message . accountName , message . queueName , context ) ; const saveMessage = this . messageCopy ( message ) ; saveMessage . timeNextVisible = saveMessage . timeNextVisible . getTime ( ) as any ; const coll = this . db . getCollection ( this . MESSAGES_COLLECTION ) ; coll . insert ( saveMessage ) ; } param { string } account param { string } queue param { number } numOfMessages param { Date } queryDate param { Context } context returns { Promise < MessageModel [ ] > } memberof public async peekMessages ( account , queue , numOfMessages ? , queryDate ? , context ? ) { this . checkQueueExist ( account , queue , context ) ; this . clearExpiredMessages ( account , queue , context ) ; const coll = this . db . getCollection ( this . MESSAGES_COLLECTION ) ; const queryTime = queryDate ? queryDate . getTime ( ) : new Date ( ) . getTime ( ) ; const query = { accountName : account , queueName : queue , timeNextVisible : { $lte : queryTime } } ; if ( numOfMessages === undefined ) { numOfMessages = 1 ; } const docs = coll . chain ( ) . find ( query ) . compoundsort ( [ [ "timeNextVisible" , false ] , [ "$loki" , false ] ] ) . limit ( numOfMessages ) . data ( ) ; const res = [ ] as any ; for ( const doc of docs ) { doc . timeNextVisible = new Date ( Number ( doc . timeNextVisible ) ) ; res . push ( this . messageCopy ( doc ) ) ; doc . timeNextVisible = doc . timeNextVisible . getTime ( ) ; } return res ; } param { string } account param { string } queue param { Date } timeNextVisible param { string } popReceipt param { number } numOfMessages param { Date } queryDate param { Context } context returns { Promise < MessageModel [ ] > } memberof public async getMessages ( account , queue , timeNextVisible , popReceipt , numOfMessages ? , queryDate ? , context ? ) { this . checkQueueExist ( account , queue , context ) ; this . clearExpiredMessages ( account , queue , context ) ; const coll = this . db . getCollection ( this . MESSAGES_COLLECTION ) ; const queryTime = queryDate ? queryDate . getTime ( ) : new Date ( ) . getTime ( ) ; const query = { accountName : account , queueName : queue , timeNextVisible : { $lte : queryTime } } ; if ( numOfMessages === undefined || numOfMessages < 1 ) { numOfMessages = 1 ; } const docs = coll . chain ( ) . find ( query ) . compoundsort ( [ [ "timeNextVisible" , false ] , [ "$loki" , false ] ] ) . limit ( numOfMessages ) . data ( ) ; const visibleTimeInMillisecond = timeNextVisible . getTime ( ) ; for ( const doc of docs ) { doc . timeNextVisible = visibleTimeInMillisecond ; doc . popReceipt = popReceipt ; doc . dequeueCount = Number ( doc . dequeueCount ) + 1 ; } coll . update ( docs ) ; const res = [ ] as any ; for ( const doc of docs ) { doc . timeNextVisible = timeNextVisible ; res . push ( this . messageCopy ( doc ) ) ; doc . timeNextVisible = visibleTimeInMillisecond ; } return res ; } param { string } account param { string } queue param { string } messageId param { string } validatingPopReceipt param { Context } context returns { Promise < void > } memberof public async deleteMessage ( account , queue , messageId , validatingPopReceipt , context ? ) { this . checkQueueExist ( account , queue , context ) ; this . clearExpiredMessages ( account , queue , context ) ; const coll = this . db . getCollection ( this . MESSAGES_COLLECTION ) ; const doc = coll . findOne ( { accountName : account , queueName : queue , messageId } ) ; if ( ! doc ) { const requestId = context ? context . contextID : undefined ; throw StorageErrorFactory . getMessageNotFound ( requestId ) ; } if ( doc . popReceipt !== validatingPopReceipt ) { const requestId = context ? context . contextID : undefined ; throw StorageErrorFactory . getPopReceiptMismatch ( requestId ) ; } coll . remove ( doc ) ; } param { MessageUpdateProperties } message param { string } validatingPopReceipt param { Context } context returns { Promise < void > } memberof public async updateMessage ( message , validatingPopReceipt , context ? ) { this . checkQueueExist ( message . accountName , message . queueName , context ) ; this . clearExpiredMessages ( message . accountName , message . queueName , context ) ; const coll = this . db . getCollection ( this . MESSAGES_COLLECTION ) ; const doc = coll . findOne ( { accountName : message . accountName , queueName : message . queueName , messageId : message . messageId } ) ; if ( ! doc ) { const requestId = context ? context . contextID : undefined ; throw StorageErrorFactory . getMessageNotFound ( requestId ) ; } if ( doc . popReceipt !== validatingPopReceipt ) { const requestId = context ? context . contextID : undefined ; throw StorageErrorFactory . getPopReceiptMismatch ( requestId ) ; } doc . popReceipt = message . popReceipt ; doc . timeNextVisible = message . timeNextVisible . getTime ( ) ; if ( message . persistency !== undefined ) { doc . persistency = message . persistency ; } coll . update ( doc ) ; } param { string } account param { string } queue param { Context } context returns { Promise < void > } memberof public async clearMessages ( account , queue , context ? ) { this . checkQueueExist ( account , queue , context ) ; const coll = this . db . getCollection ( this . MESSAGES_COLLECTION ) ; coll . findAndRemove ( { accountName : account , queueName : queue } ) ; } param { number } maxResults param { ( number | undefined ) } marker returns { ( Promise < [ MessageModel [ ] , number | undefined ] > ) } memberof public async listMessages ( maxResults ? , marker ? ) { const coll = this . db . getCollection ( this . MESSAGES_COLLECTION ) ; const query = { $loki : { $gt : marker } } ; if ( maxResults === undefined ) { maxResults = 5000 ; } const docs = coll . chain ( ) . find ( query ) . limit ( maxResults ) . data ( ) ; if ( docs . length < maxResults ) { return [ docs , undefined ] ; } const nextMarker = docs [ docs . length - 1 ] . $loki ; return [ docs , nextMarker ] ; } public iteratorExtents ( ) { return new QueueReferredExtentsAsyncIterator ( this ) ; } private param { string } account param { string } queue param { Context } context memberof private checkQueueExist ( account , queue , context ? ) { const queueColl = this . db . getCollection ( this . QUEUES_COLLECTION ) ; const queueDoc = queueColl . findOne ( { accountName : account , name : queue } ) ; if ( ! queueDoc ) { const requestId = context ? context . contextID : undefined ; throw StorageErrorFactory . getQueueNotFound ( requestId ) ; } } private param { MessageModel } message returns { MessageModel } memberof private messageCopy ( message ) { const copyMessage = { accountName : message . accountName , queueName : message . queueName , messageId : message . messageId , popReceipt : message . popReceipt , timeNextVisible : new Date ( message . timeNextVisible ) , persistency : { id : message . persistency . id , offset : message . persistency . offset , count : message . persistency . count } , insertionTime : new Date ( message . insertionTime ) , expirationTime : new Date ( message . expirationTime ) , dequeueCount : message . dequeueCount } ; return copyMessage ; } private param { QueueModel } queue returns { QueueModel } memberof private queueCopy ( queue ) { const copyQueue = { accountName : queue . accountName , name : queue . name , queueAcl : queue . queueAcl , metadata : queue . metadata } ; return copyQueue ; } private param { string } regex returns { string } memberof private escapeRegex ( regex ) { return regex . replace ( /[-\/\\^$*+?.()|[\]{}]/g , "\\$&" ) ; } param { string } account param { string } queue param { Context } context returns { Promise < void > } memberof private async clearExpiredMessages ( account , queue , context ? ) { this . checkQueueExist ( account , queue , context ) ; const coll = this . db . getCollection ( this . MESSAGES_COLLECTION ) ; const queryTime = new Date ( ) . getTime ( ) ; const query = { accountName : account , queueName : queue , expirationTime : { $lte : queryTime } } ; const docs = coll . chain ( ) . find ( query ) . data ( ) ; coll . remove ( docs ) ; } }