import { DynamoDBRecord , DynamoDBStreamEvent } from 'aws-lambda' ; import { BoosterConfig , EventEnvelope , Logger , OptimisticConcurrencyUnexpectedVersionError , UUID , } from '@boostercloud/framework-types' ; import { DynamoDB } from 'aws-sdk' ; import { eventsStoreAttributes } from '../constants' ; import { partitionKeyForEvent , partitionKeyForIndexByEntity } from './keys-helper' ; import { Converter } from 'aws-sdk/clients/dynamodb' ; import { retryIfError } from '@boostercloud/framework-common-helpers' ; const originOfTime = new Date ( 0 ) . toISOString ( ) ; export function rawEventsToEnvelopes ( rawEvents ) { return rawEvents . Records . map ( ( record ) : EventEnvelope => { if ( ! record . dynamodb ?. NewImage ) { throw new Error ( 'Received a DynamoDB stream event without "NewImage" field. It is required' ) ; } return Converter . unmarshall ( record . dynamodb ?. NewImage ) as EventEnvelope ; } ) ; } export async function readEntityEventsSince ( dynamoDB , config , logger , entityTypeName , entityID , since ? ) { const fromTime = since ? since : originOfTime ; const result = await dynamoDB . query ( { TableName : config . resourceNames . eventsStore , ConsistentRead : true , KeyConditionExpression : `${ eventsStoreAttributes . partitionKey } = :partitionKey AND ${ eventsStoreAttributes . sortKey } > :fromTime` , ExpressionAttributeValues : { ':partitionKey' : partitionKeyForEvent ( entityTypeName , entityID ) , ':fromTime' : fromTime , } , ScanIndexForward : true , } ) . promise ( ) ; logger . debug ( `[EventsAdapter#readEntityEventsSince] Loaded events for entity ${ entityTypeName } with ID ${ entityID } with result:` , result . Items ) ; return result . Items as Array < EventEnvelope > ; } export async function readEntityLatestSnapshot ( dynamoDB , config , logger , entityTypeName , entityID ) { const result = await dynamoDB . query ( { TableName : config . resourceNames . eventsStore , ConsistentRead : true , KeyConditionExpression : `${ eventsStoreAttributes . partitionKey } = :partitionKey` , ExpressionAttributeValues : { ':partitionKey' : partitionKeyForEvent ( entityTypeName , entityID , 'snapshot' ) , } , ScanIndexForward : false , Limit : 1 , } ) . promise ( ) ; const snapshot = result . Items ?. shift ( ) ; if ( snapshot ) { logger . debug ( `[EventsAdapter#readEntityLatestSnapshot] Snapshot found for entity ${ entityTypeName } with ID ${ entityID }:` , snapshot ) ; return snapshot as EventEnvelope ; } else { logger . debug ( `[EventsAdapter#readEntityLatestSnapshot] No snapshot found for entity ${ entityTypeName } with ID ${ entityID }.` ) ; return null ; } } export async function storeEvents ( dynamoDB , eventEnvelopes , config , logger ) { logger . debug ( '[EventsAdapter#storeEvents] Storing the following event envelopes:' , eventEnvelopes ) ; for ( const eventEnvelope of eventEnvelopes ) { await retryIfError ( logger , ( ) => persistEvent ( dynamoDB , config , eventEnvelope ) , OptimisticConcurrencyUnexpectedVersionError ) ; } logger . debug ( '[EventsAdapter#storeEvents] EventEnvelopes stored' ) ; } async function persistEvent ( dynamoDB , config , eventEnvelope ) { try { const partitionKey = partitionKeyForEvent ( eventEnvelope . entityTypeName , eventEnvelope . entityID , eventEnvelope . kind ) ; const sortKey = new Date ( ) . toISOString ( ) ; await dynamoDB . put ( { TableName : config . resourceNames . eventsStore , ConditionExpression : `${ eventsStoreAttributes . partitionKey } <> :partitionKey AND ${ eventsStoreAttributes . sortKey } <> :sortKey` , ExpressionAttributeValues : { ':partitionKey' : partitionKey , ':sortKey' : sortKey , } , Item : { ... eventEnvelope , [ eventsStoreAttributes . partitionKey ] : partitionKey , [ eventsStoreAttributes . sortKey ] : sortKey , [ eventsStoreAttributes . indexByEntity . partitionKey ] : partitionKeyForIndexByEntity ( eventEnvelope . entityTypeName , eventEnvelope . kind ) , } , } ) . promise ( ) ; } catch ( e ) { if ( e . name == 'ConditionalCheckFailedException' ) { throw new OptimisticConcurrencyUnexpectedVersionError ( e . message ) ; } throw e ; } }