import { Command , Event , Message , MessageAttributes , MessageAttributeMap } from '@node-ts/bus-messages' ; import { SNS , SQS } from 'aws-sdk' ; import { QueueAttributeMap } from 'aws-sdk/clients/sqs' ; import { Transport , TransportMessage , CoreDependencies , Logger } from '@node-ts/bus-core' ; import { MessageAttributeValue } from 'aws-sdk/clients/sns' ; import { SqsTransportConfiguration } from './sqs-transport-configuration' ; import { generatePolicy } from './generate-policy' ; import { normalizeMessageName , resolveDeadLetterQueueName , resolveQueueArn , resolveQueueUrl , resolveTopicArn as defaultResolveTopicArn , resolveTopicName as defaultResolveTopicName } from './queue-resolvers' ; export const MAX_SQS_DELAY_SECONDS = 900 ; export const MAX_SQS_VISIBILITY_TIMEOUT_SECONDS = 43200 ; const DEFAULT_MESSAGE_RETENTION = 1209600 ; const DEFAULT_VISIBILITY_TIMEOUT = 30 ; const DEFAULT_MAX_RETRY_COUNT = 10 ; const MILLISECONDS_IN_SECONDS = 1000 ; const DEFAULT_WAIT_TIME_SECONDS = 10 ; type Seconds = number ; type Milliseconds = number ; interface MessageRegistry { [ key ] : string ; } /**
 * This is the actual message attribute structure returned by SQS. It doesn't exist in the aws-sdk
 */ export interface SqsMessageAttributes { [ key ] : { Type : string ; Value : string ; } ; } /**
 * The shape of an SNS message has when it's in the body of an SQS message that spawned from that subscription
 */ export interface SQSMessageBody { Message : string ; MessageAttributes : SqsMessageAttributes ; } export class SqsTransport implements Transport < SQS . Message > { /**
     * A registry that tracks what messages have been sent. Sending a message first asserts that the target SNS queue
     * exists, so to avoid doing this each time assertion that the topic is created will only happen once per message.
     */ private registeredMessages : MessageRegistry = { } ; private readonly queueUrl : string ; private readonly queueArn : string ; private readonly deadLetterQueueName : string ; private readonly deadLetterQueueUrl : string ; private readonly deadLetterQueueArn : string ; private coreDependencies : CoreDependencies ; private logger : Logger ; private readonly resolveTopicName : typeof defaultResolveTopicName ; private readonly resolveTopicArn : typeof defaultResolveTopicArn ; param sqsConfiguration param sqs param sns constructor ( private readonly sqsConfiguration , private readonly sqs = new SQS ( ) , private readonly sns = new SNS ( ) ) { this . resolveTopicName = this . sqsConfiguration . resolveTopicName ?? defaultResolveTopicName ; this . resolveTopicArn = this . sqsConfiguration . resolveTopicArn ?? defaultResolveTopicArn ; this . queueUrl = resolveQueueUrl ( sqs . endpoint . href , sqsConfiguration . awsAccountId , sqsConfiguration . queueName ) ; this . queueArn = resolveQueueArn ( sqsConfiguration . awsAccountId , sqsConfiguration . awsRegion , sqsConfiguration . queueName ) ; this . deadLetterQueueName = sqsConfiguration . deadLetterQueueName ? normalizeMessageName ( sqsConfiguration . deadLetterQueueName ) : resolveDeadLetterQueueName ( ) ; this . deadLetterQueueUrl = resolveQueueUrl ( sqs . endpoint . href , sqsConfiguration . awsAccountId , this . deadLetterQueueName ) ; this . deadLetterQueueArn = resolveQueueArn ( sqsConfiguration . awsAccountId , sqsConfiguration . awsRegion , this . deadLetterQueueName ) ; } prepare ( coreDependencies ) { this . coreDependencies = coreDependencies ; this . logger = coreDependencies . loggerFactory ( '@node-ts/bus-sqs:sqs-transport' ) ; } async publish < EventType extends Event > ( event , messageAttributes ? ) { await this . publishMessage ( event , messageAttributes ) ; } async send < CommandType extends Command > ( command , messageAttributes ? ) { await this . publishMessage ( command , messageAttributes ) ; } async fail ( transportMessage ) { await this . sqs . sendMessage ( { QueueUrl : this . deadLetterQueueUrl , MessageBody : transportMessage . raw . Body ! , MessageAttributes : transportMessage . raw . MessageAttributes } ) . promise ( ) ; await this . deleteMessage ( transportMessage ) ; } async readNextMessage ( ) { const receiveRequest = { QueueUrl : this . queueUrl , WaitTimeSeconds : this . sqsConfiguration . waitTimeSeconds || DEFAULT_WAIT_TIME_SECONDS , MaxNumberOfMessages : 1 , MessageAttributeNames : [ '.*' ] , AttributeNames : [ 'ApproximateReceiveCount' ] } ; const result = await this . sqs . receiveMessage ( receiveRequest ) . promise ( ) ; if ( ! result . Messages || result . Messages . length === 0 ) { return undefined ; } if ( result . Messages . length > 1 ) { this . logger . error ( 'Received more than the expected number of messages' , { expected : 1 , received : result . Messages . length } ) ; await Promise . allSettled ( result . Messages . map ( async ( message ) => this . makeMessageVisible ( message ) ) ) ; return undefined ; } const sqsMessage = result . Messages [ 0 ] ; this . logger . debug ( 'Received message from SQS' , { sqsMessage } ) ; try { const snsMessage = JSON . parse ( sqsMessage . Body ! ) as SQSMessageBody ; if ( ! snsMessage . Message ) { this . logger . warn ( 'Message is not formatted with an SNS envelope and will be discarded' , { sqsMessage } ) ; await this . deleteSqsMessage ( sqsMessage ) ; return undefined ; } const attributes = fromMessageAttributeMap ( snsMessage . MessageAttributes ) ; this . logger . debug ( 'Received message attributes' , { transportAttributes : snsMessage . MessageAttributes , messageAttributes : attributes } ) ; const domainMessage = this . coreDependencies . messageSerializer . deserialize ( snsMessage . Message ) ; return { id : sqsMessage . MessageId , raw : sqsMessage , domainMessage , attributes } ; } catch ( error ) { this . logger . warn ( 'Could not parse message. Message will be retried, though it\'s likely to end up in the dead letter queue' , { sqsMessage , error } ) ; await this . makeMessageVisible ( sqsMessage ) ; return undefined ; } } async deleteMessage ( message ) { await this . deleteSqsMessage ( message . raw ) ; } async returnMessage ( message ) { await this . makeMessageVisible ( message . raw ) ; } async initialize ( ) { await this . assertServiceQueue ( ) ; } private async assertServiceQueue ( ) { await this . assertSqsQueue ( this . deadLetterQueueName , { MessageRetentionPeriod : ( this . sqsConfiguration . messageRetentionPeriod || DEFAULT_MESSAGE_RETENTION ) . toString ( ) } ) ; const serviceQueueAttributes = { VisibilityTimeout : `${ this . sqsConfiguration . visibilityTimeout || DEFAULT_VISIBILITY_TIMEOUT }` , RedrivePolicy : JSON . stringify ( { maxReceiveCount : this . sqsConfiguration . maxReceiveCount ?? DEFAULT_MAX_RETRY_COUNT , deadLetterTargetArn : this . deadLetterQueueArn } ) } ; await this . assertSqsQueue ( this . sqsConfiguration . queueName , serviceQueueAttributes ) ; await this . subscribeQueueToMessages ( ) ; await this . attachPolicyToQueue ( this . queueUrl , this . sqsConfiguration . awsAccountId , this . sqsConfiguration . awsRegion ) ; await this . syncQueueAttributes ( this . queueUrl , serviceQueueAttributes ) ; } param message private async assertSnsTopic ( message ) { const messageName = message . $name ; if ( ! this . registeredMessages [ messageName ] ) { const snsTopicName = this . resolveTopicName ( messageName ) ; const snsTopicArn = this . resolveTopicArn ( this . sqsConfiguration . awsAccountId , this . sqsConfiguration . awsRegion , messageName ) ; await this . createSnsTopic ( snsTopicName ) ; this . registeredMessages [ messageName ] = snsTopicArn ; } } /**
     * Asserts that an SQS queue exists
     */ private async assertSqsQueue ( queueName , queueAttributes ? ) { this . logger . info ( 'Asserting sqs queue...' , { queueName , queueAttributes } ) ; const createQueueRequest = { QueueName : queueName , Attributes : queueAttributes } ; try { await this . sqs . createQueue ( createQueueRequest ) . promise ( ) ; } catch ( err ) { const error = err as { code : string ; } ; if ( error . code === 'QueueAlreadyExists' ) { this . logger . trace ( 'Queue already exists' , { queueName } ) ; } else { this . logger . error ( 'SQS queue could not be created' , { queueName , error } ) ; throw err ; } } } private async publishMessage ( message , messageAttributes = { attributes : { } , stickyAttributes : { } } ) { await this . assertSnsTopic ( message ) ; const topicName = this . resolveTopicName ( message . $name ) ; const topicArn = this . resolveTopicArn ( this . sqsConfiguration . awsAccountId , this . sqsConfiguration . awsRegion , topicName ) ; this . logger . trace ( 'Publishing message to sns' , { message , topicArn } ) ; const attributeMap = toMessageAttributeMap ( messageAttributes ) ; this . logger . debug ( 'Resolved message attributes' , { attributeMap } ) ; const snsMessage = { TopicArn : topicArn , Subject : message . $name , Message : this . coreDependencies . messageSerializer . serialize ( message ) , MessageAttributes : attributeMap } ; this . logger . debug ( 'Sending message to SNS' , { snsMessage } ) ; await this . sns . publish ( snsMessage ) . promise ( ) ; } private async subscribeQueueToMessages ( ) { const busManagedTopicArns = await Promise . all ( this . coreDependencies . handlerRegistry . getMessageNames ( ) . map ( messageName => this . resolveTopicName ( messageName ) ) . map ( topicName => this . createSnsTopic ( topicName ) ) ) ; const externallyManagedTopicArns = this . coreDependencies . handlerRegistry . getExternallyManagedTopicIdentifiers ( ) ; await Promise . all ( [ ... busManagedTopicArns , ... externallyManagedTopicArns ] . map ( async ( topicArn ) => { await this . subscribeToTopic ( this . queueArn , topicArn ) ; } ) ) ; } param topicName returns private async createSnsTopic ( topicName ) { this . logger . debug ( 'Attempting to create SNS topic if it doesn\'t exist' , { topicName } ) ; const result = await this . sns . createTopic ( { Name : topicName } ) . promise ( ) ; return result . TopicArn ! ; } private async subscribeToTopic ( queueArn , topicArn ) { const subscribeRequest = { TopicArn : topicArn , Protocol : 'sqs' , Endpoint : queueArn } ; this . logger . info ( 'Subscribing sqs queue to sns topic' , { serviceQueueArn : queueArn , topicArn } ) ; await this . sns . subscribe ( subscribeRequest ) . promise ( ) ; } private async makeMessageVisible ( sqsMessage ) { const changeVisibilityRequest = { QueueUrl : this . queueUrl , ReceiptHandle : sqsMessage . ReceiptHandle ! , VisibilityTimeout : this . calculateVisibilityTimeout ( sqsMessage ) } ; await this . sqs . changeMessageVisibility ( changeVisibilityRequest ) . promise ( ) ; } private async deleteSqsMessage ( sqsMessage ) { const deleteMessageRequest = { QueueUrl : this . queueUrl , ReceiptHandle : sqsMessage . ReceiptHandle ! } ; this . logger . debug ( 'Deleting message from sqs queue' , { deleteMessageRequest } ) ; await this . sqs . deleteMessage ( deleteMessageRequest ) . promise ( ) ; } private async attachPolicyToQueue ( queueUrl , awsAccountId , awsRegion ) { const policy = this . sqsConfiguration . queuePolicy || generatePolicy ( awsAccountId , awsRegion ) ; const setQueuePolicyRequest = { QueueUrl : queueUrl , Attributes : { Policy : policy } } ; this . logger . info ( 'Attaching IAM policy to queue' , { policy , serviceQueueUrl : queueUrl } ) ; await this . sqs . setQueueAttributes ( setQueuePolicyRequest ) . promise ( ) ; } private async syncQueueAttributes ( queueUrl , attributes ) { const existingAttributes = await this . sqs . getQueueAttributes ( { QueueUrl : queueUrl } ) . promise ( ) ; if ( existingAttributes . Attributes !== attributes ) { await this . sqs . setQueueAttributes ( { QueueUrl : queueUrl , Attributes : attributes } ) . promise ( ) ; } } private calculateVisibilityTimeout ( sqsMessage ) { const currentReceiveCount = parseInt ( sqsMessage . Attributes && sqsMessage . Attributes . ApproximateReceiveCount || '0' , 10 ) ; const delay = this . coreDependencies . retryStrategy . calculateRetryDelay ( currentReceiveCount ) ; return delay / MILLISECONDS_IN_SECONDS ; } } export function toMessageAttributeMap ( messageOptions ) { const map = { } ; const toAttributeValue = ( value ) => { const attribute = { DataType : typeof value === 'number' ? 'Number' : typeof value === 'boolean' ? 'Boolean' : 'String' , StringValue : value . toString ( ) } ; return attribute ; } ; if ( messageOptions . attributes ) { Object . keys ( messageOptions . attributes ) . forEach ( key => { const value = messageOptions . attributes [ key ] ; if ( ! ! value ) { map [ `attributes.${ key }` ] = toAttributeValue ( value ) ; } } ) ; } if ( messageOptions . stickyAttributes ) { Object . keys ( messageOptions . stickyAttributes ) . forEach ( key => { const value = messageOptions . stickyAttributes [ key ] ; if ( ! ! value ) { map [ `stickyAttributes.${ key }` ] = toAttributeValue ( value ) ; } } ) ; } if ( messageOptions . correlationId ) { map . correlationId = { DataType : 'String' , StringValue : messageOptions . correlationId } ; } return map ; } export function fromMessageAttributeMap ( sqsAttributes ) { const messageOptions = { attributes : { } , stickyAttributes : { } } ; if ( sqsAttributes ) { messageOptions . correlationId = sqsAttributes . correlationId ? sqsAttributes . correlationId . Value : undefined ; const attributes = { } ; const stickyAttributes = { } ; Object . keys ( sqsAttributes ) . forEach ( key => { let cleansedKey ; if ( key . startsWith ( 'attributes.' ) ) { cleansedKey = key . substr ( 'attributes.' . length ) ; attributes [ cleansedKey ] = getAttributeValue ( sqsAttributes , key ) ; } else if ( key . startsWith ( 'stickyAttributes.' ) ) { cleansedKey = key . substr ( 'stickyAttributes.' . length ) ; stickyAttributes [ cleansedKey ] = getAttributeValue ( sqsAttributes , key ) ; } } ) ; messageOptions . attributes = Object . keys ( attributes ) . length ? attributes : { } ; messageOptions . stickyAttributes = Object . keys ( stickyAttributes ) . length ? stickyAttributes : { } ; } return messageOptions ; } function getAttributeValue ( attributes , key ) { const attribute = attributes [ key ] ; const value = attribute . Type === 'Number' ? Number ( attribute . Value ) : attribute . Type === 'Boolean' ? attribute . Value === 'true' : attribute . Value ; return value ; }