import { BoosterConfig , Logger , OptimisticConcurrencyUnexpectedVersionError , ReadModelEnvelope , ReadModelInterface , ReadOnlyNonEmptyArray , SequenceKey , TimeKey , UUID , } from '@boostercloud/framework-types' ; import { DynamoDBRecord , DynamoDBStreamEvent } from 'aws-lambda' ; import { DynamoDB } from 'aws-sdk' ; import { Converter } from 'aws-sdk/clients/dynamodb' ; import { Arn } from './arn' ; export async function rawReadModelEventsToEnvelopes ( config , logger , rawEvents ) { return rawEvents . Records . map ( toReadModelEnvelope . bind ( null , config ) ) ; } export async function fetchReadModel ( db , config , logger , readModelName , readModelID , sequenceKey ? ) { const sequenceKeyCondition = sequenceKey ?. value ? ` AND #${ sequenceKey . name } = :${ sequenceKey . name }` : '' ; const sequenceAttributeNames = sequenceKey ?. value ? { [ `#${ sequenceKey . name }` ] : sequenceKey . name } : { } ; const sequenceAttributeValues = sequenceKey ?. value ? { [ `:${ sequenceKey . name }` ] : sequenceKey . value } : { } ; logger . debug ( `[ReadModelAdapter#fetchReadModel] Performing query for ${ readModelName }, with ID ${ readModelID } and sequenceKey` , sequenceKey ) ; const queryParams = { TableName : config . resourceNames . forReadModel ( readModelName ) , KeyConditionExpression : '#id = :id' + sequenceKeyCondition , ExpressionAttributeNames : { '#id' : 'id' , ... sequenceAttributeNames , } , ExpressionAttributeValues : { ':id' : readModelID , ... sequenceAttributeValues , } , ConsistentRead : true , } ; const response = await db . query ( queryParams ) . promise ( ) ; logger . debug ( `[ReadModelAdapter#fetchReadModel] Loaded read model ${ readModelName } with ID ${ readModelID } with result:` , response . Items ) ; if ( ! response . Items || response . Items ?. length === 0 ) { return response . Items as unknown as ReadOnlyNonEmptyArray < ReadModelInterface > ; } return ( response . Items as unknown as ReadOnlyNonEmptyArray < ReadModelInterface > ) . map ( ( value ) => { const optimisticConcurrencyValue = generateOptimisticConcurrencyValue ( value ) ; return { ... value , boosterMetadata : { ... value ?. boosterMetadata , optimisticConcurrencyValue , } , } as ReadModelInterface ; } ) as unknown as ReadOnlyNonEmptyArray < ReadModelInterface > ; } export async function storeReadModel ( db , config , logger , readModelName , readModel , expectedCurrentVersion ) { try { await db . put ( { TableName : config . resourceNames . forReadModel ( readModelName ) , Item : readModel , ConditionExpression : 'attribute_not_exists(boosterMetadata.optimisticConcurrencyValue) OR boosterMetadata.optimisticConcurrencyValue = :optimisticConcurrencyValue' , ExpressionAttributeValues : { ':optimisticConcurrencyValue' : expectedCurrentVersion } , } ) . promise ( ) ; } catch ( e ) { if ( e . name == 'ConditionalCheckFailedException' ) { throw new OptimisticConcurrencyUnexpectedVersionError ( e . message ) ; } throw e ; } logger . debug ( '[ReadModelAdapter#storeReadModel] Read model stored' ) ; } export async function deleteReadModel ( db , config , logger , readModelName , readModel ) { const sequenceKeyName = config . readModelSequenceKeys [ readModelName ] ; await db . delete ( { TableName : config . resourceNames . forReadModel ( readModelName ) , Key : buildKey ( readModel . id , sequenceKeyName , readModel [ sequenceKeyName ] ) , } ) . promise ( ) ; logger . debug ( `[ReadModelAdapter#deleteReadModel] Read model deleted. ID = ${ readModel . id }` ) ; } function toReadModelEnvelope ( config , record ) { if ( ! record . dynamodb ?. NewImage || ! record . eventSourceARN ) { throw new Error ( 'Received a DynamoDB stream event without "eventSourceARN" or "NewImage" field. They are required' ) ; } const tableARNComponents = Arn . parse ( record . eventSourceARN ) ; if ( ! tableARNComponents . resourceName ) { throw new Error ( 'Could not extract the table name from the eventSourceARN' ) ; } const readModelTableName = tableARNComponents . resourceName . split ( '/' ) [ 0 ] ; return { typeName : config . readModelNameFromResourceName ( readModelTableName ) , value : Converter . unmarshall ( record . dynamodb . NewImage ) as ReadModelInterface , } ; } function buildKey ( readModelID , sequenceKeyName ? , sequenceKeyValue ? ) { if ( sequenceKeyName && sequenceKeyValue ) return { id : readModelID , [ sequenceKeyName ] : sequenceKeyValue } ; else return { id : readModelID } ; } function generateOptimisticConcurrencyValue ( value ) { return ( value . boosterMetadata ?. version ?? 0 ) + 1 ; }