import { stat } from "fs" ; import Loki from "lokijs" ; import NotImplementedError from "../errors/NotImplementedError" ; import StorageErrorFactory from "../errors/StorageErrorFactory" ; import * as Models from "../generated/artifacts/models" ; import Context from "../generated/Context" ; import { Entity , Table } from "../persistence/ITableMetadataStore" ; import { ODATA_TYPE , QUERY_RESULT_MAX_NUM } from "../utils/constants" ; import { getTimestampString } from "../utils/utils" ; import ITableMetadataStore , { TableACL } from "./ITableMetadataStore" ; /** MODELS FOR SERVICE */ interface IServiceAdditionalProperties { accountName : string ; } export type ServicePropertiesModel = Models . TableServiceProperties & IServiceAdditionalProperties ; export default class LokiTableMetadataStore implements ITableMetadataStore { private readonly db : Loki ; private readonly TABLES_COLLECTION = "$TABLES_COLLECTION$" ; private readonly SERVICES_COLLECTION = "$SERVICES_COLLECTION$" ; private initialized : boolean = false ; private closed : boolean = false ; public constructor ( public readonly lokiDBPath ) { this . db = new Loki ( lokiDBPath , { autosave : true , autosaveInterval : 5000 } ) ; } 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 ( ) ; } } ) ; } ) ; if ( this . db . getCollection ( this . TABLES_COLLECTION ) === null ) { this . db . addCollection ( this . TABLES_COLLECTION , { indices : [ "account" , "table" ] } ) ; } let servicePropertiesColl = this . db . getCollection ( this . SERVICES_COLLECTION ) ; if ( servicePropertiesColl === null ) { servicePropertiesColl = this . db . addCollection ( this . SERVICES_COLLECTION , { unique : [ "accountName" ] } ) ; } await new Promise < void > ( ( resolve , reject ) => { this . db . saveDatabase ( ( err ) => { if ( err ) { reject ( err ) ; } else { resolve ( ) ; } } ) ; } ) ; this . initialized = true ; this . closed = false ; } public async close ( ) { await new Promise < void > ( ( resolve , reject ) => { this . db . close ( ( err ) => { if ( err ) { reject ( err ) ; } else { resolve ( ) ; } } ) ; } ) ; this . closed = true ; } public isInitialized ( ) { return this . initialized ; } public isClosed ( ) { return this . closed ; } public async createTable ( context , tableModel ) { const coll = this . db . getCollection ( this . TABLES_COLLECTION ) ; const doc = coll . findOne ( { account : tableModel . account , table : tableModel . table } ) ; if ( doc ) { throw StorageErrorFactory . getTableAlreadyExists ( context ) ; } coll . insert ( tableModel ) ; const tableCollectionName = this . getTableCollectionName ( tableModel . account , tableModel . table ) ; const extentColl = this . db . getCollection ( tableCollectionName ) ; if ( extentColl ) { this . db . removeCollection ( tableCollectionName ) ; } this . db . addCollection ( tableCollectionName , { indices : [ "PartitionKey" , "RowKey" ] } ) ; } public async deleteTable ( context , table , account ) { const coll = this . db . getCollection ( this . TABLES_COLLECTION ) ; const doc = coll . findOne ( { account , table } ) ; if ( doc ) { coll . remove ( doc ) ; } else { throw StorageErrorFactory . ResourceNotFound ( context ) ; } const tableCollectionName = this . getTableCollectionName ( account , table ) ; const tableEntityCollection = this . db . getCollection ( tableCollectionName ) ; if ( tableEntityCollection ) { this . db . removeCollection ( tableCollectionName ) ; } } param { string } account param { string } table param { TableACL } tableACL param { Context } context returns { Promise < void > } memberof public async setTableACL ( account , table , context , tableACL ? ) { const coll = this . db . getCollection ( this . TABLES_COLLECTION ) ; const doc = coll . findOne ( { account , table } ) ; if ( ! doc ) { throw StorageErrorFactory . getTableNotFound ( context ) ; } doc . tableAcl = tableACL ; coll . update ( doc ) ; } public async getTable ( account , table , context ) { const coll = this . db . getCollection ( this . TABLES_COLLECTION ) ; const doc = coll . findOne ( { account , table } ) ; if ( ! doc ) { throw StorageErrorFactory . getTableNotFound ( context ) ; } return doc ; } public async queryTable ( context , account , queryOptions , nextTable ? ) { const coll = this . db . getCollection ( this . TABLES_COLLECTION ) ; const filter = { account } as any ; if ( nextTable ) { filter . table = { $gte : nextTable } ; } let queryWhere ; try { queryWhere = this . generateQueryTableWhereFunction ( queryOptions . filter ) ; } catch ( e ) { throw StorageErrorFactory . getQueryConditionInvalid ( context ) ; } const top = queryOptions . top || 1000 ; const docList = coll . chain ( ) . find ( filter ) . where ( queryWhere ) . simplesort ( "table" ) . limit ( top + 1 ) . data ( ) ; let nextTableName ; if ( docList . length > top ) { const tail = docList . pop ( ) ; nextTableName = tail . table ; } if ( ! docList ) { throw StorageErrorFactory . getEntityNotFound ( context ) ; } return [ docList , nextTableName ] ; } public async insertTableEntity ( context , table , account , entity ) { const tablesCollection = this . db . getCollection ( this . TABLES_COLLECTION ) ; const tableDocument = tablesCollection . findOne ( { account , table } ) ; if ( ! tableDocument ) { throw StorageErrorFactory . getTableNotExist ( context ) ; } const tableEntityCollection = this . db . getCollection ( this . getTableCollectionName ( account , table ) ) ; if ( ! tableEntityCollection ) { throw StorageErrorFactory . getTableNotExist ( context ) ; } const doc = tableEntityCollection . findOne ( { PartitionKey : entity . PartitionKey , RowKey : entity . RowKey } ) ; if ( doc ) { throw StorageErrorFactory . getEntityAlreadyExist ( context ) ; } entity . properties . Timestamp = getTimestampString ( entity . lastModifiedTime ) ; entity . properties [ "Timestamp@odata.type" ] = "Edm.DateTime" ; tableEntityCollection . insert ( entity ) ; return entity ; } public async insertOrUpdateTableEntity ( context , table , account , entity , ifMatch ? ) { if ( ifMatch === undefined ) { const existingEntity = await this . queryTableEntitiesWithPartitionAndRowKey ( context , table , account , entity . PartitionKey , entity . RowKey ) ; if ( existingEntity ) { return this . updateTableEntity ( context , table , account , entity , ifMatch ) ; } else { return this . insertTableEntity ( context , table , account , entity ) ; } } else { return this . updateTableEntity ( context , table , account , entity , ifMatch ) ; } } public async insertOrMergeTableEntity ( context , table , account , entity , ifMatch ? ) { if ( ifMatch === undefined ) { const existingEntity = await this . queryTableEntitiesWithPartitionAndRowKey ( context , table , account , entity . PartitionKey , entity . RowKey ) ; if ( existingEntity ) { return this . mergeTableEntity ( context , table , account , entity , ifMatch ) ; } else { return this . insertTableEntity ( context , table , account , entity ) ; } } else { return this . mergeTableEntity ( context , table , account , entity , ifMatch ) ; } } public async deleteTableEntity ( context , table , account , partitionKey , rowKey , etag ) { const tablesCollection = this . db . getCollection ( this . TABLES_COLLECTION ) ; const tableDocument = tablesCollection . findOne ( { account , table } ) ; if ( ! tableDocument ) { throw StorageErrorFactory . getTableNotExist ( context ) ; } const tableEntityCollection = this . db . getCollection ( this . getTableCollectionName ( account , table ) ) ; if ( ! tableEntityCollection ) { throw StorageErrorFactory . getTableNotExist ( context ) ; } if ( partitionKey !== undefined && rowKey !== undefined ) { const doc = tableEntityCollection . findOne ( { PartitionKey : partitionKey , RowKey : rowKey } ) as Entity ; if ( ! doc ) { throw StorageErrorFactory . getEntityNotFound ( context ) ; } else { if ( etag !== "*" && doc . eTag !== etag ) { throw StorageErrorFactory . getPreconditionFailed ( context ) ; } } tableEntityCollection . remove ( doc ) ; return ; } throw StorageErrorFactory . getPropertiesNeedValue ( context ) ; } public async queryTableEntities ( context , account , table , queryOptions , nextPartitionKey ? , nextRowKey ? ) { const tablesCollection = this . db . getCollection ( this . TABLES_COLLECTION ) ; const tableDocument = tablesCollection . findOne ( { account , table } ) ; if ( ! tableDocument ) { throw StorageErrorFactory . getTableNotExist ( context ) ; } const tableEntityCollection = this . db . getCollection ( this . getTableCollectionName ( account , table ) ) ; if ( ! tableEntityCollection ) { throw StorageErrorFactory . getTableNotExist ( context ) ; } let queryWhere ; try { queryWhere = this . generateQueryEntityWhereFunction ( queryOptions . filter ) ; } catch ( e ) { throw StorageErrorFactory . getQueryConditionInvalid ( context ) ; } const maxResults = queryOptions . top || QUERY_RESULT_MAX_NUM ; const decodedNextPartitionKey = this . decodeContinuationHeader ( nextPartitionKey ) ; const decodedNextRowKey = this . decodeContinuationHeader ( nextRowKey ) ; const result = tableEntityCollection . chain ( ) . where ( queryWhere ) . where ( ( data ) => { if ( decodedNextPartitionKey !== undefined ) { if ( data . PartitionKey > decodedNextPartitionKey ) { return true ; } } if ( decodedNextRowKey !== undefined ) { if ( data . RowKey >= decodedNextRowKey && ( data . PartitionKey === decodedNextPartitionKey || data . PartitionKey === undefined ) ) { return true ; } return false ; } return true ; } ) . sort ( ( obj1 , obj2 ) => { if ( obj1 . PartitionKey > obj2 . PartitionKey ) { return 1 ; } else if ( obj1 . PartitionKey === obj2 . PartitionKey ) { if ( obj1 . RowKey > obj2 . RowKey ) { return 1 ; } else if ( obj1 . RowKey === obj2 . RowKey ) { return 0 ; } else { return - 1 ; } } else { return - 1 ; } } ) . limit ( maxResults + 1 ) . data ( ) ; let nextPartitionKeyResponse ; let nextRowKeyResponse ; if ( result . length > maxResults ) { const tail = result . pop ( ) ; nextPartitionKeyResponse = this . encodeContinuationHeader ( tail . PartitionKey ) ; nextRowKeyResponse = this . encodeContinuationHeader ( tail . RowKey ) ; } return [ result , nextPartitionKeyResponse , nextRowKeyResponse ] ; } private decodeContinuationHeader ( input ? ) { if ( input !== undefined ) { return Buffer . from ( input , 'base64' ) . toString ( 'utf8' ) ; } } private encodeContinuationHeader ( input ? ) { if ( input !== undefined ) { return Buffer . from ( input , 'utf8' ) . toString ( 'base64' ) ; } } public async queryTableEntitiesWithPartitionAndRowKey ( context , table , account , partitionKey , rowKey ) { const tableColl = this . db . getCollection ( this . getTableCollectionName ( account , table ) ) ; if ( ! tableColl ) { throw StorageErrorFactory . getTableNotExist ( context ) ; } const requestedDoc = tableColl . findOne ( { PartitionKey : partitionKey , RowKey : rowKey } ) as Entity ; return requestedDoc ; } public async getTableAccessPolicy ( context , table ) { throw new NotImplementedError ( context ) ; } public async setTableAccessPolicy ( context , table ) { throw new NotImplementedError ( context ) ; } private async updateTableEntity ( context , table , account , entity , ifMatch ? ) { const tablesCollection = this . db . getCollection ( this . TABLES_COLLECTION ) ; const tableDocument = tablesCollection . findOne ( { account , table } ) ; if ( ! tableDocument ) { throw StorageErrorFactory . getTableNotExist ( context ) ; } const tableEntityCollection = this . db . getCollection ( this . getTableCollectionName ( account , table ) ) ; if ( ! tableEntityCollection ) { throw StorageErrorFactory . getTableNotExist ( context ) ; } const doc = tableEntityCollection . findOne ( { PartitionKey : entity . PartitionKey , RowKey : entity . RowKey } ) as Entity ; if ( ! doc ) { throw StorageErrorFactory . getEntityNotFound ( context ) ; } const encodedEtag = doc . eTag . replace ( ":" , "%3A" ) . replace ( ":" , "%3A" ) ; let encodedIfMatch ; if ( ifMatch !== undefined ) { encodedIfMatch = ifMatch ! . replace ( ":" , "%3A" ) . replace ( ":" , "%3A" ) ; } if ( encodedIfMatch === undefined || encodedIfMatch === "*" || ( encodedIfMatch !== undefined && encodedEtag === encodedIfMatch ) ) { tableEntityCollection . remove ( doc ) ; entity . properties . Timestamp = getTimestampString ( entity . lastModifiedTime ) ; entity . properties [ "Timestamp@odata.type" ] = "Edm.DateTime" ; tableEntityCollection . insert ( entity ) ; return entity ; } throw StorageErrorFactory . getPreconditionFailed ( context ) ; } private async mergeTableEntity ( context , table , account , entity , ifMatch ? ) { const tablesCollection = this . db . getCollection ( this . TABLES_COLLECTION ) ; const tableDocument = tablesCollection . findOne ( { account , table } ) ; if ( ! tableDocument ) { throw StorageErrorFactory . getTableNotExist ( context ) ; } const tableEntityCollection = this . db . getCollection ( this . getTableCollectionName ( account , table ) ) ; if ( ! tableEntityCollection ) { throw StorageErrorFactory . getTableNotExist ( context ) ; } const doc = tableEntityCollection . findOne ( { PartitionKey : entity . PartitionKey , RowKey : entity . RowKey } ) as Entity ; if ( ! doc ) { throw StorageErrorFactory . getEntityNotFound ( context ) ; } const encodedEtag = doc . eTag . replace ( ":" , "%3A" ) . replace ( ":" , "%3A" ) ; let encodedIfMatch ; if ( ifMatch !== undefined ) { encodedIfMatch = ifMatch ! . replace ( ":" , "%3A" ) . replace ( ":" , "%3A" ) ; } if ( encodedIfMatch === undefined || encodedIfMatch === "*" || ( encodedIfMatch !== undefined && encodedEtag === encodedIfMatch ) ) { const mergedDEntity = { ... doc , ... entity , properties : { ... doc . properties } } ; for ( const key in entity . properties ) { if ( Object . prototype . hasOwnProperty . call ( entity . properties , key ) ) { if ( key . endsWith ( ODATA_TYPE ) ) { continue ; } const value = entity . properties [ key ] ; mergedDEntity . properties [ key ] = value ; if ( entity . properties [ `${ key }${ ODATA_TYPE }` ] !== undefined ) { mergedDEntity . properties [ `${ key }${ ODATA_TYPE }` ] = entity . properties [ `${ key }${ ODATA_TYPE }` ] ; } else { delete mergedDEntity . properties [ `${ key }${ ODATA_TYPE }` ] ; } } } tableEntityCollection . update ( mergedDEntity ) ; return mergedDEntity ; } else { throw StorageErrorFactory . getPreconditionFailed ( context ) ; } } private getTableCollectionName ( account , table ) { return `${ account }$${ table }` ; } private static tokenizeQuery ( originalQuery ) { const query = originalQuery . replace ( /`/g , "\\`" ) ; let tokenStart = 0 ; const tokens = [ ] ; let inString = false ; let i ; function appendToken ( ) { if ( i - tokenStart > 0 ) { let token ; if ( inString ) { token = query . substring ( tokenStart , i ) . replace ( /''/g , "'" ) ; const stringStart = token . indexOf ( "'" ) ; const typePrefix = token . substring ( 0 , stringStart ) ; const backtickString = "`" + token . substring ( typePrefix . length + 1 ) + "`" ; if ( typePrefix === "guid" ) { token = backtickString ; } else { token = typePrefix + backtickString ; } } else { token = convertToken ( query . substring ( tokenStart , i ) ) ; } if ( token ) { tokens . push ( token ) ; } } tokenStart = i + 1 ; } function convertToken ( token ) { switch ( token ) { case "TableName" : return "name" ; case "eq" : return "===" ; case "gt" : return ">" ; case "ge" : return ">=" ; case "lt" : return "<" ; case "le" : return "<=" ; case "ne" : return "!==" ; case "and" : return "&&" ; case "or" : return "||" ; case "not" : return "!" ; default : return token ; } } for ( i = 0 ; i < query . length ; i ++ ) { if ( inString ) { if ( i < query . length - 1 && query [ i ] === "'" && query [ i + 1 ] === "'" ) { i ++ ; continue ; } else if ( query [ i ] === "'" ) { appendToken ( ) ; inString = false ; } } else if ( query [ i ] === "(" || query [ i ] === ")" ) { if ( ( i !== 0 && ( query [ i - 1 ] . match ( /\d/ ) !== null || ( i >= 5 && query . slice ( i - 5 , i ) === " true" ) || ( i >= 6 && query . slice ( i - 6 , i ) === " false" ) ) ) || query . substring ( tokenStart , i ) . match ( /\b[0-9]+L\b/g ) != null ) { appendToken ( ) ; } i -- ; appendToken ( ) ; i ++ ; tokens . push ( query [ i ] ) ; tokenStart ++ ; } else if ( /\s/ . test ( query [ i ] ) ) { appendToken ( ) ; } else if ( query [ i ] === "'" ) { inString = true ; } } appendToken ( ) ; return tokens ; } param query private generateQueryTableWhereFunction ( query ) { if ( query === undefined ) { return ( ) => true ; } const transformedQuery = LokiTableMetadataStore . transformTableQuery ( query ) ; return new Function ( "item" , transformedQuery ) as any ; } /**
     * Azurite V2 query tables implementation.
     */ public static transformTableQuery ( query ) { const systemProperties = new Map < string , string > ( [ [ "name" , "table" ] ] ) ; const allowCustomProperties = false ; return LokiTableMetadataStore . transformQuery ( query , systemProperties , allowCustomProperties ) ; } param query private generateQueryEntityWhereFunction ( query ) { if ( query === undefined ) { return ( ) => true ; } const transformedQuery = LokiTableMetadataStore . transformEntityQuery ( query ) ; return new Function ( "item" , transformedQuery ) as any ; } /**
     * Azurite V2 query entities implementation as temporary workaround before new refactored implementation of querying.
     * TODO: Handle query types
     */ public static transformEntityQuery ( query ) { const systemProperties = new Map < string , string > ( [ [ "PartitionKey" , "PartitionKey" ] , [ "RowKey" , "RowKey" ] ] ) ; const allowCustomProperties = true ; return LokiTableMetadataStore . transformQuery ( query , systemProperties , allowCustomProperties ) ; } private static transformQuery ( query , systemProperties , allowCustomProperties ) { let transformedQuery = "return ( " ; let isOp = false ; let previousIsOp = false ; const tokens = LokiTableMetadataStore . tokenizeQuery ( query ) ; let counter = - 1 ; for ( let token of tokens ) { counter ++ ; if ( token === "" ) { continue ; } previousIsOp = isOp ; isOp = [ "===" , ">" , ">=" , "<" , "<=" , "!==" ] . includes ( token ) ; if ( ! token . match ( /\b\d+/ ) && token !== "true" && token !== "false" && ! token . includes ( "`" ) && ! [ "===" , ">" , ">=" , "<" , "<=" , "!==" , "&&" , "||" , "!" , "(" , ")" ] . includes ( token ) ) { if ( systemProperties . has ( token ) ) { transformedQuery += `item.${ systemProperties . get ( token ) } ` ; } else if ( allowCustomProperties ) { if ( counter + 2 <= tokens . length - 1 && tokens [ counter + 2 ] . startsWith ( "datetime" ) ) { transformedQuery += `new Date(item.properties.${ token }).getTime() ` ; } else { transformedQuery += `item.properties.${ token } ` ; } } else { throw Error ( "Custom properties are not supported on this query type." ) ; } } else { const matchLongInt = token . match ( /\b[0-9]*L\b/g ) ; if ( previousIsOp && matchLongInt !== null && matchLongInt . length === 1 ) { token = token . replace ( /L\b/g , "" ) ; token = "'" + token + "'" ; } else if ( previousIsOp && token . startsWith ( "datetime" ) ) { token = token . replace ( /\bdatetime\b/g , "" ) ; token = `new Date(${ token }).getTime()` ; } else if ( previousIsOp && ( token . startsWith ( "X" ) || token . startsWith ( "binary" ) ) ) { throw Error ( "Binary filter is not supported yet." ) ; } transformedQuery += `${ token } ` ; } } transformedQuery += ")" ; return transformedQuery ; } param { string } account returns { Promise < ServicePropertiesModel | undefined > } memberof public async getServiceProperties ( context , account ) { const coll = this . db . getCollection ( this . SERVICES_COLLECTION ) ; if ( coll ) { const doc = coll . by ( "accountName" , account ) ; return doc ? doc : undefined ; } return undefined ; } param { ServicePropertiesModel } serviceProperties returns { Promise < ServicePropertiesModel > } memberof public async setServiceProperties ( context , serviceProperties ) { const coll = this . db . getCollection ( this . SERVICES_COLLECTION ) ; const doc = coll . by ( "accountName" , serviceProperties . accountName ) ; if ( doc ) { doc . cors = serviceProperties . cors === undefined ? doc . cors : serviceProperties . cors ; doc . hourMetrics = serviceProperties . hourMetrics === undefined ? doc . hourMetrics : serviceProperties . hourMetrics ; doc . logging = serviceProperties . logging === undefined ? doc . logging : serviceProperties . logging ; doc . minuteMetrics = serviceProperties . minuteMetrics === undefined ? doc . minuteMetrics : serviceProperties . minuteMetrics ; return coll . update ( doc ) ; } else { return coll . insert ( serviceProperties ) ; } } }