[Перевод] Использование Slack для отслеживания очереди недоставленных сообщений SQS28.12.2020 18:17

service: slack-sqs-monitor
frameworkVersion: "2.9.0"
provider:
  name: aws
  versionFunctions: false
  runtime: nodejs12.x
  region: ap-southeast-2
  iamRoleStatements:
    - Effect: "Allow"
      Action:
        # You should only give least permissions to your functions.
        - "sqs:*"
      Resource:
        - arn:aws:sqs:ap-southeast-2:xxxxxxxx:sqs.fifo # The original SQS arn
        - arn:aws:sqs:ap-southeast-2:xxxxxxxx:deadletter.fifo # The dead letter queue arn

plugins:
  - serverless-webpack
  - serverless-domain-manager

custom:
  customDomain:
    rest:
      domainName: labs.mianio.com
      basePath: sqs-command
      createRoute53Record: true
      securityPolicy: tls_1_2
  webpack:
    webpackConfig: "webpack.config.js"
    packager: "yarn"

functions:
  monitor:
    handler: functions/monitor.handler
    desciption: The function has the dead letter queue as the event, and forward the event to Slack
    tags:
      name: Monitor
    environment:
#     This is the webhook URL from the previous step
      SLACK_ENDPOINT: https://hooks.slack.com/services/XXXXXXX/XXXXXX/XXXXXXXXX
    events:
      - sqs:
#       Dead letter queue ARN
          arn: arn:aws:sqs:ap-southeast-2:xxxxxxxx:deadletter.fifo

  command:
    handler: functions/command.handler
    tags:
      name: Command
      desciption: The function handles Slack action and place the message back to the queue
    environment:
    # Credentials should be retrieved from Parameter Store 
      SLACK_SIGNING_SECRET: ${ssm:/deadletter/slack/signing-secret~true}
      SLACK_OAUTH_TOKEN: ${ssm:/deadletter/slack/oauth-token~true}
    events:
      - http:
          path: slack
          method: post
          cors: true

import middy from "@middy/core";
import axios from "axios";
import doNotWaitForEmptyEventLoop from "@middy/do-not-wait-for-empty-event-loop";
export const monitor = async (event: any): Promise => {
  const records = event.Records;
  await Promise.all(
    records.map((record: any) => {
      const messageGroupId = record?.attributes?.MessageGroupId;
      const messageDeduplicationId = record?.attributes?.MessageDeduplicationId;
      const approximateReceiveCount =
        record?.attributes?.ApproximateReceiveCount;

      return axios({
        method: "post",
        url: process.env.SLACK_ENDPOINT,
        data: {
          blocks: [
            {
              type: "section",
              text: {
                type: "mrkdwn",
                text: `*Messsge ID*: ${record.messageId}`,
              },
            },
            {
              type: "section",
              text: {
                type: "mrkdwn",
                text: `*Message Group Id*: ${messageGroupId}`,
              },
            },
            {
              type: "section",
              text: {
                type: "mrkdwn",
                text: `*Message Deduplication Id*: ${messageDeduplicationId}`,
              },
            },
            {
              type: "section",
              text: {
                type: "mrkdwn",
                text: `*Approximate Receive Count*: ${approximateReceiveCount}`,
              },
            },
            {
              type: "section",
              text: {
                type: "mrkdwn",
                text: record.body,
              },
            },
            {
              type: "actions",
              elements: [
                {
                  type: "button",
                  style: "primary",
                  text: {
                    type: "plain_text",
                    text: "Send back",
                  },
                  action_id: "sendback",
                  value: record.body,
                },
              ],
            },
          ],
        },
        headers: {
          "Content-type": "application/json; charset=utf-8",
        },
      });
    })
  );
  return;
};

export const salesforceDeadLetterMonitor = middy(monitor).use(
  doNotWaitForEmptyEventLoop()
);

import { APIGatewayEvent } from "aws-lambda";
import AWS from "aws-sdk";
import qs from "qs";
import axios from "axios";
import middy from "@middy/core";
import doNotWaitForEmptyEventLoop from "@middy/do-not-wait-for-empty-event-loop";
import httpHeaderNormalizer from "@middy/http-header-normalizer";
import httpEventNormalizer from "@middy/http-event-normalizer";
import httpErrorHandler from "@middy/http-error-handler";
import { slackVerifier } from "../../middlewares/slack/verify";

const sqs = new AWS.SQS({ region: "ap-southeast-2" });
const command = async (event: APIGatewayEvent) => {
  if (!event.body) return { statusCode: 200 };
  const requestBody: any = qs.parse(event.body);
  const payload: any = JSON.parse(requestBody.payload);
  let response;

  const action = payload.actions[0];
  if (action.action_id === "sendback") {
    try {
      const sqsPayload = payload.message.blocks.find(
        (block: any) => block.block_id === "payload"
      );
      if (sqsPayload?.text?.text && action?.value) {
        const payload = JSON.parse(sqsPayload.text.text);

        await putBack(payload.jobName, payload.jobData, action.value);

        response = {
          payload: {
            attachments: [
              {
                color: "good",
                text: "Job was sent back",
              },
            ],
            response_type: "in_channel",
          },
        };
      }
    } catch (error) {
      console.error(error);
    }
  }

  if (payload.response_url) {
    await axios({
      method: "post",
      url: payload.response_url,
      data: response.payload,
      headers: {
        "Content-type": "application/json; charset=utf-8",
        Authorization: `Bearer ${process.env.SLACK_OAUTH_TOKEN}`,
      },
    });
  } else if (response && !payload.response_url && response.payload) {
    return {
      body: JSON.stringify(response.payload),
      statusCode: 200,
    };
  } else {
    return {
      statusCode: 200,
    };
  }
};

const putBack = async (name: string, data: any, workerUrl: string) => {
  const params: any = {
    MessageBody: JSON.stringify({ jobName: name, jobData: data }),
    QueueUrl: workerUrl,
  };
  return new Promise((resolve: Function, reject: Function): any => {
    sqs.sendMessage(params, (err: any, data: any): any => {
      if (err) {
        reject(err);
      } else {
        resolve(data);
      }
    });
  });
};

export const handler = middy(command)
  .use(doNotWaitForEmptyEventLoop())
  .use(httpEventNormalizer())
  .use(httpHeaderNormalizer())
  .use(slackVerifier())
  .use(httpErrorHandler());

import crypto from 'crypto';
import qs from 'qs';

export const slackVerifier = () => {
  return {
    before: async (handler: any) => {
      const slackSignature =
        handler.event.headers && handler.event.headers['x-slack-signature'];
      const timestamp =
        handler.event.headers &&
        handler.event.headers['x-slack-request-timestamp'];
      const time = Math.floor(new Date().getTime() / 1000);
      if (Math.abs(time - timestamp) > 300) {
        //  The request timestamp is more than five minutes from local time.
        // It could be a replay attack, so let's ignore it.

        return {
          statusCode: 401,
          body: JSON.stringify('Too old'),
        };
      }
      const body = handler.event.body;
      const sigBasestring = `v0:${timestamp}:${body}`;

      const hash = crypto
        .createHmac('sha256', process.env.SLACK_SIGNING_SECRET)
        .update(sigBasestring, 'utf8')
        .digest('hex');

      const mySignature = `v0=${hash}`;
      if (
        !crypto.timingSafeEqual(
          Buffer.from(mySignature, 'utf8'),
          Buffer.from(slackSignature, 'utf8')
        )
      ) {
        return {
          statusCode: 401,
          body: JSON.stringify('Invalid Signature'),
        };
      }
      return;
    },

    onError: (handler: any) => {
      return handler.callback(null, handler.error);
    },
  };
};
view raw