[MS-754] feat: Imporove logging. WIP logic for sending emails

src/api/rest/saleor/webhooks.ts

@@ -10,7 +10,7 @@ import {
   OrderUpdatedSubscriptionDocument,
 } from "@/graphql/operations/subscriptions/generated";
 import { type WebhookEventTypeAsyncEnum } from "@/graphql/schema";
-import { getJSONFormatHeader } from "@/lib/saleor/apps/utils";
+import { serializePayload } from "@/lib/emails/events/helpers";
 import { verifyWebhookSignature } from "@/lib/saleor/auth";
 import { saleorWebhookHeaders } from "@/lib/saleor/schema";
 import { getJWKSProvider } from "@/providers/jwks";
@@ -32,20 +32,6 @@ export const EVENT_HANDLERS: {
   },
 ];
 
-export const serializePayload = ({
-  data,
-  event,
-}: {
-  data: FastifyRequest["body"];
-  event: Lowercase<WebhookEventTypeAsyncEnum>;
-}) => ({
-  format: getJSONFormatHeader({ name: CONFIG.NAME }),
-  payload: {
-    event,
-    data,
-  },
-});
-
 export const webhooks: FastifyPluginAsync = async (fastify) => {
   await fastify.register(rawBody);
 

src/emails-sender-proxy.ts

@@ -1,9 +1,8 @@
 import http, { type IncomingMessage } from "http";
 
 import { CONFIG } from "@/config";
-import { handler } from "@/emails-sender";
+import { handler, logger } from "@/emails-sender";
 import { getJSONFormatHeader } from "@/lib/saleor/apps/utils";
-import { logger } from "@/providers/logger";
 
 /**
 {
@@ -60,8 +59,13 @@ http
     response.write("OK");
     response.end();
   })
-  .on("error", logger.error)
+  .on("error", (error) => {
+    logger.error("Proxy error.", { error });
+  })
+  .on("clientError", (error) => {
+    logger.error("Proxy client error.", { error });
+  })
   .on("listening", () =>
-    logger.info(`Proxy is listening on port ${CONFIG.PROXY_PORT}`)
+    logger.info(`Proxy is listening on port ${CONFIG.PROXY_PORT}.`)
   )
   .listen({ port: CONFIG.PROXY_PORT, host: "0.0.0.0" });

src/emails-sender.ts

@@ -1,11 +1,35 @@
-// TODO: Mails sender serverless
-import { type Context, type SQSBatchResponse, type SQSEvent } from "aws-lambda";
+import {
+  type Context,
+  type SQSBatchResponse,
+  type SQSEvent,
+  type SQSRecord,
+} from "aws-lambda";
 
 import { CONFIG } from "@/config";
+import { EmailParsePayloadError } from "@/lib/emails/errors";
 import { getEmailProvider } from "@/providers/email";
-import { logger } from "@/providers/logger";
+import { getLogger } from "@/providers/logger";
+import { OrderCreatedEmail } from "@/templates/OrderCreatedEmail";
 
-import { OrderCreatedEmail } from "./templates/OrderCreatedEmail";
+import { type WebhookEventTypeAsyncEnum } from "./graphql/schema";
+import { type SerializedPayload } from "./lib/emails/events/helpers";
+import { getJSONFormatHeader } from "./lib/saleor/apps/utils";
+
+export const logger = getLogger("emails-sender");
+
+const parseRecord = (record: SQSRecord) => {
+  try {
+    // FIXME: Proxy events has invalid format? Test with real data & localstack.
+    const data = JSON.parse((record as any).Body);
+    return data as SerializedPayload;
+  } catch (error) {
+    logger.error("Failed to parse record payload.", { record, error });
+
+    throw new EmailParsePayloadError("Failed to parse record payload.", {
+      cause: { source: error as Error },
+    });
+  }
+};
 
 export const handler = async (event: SQSEvent, context: Context) => {
   const failures: string[] = [];
@@ -16,23 +40,46 @@ export const handler = async (event: SQSEvent, context: Context) => {
     /**
      * Process event
      */
-    logger.info({ message: "Processing record", record });
+    logger.debug("Processing record", { record });
 
-    const sender = getEmailProvider({
-      fromEmail: `piotr.grundas+${CONFIG.NAME}@mirumee.com`,
-      from: CONFIG.RELEASE,
-      toEmail: "[email protected]",
-    });
+    const { format, payload } = parseRecord(record);
 
-    const html = await sender.render({
-      props: {},
-      template: OrderCreatedEmail,
-    });
+    if (format === getJSONFormatHeader({ name: CONFIG.NAME })) {
+      const TEMPLATES_MAP: {
+        [key in Lowercase<WebhookEventTypeAsyncEnum>]?: any;
+      } = {
+        order_created: OrderCreatedEmail,
+      };
+      const template = TEMPLATES_MAP[payload.event];
 
-    await sender.send({
-      html,
-      subject: "Order created",
-    });
+      if (!template) {
+        return logger.warn("Received payload with unhandled template.", {
+          format,
+          payload,
+        });
+      }
+
+      const sender = getEmailProvider({
+        fromEmail: `piotr.grundas+${CONFIG.NAME}@mirumee.com`,
+        from: CONFIG.RELEASE,
+        toEmail: "[email protected]",
+      });
+
+      const html = await sender.render({
+        props: {},
+        template: OrderCreatedEmail,
+      });
+
+      await sender.send({
+        html,
+        subject: "Order created",
+      });
+    } else {
+      return logger.warn("Received payload with unsupported format.", {
+        format,
+        payload,
+      });
+    }
   }
 
   if (failures.length) {

src/lib/api/errorHandler.ts

@@ -4,7 +4,7 @@ import { type FastifyInstance } from "fastify";
 import { JOSEError } from "jose/errors";
 import { ZodError } from "zod";
 
-import { logger } from "@/providers/logger";
+import { logger } from "@/server";
 
 const formatCode = (code: number | string) =>
   code.toString().toUpperCase().replaceAll(" ", "_");

src/lib/emails/errors.ts

@@ -0,0 +1,5 @@
+import { BaseError } from "@/lib/errors";
+
+export class EmailSendError extends BaseError {}
+
+export class EmailParsePayloadError extends BaseError {}

src/lib/emails/events/helpers.ts

@@ -0,0 +1,21 @@
+import { type FastifyRequest } from "fastify";
+
+import { CONFIG } from "@/config";
+import { type WebhookEventTypeAsyncEnum } from "@/graphql/schema";
+import { getJSONFormatHeader } from "@/lib/saleor/apps/utils";
+
+export const serializePayload = ({
+  data,
+  event,
+}: {
+  data: FastifyRequest["body"];
+  event: Lowercase<WebhookEventTypeAsyncEnum>;
+}) => ({
+  format: getJSONFormatHeader({ name: CONFIG.NAME }),
+  payload: {
+    event,
+    data,
+  },
+});
+
+export type SerializedPayload = ReturnType<typeof serializePayload>;

src/lib/emails/providers/awsSESEmailProvider.ts

@@ -1,8 +1,9 @@
-import { type Body, SendEmailCommand, SESClient } from "@aws-sdk/client-ses";
+import { SendEmailCommand, SESClient } from "@aws-sdk/client-ses";
 import { z } from "zod";
 
 import { prepareConfig } from "@/lib/zod/util";
 
+import { EmailSendError } from "../errors";
 import { renderEmail } from "../helpers";
 import { type EmailProviderFactory } from "./types";
 
@@ -47,7 +48,7 @@ export const awsSESEmailProvider: EmailProviderFactory = ({
     const { $metadata } = await client.send(command);
 
     if ($metadata.httpStatusCode !== 200) {
-      throw new Error("Failed to send email.", {
+      throw new EmailSendError("Failed to send email.", {
         cause: {
           statusCode: $metadata.httpStatusCode,
           subject,

src/lib/errors.ts

@@ -2,14 +2,36 @@ import { type FastifyError } from "@fastify/error";
 
 import { type RequireAtLeastOne } from "@/lib/types";
 
+interface BaseErrorOptions {
+  cause?: { source?: Error } & Record<string, unknown>;
+}
+
+export class BaseError extends Error {
+  constructor(message?: string, options?: BaseErrorOptions) {
+    super(message, options);
+    this.name = this.constructor.name;
+
+    if (Error.captureStackTrace) {
+      Error.captureStackTrace(this, BaseError);
+    }
+  }
+}
+
+export class BaseAggregateError extends AggregateError {
+  constructor(errors: Iterable<any>, message?: string) {
+    super(errors, message);
+    this.name = this.constructor.name;
+  }
+}
+
 type WithSource = { source: FastifyError };
 type WithError = RequireAtLeastOne<{
   code: string;
   message: string | object;
   statusCode: number;
 }>;
 
-export class HttpError extends Error {
+export class HttpError extends BaseError {
   /**
    * https://github.com/Microsoft/TypeScript/issues/3841@/issuecomment-1488919713
    */

src/lib/plugins/winstonLoggingPlugin/logger.ts

@@ -1,5 +1,9 @@
 import { type FastifyBaseLogger } from "fastify";
-import winston from "winston";
+import {
+  createLogger as createWinstonLogger,
+  format,
+  transports,
+} from "winston";
 
 import { PLUGIN_CONFIG } from "./config";
 
@@ -11,26 +15,34 @@ export const createLogger = ({
   service: string;
 }) => {
   const formatters = PLUGIN_CONFIG.IS_DEVELOPMENT
-    ? [winston.format.prettyPrint({ colorize: true })]
-    : [winston.format.json()];
+    ? [
+        format.colorize(),
+        format.printf((info) => {
+          const { timestamp, message, level, ...args } = info;
+          return `[${timestamp} ${level}]: ${message}\n${Object.keys(args).length ? JSON.stringify(args, null, 2) : ""}`;
+        }),
+      ]
+    : [format.json()];
 
-  return winston.createLogger({
+  return createWinstonLogger({
     defaultMeta: {
       environment,
       nodeEvn: PLUGIN_CONFIG.NODE_ENV,
       service,
     },
-    format: winston.format.combine(
-      winston.format((info) => {
+    format: format.combine(
+      format((info) => {
         info.level = info.level.toUpperCase();
         return info;
       })(),
-      winston.format.timestamp({
+      format.errors({ stack: true }),
+      format.timestamp({
         format: "DD/MM/YYYY HH:mm:ss",
       }),
       ...formatters
     ),
 
+    handleExceptions: false,
     level: PLUGIN_CONFIG.LOG_LEVEL,
 
     levels: {
@@ -41,7 +53,8 @@ export const createLogger = ({
       trace: 4,
       warn: 2,
     },
-    transports: [new winston.transports.Console()],
+
+    transports: [new transports.Console({ handleExceptions: true })],
   }) as unknown as FastifyBaseLogger;
   /**
    * Fastify defaults to pino.logger and has some problems with fatal & trace type compatibility.

src/lib/plugins/winstonLoggingPlugin/plugin.ts

@@ -3,13 +3,11 @@ import fastifyPlugin from "fastify-plugin";
 
 const plugin: FastifyPluginCallback = (fastify, {}, next) => {
   fastify.addHook("onRequest", (req, reply, done) => {
-    req.log.info({
-      message: {
-        body: req.body,
-        method: req.method,
-        query: req.query,
-        url: req.raw.url,
-      },
+    req.log.info("Incoming request", {
+      body: req.body,
+      method: req.method,
+      query: req.query,
+      url: req.raw.url,
       statusCode: reply.raw.statusCode,
       type: "REQUEST",
     });
@@ -18,11 +16,9 @@ const plugin: FastifyPluginCallback = (fastify, {}, next) => {
   });
 
   fastify.addHook("onResponse", (req, reply, done) => {
-    req.log.info({
-      message: {
-        method: req.method,
-        url: req.raw.url,
-      },
+    req.log.info("Outgoing response", {
+      method: req.method,
+      url: req.raw.url,
       statusCode: reply.raw.statusCode,
       elapsedTime: reply.elapsedTime,
       type: "RESPONSE",

src/providers/logger.ts

@@ -1,7 +1,8 @@
 import { CONFIG } from "@/config";
 import { createLogger } from "@/lib/plugins/winstonLoggingPlugin";
 
-export const logger = createLogger({
-  environment: CONFIG.ENVIRONMENT,
-  service: CONFIG.RELEASE,
-});
+export const getLogger = (service: "emails-sender" | "events-receiver") =>
+  createLogger({
+    environment: CONFIG.ENVIRONMENT,
+    service: `${CONFIG.NAME}/${service}@${CONFIG.VERSION}`,
+  });

src/server.ts

@@ -15,7 +15,10 @@ import { type FastifyPlugin } from "@/lib/plugins/types";
 import UrlForPlugin from "@/lib/plugins/urlForPlugin";
 import UrlPlugin from "@/lib/plugins/urlPlugin";
 import WinstonLoggingPlugin from "@/lib/plugins/winstonLoggingPlugin";
-import { logger } from "@/providers/logger";
+
+import { getLogger } from "./providers/logger";
+
+export const logger = getLogger("emails-sender");
 
 export async function createServer() {
   const registrations = [];