suggestions to improve new SQS handler

johnduffell · johnduffell · commit 0ba9e46e994c · 2026-01-08T21:50:27.000Z
diff --git a/handlers/zuora-callout-apis/src/main/scala/com/gu/autoCancel/AutoCancelHandler.scala b/handlers/zuora-callout-apis/src/main/scala/com/gu/autoCancel/AutoCancelHandler.scala
@@ -44,7 +44,7 @@ object AutoCancelHandler extends App with Logging {
       AutoCancelSteps(
         AutoCancel.apply(zuoraRequest),
         cancelRequestsProducer,
-        ZuoraEmailSteps.sendEmailRegardingAccount(
+        new ZuoraEmailSteps(
           EmailSendSteps(awsSQSSend(EmailQueueName)),
           ZuoraGetInvoiceTransactions(ZuoraRestRequestMaker(response, zuoraRestConfig)),
         ),
diff --git a/handlers/zuora-callout-apis/src/main/scala/com/gu/autoCancel/AutoCancelSqsHandler.scala b/handlers/zuora-callout-apis/src/main/scala/com/gu/autoCancel/AutoCancelSqsHandler.scala
@@ -1,25 +1,35 @@
 package com.gu.autoCancel
 
-import com.amazonaws.services.lambda.runtime.{Context, RequestHandler}
 import com.amazonaws.services.lambda.runtime.events.SQSEvent
-import com.gu.effects.sqs.AwsSQSSend.{EmailQueueName, Payload, QueueName}
+import com.amazonaws.services.lambda.runtime.{Context, RequestHandler}
+import com.gu.autoCancel.GetPaymentData.PaymentFailureInformation
+import com.gu.effects.sqs.AwsSQSSend.EmailQueueName
 import com.gu.effects.sqs.SqsSync
 import com.gu.effects.{GetFromS3, RawEffects}
 import com.gu.util.Logging
-import com.gu.util.apigateway.{ApiGatewayResponse, Auth}
+import com.gu.util.apigateway.ApiGatewayResponse.unauthorized
+import com.gu.util.apigateway.Auth
 import com.gu.util.apigateway.Auth.TrustedApiConfig
-import com.gu.util.config.LoadConfigModule
-import com.gu.util.config.LoadConfigModule.StringFromS3
-import com.gu.util.config.Stage
-import com.gu.util.email.EmailSendSteps
+import com.gu.util.config.{ConfigReads, LoadConfigModule}
+import com.gu.util.email.{EmailId, EmailSendSteps}
 import com.gu.util.reader.Types._
+import com.gu.util.resthttp.RestRequestMaker
 import com.gu.util.zuora._
-import okhttp3.{Request, Response}
 import play.api.libs.json.{Json, Reads}
 
 import java.time.LocalDateTime
 import scala.jdk.CollectionConverters._
-import scala.util.Try
+import scala.util.{Failure, Try}
+
+/** Represents the message format from ApiGatewayToSqs */
+case class ApiGatewayToSqsMessage(
+    queryStringParameters: Map[String, String],
+    body: String,
+)
+
+object ApiGatewayToSqsMessage {
+  implicit val reads: Reads[ApiGatewayToSqsMessage] = Json.reads[ApiGatewayToSqsMessage]
+}
 
 /** Processes auto-cancel requests from SQS. Triggered by an EventSourceMapping with maxConcurrency: 5 to avoid Zuora
   * rate limits.
@@ -29,23 +39,23 @@ import scala.util.Try
   */
 class AutoCancelSqsHandler extends RequestHandler[SQSEvent, Unit] with Logging {
 
-  /** Represents the message format from ApiGatewayToSqs */
-  case class ApiGatewayToSqsMessage(
-      queryStringParameters: Map[String, String],
-      body: String,
-  )
-
-  object ApiGatewayToSqsMessage {
-    implicit val reads: Reads[ApiGatewayToSqsMessage] = Json.reads[ApiGatewayToSqsMessage]
-  }
-
+  // main entry point from AWS lambda
   override def handleRequest(event: SQSEvent, context: Context): Unit = {
     val records = event.getRecords.asScala.toList
     logger.info(s"Processing ${records.size} SQS message(s)")
 
-    val results = records.map(processRecord)
+    val results = for {
+      record <- records
+    } yield {
+      for {
+        parsedCallout <- parseRecord(record)
+        (zuoraCalloutRecord, apiToken) = parsedCallout
+        processor <- ProcessCalloutSteps.build().toTry
+        _ <- processor.execute(zuoraCalloutRecord, apiToken).toTry(())
+      } yield ()
+    }
 
-    val failures = results.collect { case Left(error) => error }
+    val failures = results.collect { case Failure(error) => error }
     if (failures.nonEmpty) {
       logger.error(s"${failures.size} message(s) failed to process")
       // Throw exception to trigger retry/DLQ
@@ -55,84 +65,69 @@ class AutoCancelSqsHandler extends RequestHandler[SQSEvent, Unit] with Logging {
     logger.info(s"Successfully processed ${records.size} message(s)")
   }
 
-  private def processRecord(record: SQSEvent.SQSMessage): Either[String, Unit] = {
+  private def parseRecord(record: SQSEvent.SQSMessage): Try[(AutoCancelCallout, String)] = {
     val messageId = record.getMessageId
-    logger.info(s"Processing SQS message: $messageId")
-
-    Try {
-      val rawBody = record.getBody
-      logger.info(s"Message body: $rawBody")
-
-      // Parse the ApiGatewayToSqs envelope
-      val envelope = Json.parse(rawBody).as[ApiGatewayToSqsMessage]
-      val maybeApiToken = envelope.queryStringParameters.get("apiToken")
-      val calloutBody = envelope.body
-
-      // Parse the actual callout
-      val calloutResult = Json.parse(calloutBody).validate[AutoCancelCallout]
-
-      calloutResult.fold(
-        errors => {
-          val errorMsg = s"Failed to parse callout from message $messageId: $errors"
-          logger.error(errorMsg)
-          Left(errorMsg)
-        },
-        callout => {
-          logger.info(s"Processing auto-cancel for account: ${callout.accountId}, invoice: ${callout.invoiceId}")
-          processCallout(callout, maybeApiToken) match {
-            case Right(_) =>
-              logger.info(s"Successfully processed message $messageId")
-              Right(())
-            case Left(error) =>
-              logger.error(s"Failed to process message $messageId: $error")
-              Left(error)
-          }
-        },
+    val rawBody = record.getBody
+    logger.info(s"$messageId: Message body: $rawBody")
+
+    for {
+      originalApiGatewayEvent <- Try(Json.parse(rawBody).as[ApiGatewayToSqsMessage])
+      apiToken <- originalApiGatewayEvent.queryStringParameters
+        .get("apiToken")
+        .toRight(new RuntimeException("no apiToken header, untrusted content"))
+        .toTry
+      httpRequestBody = originalApiGatewayEvent.body
+      zuoraCalloutRecord <- Try(Json.parse(httpRequestBody).as[AutoCancelCallout])
+      _ = logger.info(
+        s"Processing auto-cancel for account: ${zuoraCalloutRecord.accountId}, invoice: ${zuoraCalloutRecord.invoiceId}",
       )
-    }.toEither.left.map { e =>
-      val errorMsg = s"Exception processing message $messageId: ${e.getMessage}"
-      logger.error(errorMsg, e)
-      errorMsg
-    }.flatten
+    } yield (zuoraCalloutRecord, apiToken)
+
   }
 
-  private def processCallout(
-      callout: AutoCancelCallout,
-      maybeApiToken: Option[String],
-  ): Either[String, Unit] = {
+}
+
+object ProcessCalloutSteps extends Logging {
+
+  def build(): Either[ConfigReads.ConfigFailure, ProcessCalloutSteps] = {
     val stage = RawEffects.stage
     val fetchString = GetFromS3.fetchString _
     val response = RawEffects.response
     val now = RawEffects.now
     val sqsSend = SqsSync.send(SqsSync.buildClient) _
-
-    processCalloutWithEffects(stage, fetchString, response, now, sqsSend)(callout, maybeApiToken) match {
-      case ApiGatewayOp.ContinueProcessing(_) => Right(())
-      case ApiGatewayOp.ReturnWithResponse(resp) =>
-        if (resp.statusCode.startsWith("2")) Right(())
-        else Left(s"Processing returned non-success response: ${resp.statusCode} - ${resp.body}")
-    }
-  }
-
-  private def processCalloutWithEffects(
-      stage: Stage,
-      fetchString: StringFromS3,
-      response: Request => Response,
-      now: () => LocalDateTime,
-      awsSQSSend: QueueName => Payload => Try[Unit],
-  )(callout: AutoCancelCallout, maybeApiToken: Option[String]): ApiGatewayOp[Unit] = {
     val loadConfigModule = LoadConfigModule(stage, fetchString)
 
     for {
-      // Load and validate authentication
-      trustedApiConfig <- loadConfigModule.load[TrustedApiConfig].toApiGatewayOp("load trusted Api config")
-      _ <- validateAuth(trustedApiConfig, maybeApiToken)
-
-      zuoraRestConfig <- loadConfigModule.load[ZuoraRestConfig].toApiGatewayOp("load zuora config")
+      zuoraRestConfig <- loadConfigModule.load[ZuoraRestConfig]
       _ = logger.info(s"Loaded Zuora config for stage: $stage")
 
       zuoraRequest = ZuoraRestRequestMaker(response, zuoraRestConfig)
 
+      zuoraEmailSteps = new ZuoraEmailSteps(
+        EmailSendSteps(sqsSend(EmailQueueName)),
+        ZuoraGetInvoiceTransactions(ZuoraRestRequestMaker(response, zuoraRestConfig)),
+      )
+      trustedApiConfig <- loadConfigModule.load[TrustedApiConfig]
+
+    } yield new ProcessCalloutSteps(zuoraRequest, now, trustedApiConfig, zuoraEmailSteps)
+  }
+
+}
+
+class ProcessCalloutSteps(
+    zuoraRequest: RestRequestMaker.Requests,
+    now: () => LocalDateTime,
+    trustedApiConfig: TrustedApiConfig,
+    zuoraEmailSteps: ZuoraEmailSteps,
+) {
+  def execute(autoCancelCallout: AutoCancelCallout, apiToken: String): ApiGatewayOp[Unit] = {
+
+    for {
+      _ <- Auth
+        .credentialsAreValid(trustedApiConfig, Auth.RequestAuth(Some(apiToken)))
+        .toApiGatewayContinueProcessing(unauthorized)
+        .withLogging("authentication")
+
       cancelRequestsProducer = AutoCancelDataCollectionFilter(
         now().toLocalDate,
         ZuoraGetAccountSummary(zuoraRequest),
@@ -141,64 +136,26 @@ class AutoCancelSqsHandler extends RequestHandler[SQSEvent, Unit] with Logging {
       ) _
 
       // Check if we should process this callout
-      _ <- AutoCancelInputFilter(callout, onlyCancelDirectDebit = false)
+      _ <- AutoCancelInputFilter(autoCancelCallout, onlyCancelDirectDebit = false)
 
       // Get the auto-cancel requests
-      autoCancelRequests <- cancelRequestsProducer(callout).withLogging(
-        s"auto-cancellation requests for ${callout.accountId}",
+      autoCancelRequests <- cancelRequestsProducer(autoCancelCallout).withLogging(
+        s"auto-cancellation requests for ${autoCancelCallout.accountId}",
       )
 
       // Execute the cancellation
-      _ <- AutoCancel
-        .apply(zuoraRequest)(
-          autoCancelRequests,
-          AutoCancelSteps.AutoCancelUrlParams(onlyCancelDirectDebit = false, dryRun = false),
-        )
-        .withLogging(s"auto-cancellation for ${callout.accountId}")
+      _ <- AutoCancel(zuoraRequest)(
+        autoCancelRequests,
+        AutoCancelSteps.AutoCancelUrlParams(onlyCancelDirectDebit = false, dryRun = false),
+      )
+        .withLogging(s"auto-cancellation for ${autoCancelCallout.accountId}")
 
-      // Send email notification
-      _ = sendEmailNotification(callout, response, zuoraRestConfig, awsSQSSend)
+      request = ToMessage(autoCancelCallout, _: PaymentFailureInformation, EmailId.cancelledId)
+      _ <- zuoraEmailSteps
+        .sendEmailRegardingAccount(autoCancelCallout.accountId, request)
+        .toDisjunction
+        .toApiGatewayOp("send email")
 
     } yield ()
   }
-
-  private def validateAuth(trustedApiConfig: TrustedApiConfig, maybeApiToken: Option[String]): ApiGatewayOp[Unit] = {
-    val requestAuth = Auth.RequestAuth(maybeApiToken)
-    if (Auth.credentialsAreValid(trustedApiConfig, requestAuth)) {
-      logger.info("Authentication successful")
-      ApiGatewayOp.ContinueProcessing(())
-    } else {
-      logger.warn("Authentication failed: invalid or missing apiToken")
-      ApiGatewayOp.ReturnWithResponse(ApiGatewayResponse.unauthorized)
-    }
-  }
-
-  private def sendEmailNotification(
-      callout: AutoCancelCallout,
-      response: Request => Response,
-      zuoraRestConfig: ZuoraRestConfig,
-      awsSQSSend: QueueName => Payload => Try[Unit],
-  ): Unit = {
-    try {
-      val toMessageFn: GetPaymentData.PaymentFailureInformation => Either[String, com.gu.util.email.EmailMessage] =
-        paymentInfo => ToMessage(callout, paymentInfo, com.gu.util.email.EmailId.cancelledId)
-
-      val sendEmailResult = ZuoraEmailSteps.sendEmailRegardingAccount(
-        EmailSendSteps(awsSQSSend(EmailQueueName)),
-        ZuoraGetInvoiceTransactions(ZuoraRestRequestMaker(response, zuoraRestConfig)),
-      )(
-        callout.accountId,
-        toMessageFn,
-      )
-      sendEmailResult match {
-        case com.gu.util.resthttp.Types.ClientSuccess(_) =>
-          logger.info(s"Successfully sent cancellation email for account ${callout.accountId}")
-        case failure: com.gu.util.resthttp.Types.ClientFailure =>
-          logger.warn(s"Failed to send cancellation email for account ${callout.accountId}: ${failure.message}")
-      }
-    } catch {
-      case e: Exception =>
-        logger.warn(s"Exception sending cancellation email for account ${callout.accountId}: ${e.getMessage}", e)
-    }
-  }
 }
diff --git a/handlers/zuora-callout-apis/src/main/scala/com/gu/autoCancel/AutoCancelSteps.scala b/handlers/zuora-callout-apis/src/main/scala/com/gu/autoCancel/AutoCancelSteps.scala
@@ -38,10 +38,7 @@ object AutoCancelSteps extends Logging {
   def apply(
       callZuoraAutoCancel: (List[AutoCancelRequest], AutoCancelUrlParams) => ApiGatewayOp[Unit],
       autoCancelReqProducer: AutoCancelCallout => ApiGatewayOp[List[AutoCancelRequest]],
-      sendEmailRegardingAccount: (
-          String,
-          PaymentFailureInformation => Either[String, EmailMessage],
-      ) => ClientFailableOp[Unit],
+      zuoraEmailSteps: ZuoraEmailSteps,
   ): Operation = Operation.noHealthcheck({ apiGatewayRequest: ApiGatewayRequest =>
     (for {
       autoCancelCallout <- apiGatewayRequest.bodyAsCaseClass[AutoCancelCallout]()
@@ -54,7 +51,7 @@ object AutoCancelSteps extends Logging {
         s"auto-cancellation for ${autoCancelCallout.accountId}",
       )
       request = makeRequest(autoCancelCallout) _
-      _ <- handleSendPaymentFailureEmail(autoCancelCallout.accountId, request, sendEmailRegardingAccount, urlParams)
+      _ <- handleSendPaymentFailureEmail(autoCancelCallout.accountId, request, zuoraEmailSteps, urlParams)
     } yield ApiGatewayResponse.successfulExecution).apiResponse
   })
 
@@ -66,17 +63,14 @@ object AutoCancelSteps extends Logging {
   private def handleSendPaymentFailureEmail(
       accountId: String,
       request: PaymentFailureInformation => Either[String, EmailMessage],
-      sendEmailRegardingAccount: (
-          String,
-          PaymentFailureInformation => Either[String, EmailMessage],
-      ) => ClientFailableOp[Unit],
+      zuoraEmailSteps: ZuoraEmailSteps,
       urlParams: AutoCancelUrlParams,
   ) = {
     if (urlParams.dryRun) {
       val msg = "DryRun of SendPaymentFailureEmail"
       logger.info(msg)
       ContinueProcessing(())
-    } else logErrorsAndContinueProcessing(sendEmailRegardingAccount(accountId, request))
+    } else logErrorsAndContinueProcessing(zuoraEmailSteps.sendEmailRegardingAccount(accountId, request))
   }
 
   private def logErrorsAndContinueProcessing(clientFailableOp: ClientFailableOp[Unit]): ContinueProcessing[Unit] =
@@ -88,12 +82,15 @@ object AutoCancelSteps extends Logging {
     }
 }
 
-object ZuoraEmailSteps {
+class ZuoraEmailSteps(
+    sendEmail: EmailMessage => ClientFailableOp[Unit],
+    getInvoiceTransactions: String => ClientFailableOp[InvoiceTransactionSummary],
+) {
 
   def sendEmailRegardingAccount(
-      sendEmail: EmailMessage => ClientFailableOp[Unit],
-      getInvoiceTransactions: String => ClientFailableOp[InvoiceTransactionSummary],
-  )(accountId: String, toMessage: PaymentFailureInformation => Either[String, EmailMessage]): ClientFailableOp[Unit] = {
+      accountId: String,
+      toMessage: PaymentFailureInformation => Either[String, EmailMessage],
+  ): ClientFailableOp[Unit] = {
     for {
       invoiceTransactionSummary <- getInvoiceTransactions(accountId)
       paymentInformation <- GetPaymentData(accountId)(invoiceTransactionSummary).left
diff --git a/lib/handler/src/main/scala/com/gu/util/reader/Types.scala b/lib/handler/src/main/scala/com/gu/util/reader/Types.scala
@@ -15,11 +15,18 @@ object Types extends Logging {
     case class ContinueProcessing[A](a: A) extends ApiGatewayOp[A] {
       override def toDisjunction: Either[ApiResponse, A] = Right(a)
 
+      override def toTry[B](a: B): Try[B] = Success(a)
+
       override def isComplete: Boolean = false
     }
     case class ReturnWithResponse(resp: ApiResponse) extends ApiGatewayOp[Nothing] {
       override def toDisjunction: Either[ApiResponse, Nothing] = Left(resp)
 
+      override def toTry[B](a: B): Try[B] =
+        if (resp.statusCode.startsWith("2")) Success(a)
+        else
+          Failure(new RuntimeException(s"Processing returned non-success response: ${resp.statusCode} - ${resp.body}"))
+
       override def isComplete: Boolean = true
     }
 
@@ -28,6 +35,7 @@ object Types extends Logging {
     def isComplete: Boolean
 
     def toDisjunction: Either[ApiResponse, A]
+    def toTry[B](a: B): Try[B]
 
     def flatMap[B](f: A => ApiGatewayOp[B]): ApiGatewayOp[B] =
       toDisjunction.flatMap(f.andThen(_.toDisjunction)).toApiGatewayOp
