Publish message

import { SendMessageCommand, SQSClient } from "@aws-sdk/client-sqs"

/**

 * Publish message

 * Send one message to an SQS queue. A string payload is sent as-is, anything else as JSON; message attribute values are sent as strings. FIFO queues need message_group_id, and message_deduplication_id unless content-based deduplication is on.

 */

export async function main(

  auth: RT.Aws,

  queue_url: string,

  payload: any,

  message_attributes: { [key: string]: string } | undefined,

  message_group_id: string | undefined,

  message_deduplication_id: string | undefined,

  delay_seconds: number | undefined

) {

  if (payload == null || payload === "") {

    throw new Error("payload is required: SQS does not accept an empty message")

  }

  // The client sends to the queue URL's host; the region only signs the request,

  // so take it from the URL when it names one (sqs.<region>.amazonaws.com,

  // <region>.queue.amazonaws.com, VPC endpoint and FIPS hosts).

  const region =

    new URL(queue_url).hostname

      .split(".")

      .find((label) => /^[a-z]{2}(-[a-z]+)+-\d+$/.test(label)) ||

    auth.region ||

    "us-east-1"

  const client = new SQSClient({

    region,

    credentials: {

      accessKeyId: auth.awsAccessKeyId,

      secretAccessKey: auth.awsSecretAccessKey,

    },

  })

  // SQS rejects an empty attribute value: skip those, JSON-encode non-strings.

  const attributes = Object.entries(message_attributes ?? {})

    .filter(([, v]) => v != null && v !== "")

    .map(([k, v]) => [k, typeof v === "string" ? v : JSON.stringify(v)])

  // SDK errors hold cyclic references, which Windmill can't serialize: rethrow plain.

  const { $metadata, ...result } = await client

    .send(

      new SendMessageCommand({

        QueueUrl: queue_url,

        MessageBody:

          typeof payload === "string" ? payload : JSON.stringify(payload),

        MessageAttributes: attributes.length

          ? Object.fromEntries(

              attributes.map(([k, v]) => [

                k,

                { DataType: "String", StringValue: v },

              ])

            )

          : undefined,

        MessageGroupId: message_group_id || undefined,

        MessageDeduplicationId: message_deduplication_id || undefined,

        // 0 is the default, and any per-message value is refused on FIFO queues.

        DelaySeconds: delay_seconds || undefined,

      })

    )

    .catch((e) => {

      throw new Error(`${e.name}: ${e.message}`)

    })

  return result

}