import { BoosterConfig , EventEnvelope , EventFilter , EventInterface , EventSearchResponse , Logger , UUID , } from '@boostercloud/framework-types' ; import { DynamoDB } from 'aws-sdk' ; import { dynamoDbBatchGetLimit , eventsStoreAttributes } from '../constants' ; import { DocumentClient } from 'aws-sdk/clients/dynamodb' ; import { partitionKeyForEvent , partitionKeyForIndexByEntity } from './keys-helper' ; import { inChunksOf } from '../pagination-helpers' ; export async function searchEvents ( dynamoDB , config , logger , filters ) { logger . debug ( 'Initiating an events search. Filters: ' , filters ) ; const timeFilterQuery = buildSearchEventsTimeQuery ( filters . from , filters . to ) ; const eventEnvelopes = await executeSearch ( dynamoDB , config , logger , filters , timeFilterQuery ) ; logger . debug ( 'Events search result: ' , eventEnvelopes ) ; return convertToSearchResult ( eventEnvelopes ) ; } interface TimeQueryData { expression : string ; attributeValues : Record < string , string > ; } function buildSearchEventsTimeQuery ( from ? , to ? ) { let timeQueryData = { expression : '' , attributeValues : { } , } ; if ( from && to ) { timeQueryData = { expression : ` AND ${ eventsStoreAttributes . sortKey } BETWEEN :fromTime AND :toTime` , attributeValues : { ':fromTime' : from , ':toTime' : to , } , } ; } else if ( from ) { timeQueryData = { expression : ` AND ${ eventsStoreAttributes . sortKey } >= :fromTime` , attributeValues : { ':fromTime' : from } , } ; } else if ( to ) { timeQueryData = { expression : ` AND ${ eventsStoreAttributes . sortKey } <= :toTime` , attributeValues : { ':toTime' : to } , } ; } return timeQueryData ; } async function executeSearch ( dynamoDB , config , logger , filters , timeFilterQuery ) { if ( 'entity' in filters ) { if ( filters . entityID ) { return await searchEventsByEntityAndID ( dynamoDB , config , logger , filters . entity , filters . entityID , timeFilterQuery ) ; } else { return await searchEventsByEntity ( dynamoDB , config , logger , filters . entity , timeFilterQuery ) ; } } else if ( 'type' in filters ) { return await searchEventsByType ( dynamoDB , config , logger , filters . type , timeFilterQuery ) ; } else { throw new Error ( 'Invalid search event query. It is neither an search by "entity" nor a search by "type"' ) ; } } async function searchEventsByEntityAndID ( dynamoDB , config , logger , entity , entityID , timeQuery ) { const params = { TableName : config . resourceNames . eventsStore , ConsistentRead : true , ScanIndexForward : false , KeyConditionExpression : `${ eventsStoreAttributes . partitionKey } = :partitionKey${ timeQuery . expression }` , ExpressionAttributeValues : { ... timeQuery . attributeValues , ':partitionKey' : partitionKeyForEvent ( entity , entityID ) , } , } ; logger . debug ( 'Searching events by entity and entity ID. Query params: ' , params ) ; const result = await dynamoDB . query ( params ) . promise ( ) ; return ( result . Items as Array < EventEnvelope > ) ?? [ ] ; } interface EventStoreKeys { [ eventsStoreAttributes . partitionKey ] : string ; [ eventsStoreAttributes . sortKey ] : string ; } async function searchEventsByEntity ( dynamoDB , config , logger , entity , timeQuery ) { const params = { TableName : config . resourceNames . eventsStore , IndexName : eventsStoreAttributes . indexByEntity . name ( config ) , ScanIndexForward : false , KeyConditionExpression : `${ eventsStoreAttributes . indexByEntity . partitionKey } = :partitionKey${ timeQuery . expression }` , ExpressionAttributeValues : { ... timeQuery . attributeValues , ':partitionKey' : partitionKeyForIndexByEntity ( entity , 'event' ) , } , } ; logger . debug ( 'Searching events by entity. Index query params: ' , params ) ; const partialResult = await dynamoDB . query ( params ) . promise ( ) ; const indexRecords = ( partialResult . Items as Array < EventStoreKeys > ) ?? [ ] ; return findEventsDataWithKeys ( dynamoDB , config , logger , indexRecords ) ; } async function searchEventsByType ( dynamoDB , config , logger , type , timeQuery ) { const params = { TableName : config . resourceNames . eventsStore , IndexName : eventsStoreAttributes . indexByType . name ( config ) , ScanIndexForward : false , KeyConditionExpression : `${ eventsStoreAttributes . indexByType . partitionKey } = :partitionKey${ timeQuery . expression }` , ExpressionAttributeValues : { ... timeQuery . attributeValues , ':partitionKey' : type , } , } ; logger . debug ( 'Searching events by type. Index query params: ' , params ) ; const partialResult = await dynamoDB . query ( params ) . promise ( ) ; const indexRecords = ( partialResult . Items as Array < EventStoreKeys > ) ?? [ ] ; return findEventsDataWithKeys ( dynamoDB , config , logger , indexRecords ) ; } async function findEventsDataWithKeys ( dynamoDB , config , logger , keys ) { const result = [ ] ; const keysBatches = inChunksOf ( dynamoDbBatchGetLimit , keys ) ; logger . debug ( `Performing batch get for ${ keysBatches . length } batches` ) ; for ( const keysBatch of keysBatches ) { const batchResult = await performBatchGet ( dynamoDB , config , logger , keysBatch ) ; result . push ( ... batchResult ) ; } return result ; } async function performBatchGet ( dynamoDB , config , logger , keys ) { const params = { RequestItems : { [ config . resourceNames . eventsStore ] : { ConsistentRead : true , Keys : keys . map ( ( record ) => { return { [ eventsStoreAttributes . partitionKey ] : record [ eventsStoreAttributes . partitionKey ] , [ eventsStoreAttributes . sortKey ] : record [ eventsStoreAttributes . sortKey ] , } ; } ) , } , } , } ; logger . debug ( 'Finding events data for keys: ' , keys , params ) ; const result = await dynamoDB . batchGet ( params ) . promise ( ) ; return ( result . Responses ?. [ config . resourceNames . eventsStore ] as Array < EventEnvelope > ) ?? [ ] ; } function convertToSearchResult ( eventEnvelopes ) { return eventEnvelopes . map ( ( eventEnvelope ) => { return { type : eventEnvelope . typeName , entity : eventEnvelope . entityTypeName , entityID : eventEnvelope . entityID , requestID : eventEnvelope . requestID , user : eventEnvelope . currentUser , createdAt : eventEnvelope . createdAt , value : eventEnvelope . value as EventInterface , } ; } ) . sort ( ( a , b ) => { if ( a . createdAt > b . createdAt ) return - 1 ; if ( a . createdAt < b . createdAt ) return 1 ; return 0 ; } ) ; }