CloudERP Sentinel: Phase 2 Risk Pipeline

Build an AWS pipeline that turns ERP exports into idempotent risk cases.

Introduction

30 Second Summary

A single business export can trigger work across several systems. Repeated deliveries or one broken row can leave the final records unreliable.

In this project, you will deploy CloudERP Sentinel on AWS as a pipeline built around event-driven architecture. You will prove that it creates durable risk cases while handling duplicate deliveries and poison messages safely.

What You'll Build

Picture uploading an ERP export and watching two qualifying transactions appear as durable risk cases in the table.

By the end of this project, you'll have:

  • A live risk pipeline where an ERP CSV uploaded to Amazon S3 becomes two cases in Amazon DynamoDB after Amazon SQS and AWS Lambda process it.
  • A visible idempotency proof where uploading the same export again leaves the table count unchanged.
  • An operational alert trail that shows a poison message in a dead-letter queue and delivers critical case plus queue alarm emails through Amazon SNS.
  • Secret Mission: Correct and redrive the failed event. Then prove that replay leaves ERP#TX-2001 as a single DynamoDB item.

What do I need before starting?

You need an AWS account with approved non-root deployment access plus an email inbox for the notification subscription.

The project includes Windows setup for AWS CLI v2, AWS SAM CLI 1.166.2, Python 3.14.8, and Visual Studio Code.

Before We Start

Before We Start...

Before the hands-on work begins, lock in the CloudERP Sentinel data path from simulated ERP CSV exports in Amazon S3 through Amazon SQS and AWS Lambda to durable Amazon DynamoDB risk cases. The design survives malformed rows through retries plus a dead-letter queue while stable case identifiers suppress duplicates from at-least-once processing.

Prepare the Phase 2 Environment

CloudERP Sentinel will soon become a reproducible serverless deployment. Before its infrastructure can be trusted, your Windows workstation needs authenticated AWS access.

The AWS Command Line Interface gives PowerShell authenticated access to your AWS account. The AWS Serverless Application Model turns the pipeline template into a repeatable deployment.

Python matches the Lambda runtime used by the risk processor. Visual Studio Code keeps the local project in one workspace.

This step prepares your workstation only. It creates no cloud resources.

In this step, get ready to:
  • Prepare AWS CLI v2 with an approved non-root identity.
  • Install the required local tools.
  • Open clouderp-sentinel-phase2 in Visual Studio Code.
Install AWS CLI and secure access

AWS CLI v2 carries your approved identity into later deployment commands. Checking first reveals whether Windows already has the correct major version.

  • Check the installed AWS CLI version by running this command:
aws --version

What does this check show?

The command prints the installed AWS CLI version. You need AWS CLI v2 for this project.

✔️ I see AWS CLI v2

Your AWS CLI installation already meets the project requirement. Keep this version in place.

ⓧ I see AWS CLI v1

Your current major version is too old for this project. The current-user installer upgrades the command available to PowerShell.

  • Install AWS CLI v2 for your Windows user by running this command:
irm https://awscli.amazonaws.com/v2/install.ps1 | iex

What does this command do?

The official installation script installs AWS CLI v2 for the current Windows user. Administrator access is not required.

  • Close the current PowerShell window.
  • Press the Windows key to open Windows search.
  • Type PowerShell into the search field.
  • Press Enter to open a fresh terminal.
  • Confirm the upgraded version by running this command:
aws --version

What should the new terminal show?

The output should identify AWS CLI v2. A fresh terminal reloads the updated command path.

Still seeing the older version?

  • Close every open PowerShell window before checking again.
  • Restart Windows if the previous command path remains active.
  • Help me diagnose why PowerShell still finds AWS CLI v1 after the v2 installation.

ⓧ Command not found

Windows cannot find an AWS CLI installation. The official current-user installer adds AWS CLI v2 for your account.

  • Install AWS CLI v2 for your Windows user by running this command:
irm https://awscli.amazonaws.com/v2/install.ps1 | iex

What does this command do?

The official installation script downloads AWS CLI v2. It installs the command for the current Windows user.

  • Close the current PowerShell window.
  • Press the Windows key to open Windows search.
  • Type PowerShell into the search field.
  • Press Enter to open a fresh terminal.
  • Confirm the installation by running this command:
aws --version

What should the new terminal show?

The output should identify AWS CLI v2. This proves that PowerShell can find the installed command.

AWS CLI still unavailable?

  • Close every open PowerShell window before checking again.
  • Restart Windows if the new command path has not loaded.
  • Help me diagnose why AWS CLI v2 is unavailable after the current-user installation.

Good progress. PowerShell can now reach the AWS CLI required by the deployment workflow.

Credential setup deserves care. Your approved development identity should use short-lived access or another process authorized by your AWS administrator.

  • Obtain approved development access from your AWS administrator.
  • Follow your organization's AWS CLI sign-in process in PowerShell.
  • Set the approved development identity as the active profile for this PowerShell session.
  • Keep root credentials out of PowerShell.
  • Keep access keys out of clouderp-sentinel-phase2.

Unsure which identity to use?

Pause before entering credentials if your organization has not approved an AWS development identity. Ask your AWS administrator for the supported sign-in method.

Help me prepare questions for my AWS administrator about approved non-root CLI access.

Install the version-aligned tools

The deployment template targets a specific Python runtime. The matching local versions make build results easier to reproduce.

Why align the versions?

AWS SAM CLI builds the application before deployment. Python supplies the runtime expected by the risk processor.

Matching the project versions removes avoidable differences between your workstation and AWS Lambda.

  • Check the installed AWS SAM CLI version by running this command:
sam --version

What does this check show?

The command prints the AWS SAM CLI version available to PowerShell. This project uses 1.166.2.

✔️ I see the required SAM version

AWS SAM CLI 1.166.2 is ready. Keep this installation in place.

ⓧ I see another SAM version

The project is pinned to AWS SAM CLI 1.166.2. Install that release before continuing.

  • Open the official AWS SAM CLI release page.
  • Select AWS_SAM_CLI_64_PY3.msi from the release assets.
  • Run AWS_SAM_CLI_64_PY3.msi from your Downloads folder.
  • Complete the 64-bit Windows installation.
  • Close every open PowerShell window.
  • Open a fresh PowerShell window through Windows search.
  • Confirm the installed version by running this command:
sam --version

What should the version check show?

The output should report AWS SAM CLI 1.166.2. The new terminal reads the command path created by the installer.

SAM version still mismatched?

  • Confirm that you ran AWS_SAM_CLI_64_PY3.msi from the 1.166.2 release.
  • Restart Windows before checking again.
  • Help me diagnose an AWS SAM CLI version mismatch on Windows.

ⓧ Command not found

AWS SAM CLI is not installed or its command path is unavailable. Use the official 64-bit Windows installer for 1.166.2.

  • Open the official AWS SAM CLI release page.
  • Select AWS_SAM_CLI_64_PY3.msi from the release assets.
  • Run AWS_SAM_CLI_64_PY3.msi from your Downloads folder.
  • Complete the 64-bit Windows installation.
  • Close every open PowerShell window.
  • Open a fresh PowerShell window through Windows search.
  • Confirm the installation by running this command:
sam --version

What should the version check show?

The output should report AWS SAM CLI 1.166.2. This confirms that PowerShell can find the installed CLI.

SAM command still unavailable?

  • Close every open PowerShell window before checking again.
  • Restart Windows if the installer path remains unavailable.
  • Help me diagnose why the SAM command is unavailable after installation.

That is one important dependency settled. AWS SAM CLI now matches the deployment workflow used in the next step.

  • Check the installed Python version by running this command:
python --version

What does this check show?

The command prints the Python version available to PowerShell. This project uses Python 3.14.8.

✔️ I see the required Python version

Python 3.14.8 is ready for the local build.

ⓧ I see another Python version

PowerShell currently resolves to a different Python installation. Install Python 3.14.8 for this project.

  • Open the official Python downloads page for Windows.
  • Download the 64-bit Windows installer for Python 3.14.8.
  • Run the Python installer from your Downloads folder.
  • Enable the installer option that adds Python to your command path.
  • Complete the installation.
  • Close every open PowerShell window.
  • Open a fresh PowerShell window through Windows search.
  • Confirm the installed version by running this command:
python --version

What should the version check show?

The output should report Python 3.14.8. This confirms that PowerShell resolves to the required installation.

Still seeing another Python version?

  • Confirm that Python 3.14.8 completed its installation.
  • Restart Windows to reload the command path.
  • Help me diagnose why PowerShell resolves to the wrong Python version.

ⓧ Command not found

PowerShell cannot find Python. Install Python 3.14.8 from the official Windows download page.

  • Open the official Python downloads page for Windows.
  • Download the 64-bit Windows installer for Python 3.14.8.
  • Run the Python installer from your Downloads folder.
  • Enable the installer option that adds Python to your command path.
  • Complete the installation.
  • Close every open PowerShell window.
  • Open a fresh PowerShell window through Windows search.
  • Confirm the installation by running this command:
python --version

What should the version check show?

The output should report Python 3.14.8. This proves that PowerShell can find the installed runtime.

Python command still unavailable?

  • Confirm that you enabled the installer option for adding Python to your command path.
  • Restart Windows before checking again.
  • Help me diagnose why Python is unavailable after installation.

You now have the same Python release expected by the project. That removes a common source of build differences.

Visual Studio Code provides the workspace used for the project files. Its User setup adds the code command to the Windows path.

  • Press the Windows key to open Windows search.
  • Type Visual Studio Code into the search field.
  • Press Enter to try opening the editor.

✔️ Visual Studio Code opens

Visual Studio Code is installed. Leave the editor open for the workspace setup.

ⓧ Visual Studio Code is not installed

Install the User setup for the current Windows account. This option does not require administrator permission.

  • Open the official Visual Studio Code Windows setup guide.
  • Download the User setup installer.
  • Run the installer from your Downloads folder.
  • Complete the installation using the User setup defaults.
  • Close every open PowerShell window.
  • Press the Windows key to open Windows search.
  • Type Visual Studio Code into the search field.
  • Press Enter to open the editor.

Visual Studio Code does not open?

  • Confirm that the User setup installer completed.
  • Restart Windows before trying the search again.
  • Help me diagnose a Visual Studio Code User setup problem on Windows.

Your editor is ready. The last check brings every tool together inside the project workspace.

Create the workspace and verify access

A fresh PowerShell session reloads every command path changed by the installers. Creating the folder on your Desktop also gives the next step a predictable location.

  • Close every open PowerShell window.
  • Press the Windows key to open Windows search.
  • Type PowerShell into the search field.
  • Press Enter to open a fresh terminal.
  • Move to your Desktop by running this command:
Set-Location -Path "$HOME\Desktop" -PassThru

What does this command do?

PowerShell changes its current location to your Desktop. The pass-through option prints the resulting path so you can verify the move.

You should see a path ending in your Desktop folder.

  • Create the clouderp-sentinel-phase2 folder by running this command:
New-Item -Path ".\clouderp-sentinel-phase2" -ItemType Directory

What does this command do?

PowerShell creates the project folder inside the current Desktop location. The command output lists clouderp-sentinel-phase2 as a directory.

You should see clouderp-sentinel-phase2 in the command output. This confirms that the local workspace now exists.

  • Open the new folder in Visual Studio Code by running this command:
code .\clouderp-sentinel-phase2

What does this command do?

The code command opens the specified folder as a Visual Studio Code workspace. The editor can now keep the upcoming template files in one place.

Visual Studio Code should show clouderp-sentinel-phase2 in the Explorer sidebar. The folder is empty because no project files have been created yet.

The code command is unavailable?

  • Close every open PowerShell window after installing Visual Studio Code.
  • Restart Windows if the User setup path has not loaded.
  • Help me diagnose why the Visual Studio Code command is unavailable in PowerShell.
  • Select Terminal from the Visual Studio Code menu bar.
  • Select New Terminal.
  • Verify every installed tool from this fresh PowerShell terminal by running these commands:
aws --version
sam --version
python --version
code --version

What should these checks prove?

  • The first line reports AWS CLI v2.
  • The second line reports AWS SAM CLI 1.166.2.
  • The third line reports Python 3.14.8.
  • The fourth command prints the installed Visual Studio Code version.

You should now see four version results in the terminal. The prompt should also point to clouderp-sentinel-phase2.

One version check failed?

  • Close Visual Studio Code after completing the relevant installer.
  • Reopen the clouderp-sentinel-phase2 workspace from a fresh PowerShell window.
  • Help me diagnose a missing tool in the Visual Studio Code PowerShell terminal.

Before you test the AWS identity, which account and ARN do you expect the configured profile to return?

  • Verify the active AWS identity by running this command:
aws sts get-caller-identity

What does this identity check prove?

The command returns details about the identity used by AWS CLI. Check that the account matches your intended AWS account.

The ARN must identify your approved development identity. A root identity is unsuitable for this project.

You should see the intended account plus a non-root ARN. That confirms the AWS commands in later steps use the approved identity.

Identity missing or unexpected?

  • Repeat your organization's approved AWS CLI sign-in process.
  • Confirm that the intended development profile is active in this PowerShell session.
  • Stop if the returned ARN identifies the root user.
  • Help me diagnose an unexpected AWS CLI identity without sharing credentials.

That is the setup work complete. Your approved AWS identity and version-aligned workstation are ready for a reproducible deployment.

Your environment is ready. Next, you will build the first S3-to-SQS-to-Lambda pipeline and watch two risk cases reach DynamoDB.

Ingest the First ERP Export

Your Windows environment can reach AWS. The ERP export still needs a reliable path into durable storage.

You will build the first working slice with AWS SAM. This version ingests valid files and deliberately leaves failure isolation and alerting for later steps.

In this step, get ready to:
  • Create the local project files.
  • Define the ingestion stack and risk processor.
  • Deploy the stack and verify two risk cases.
Create the project files
  • Create the project structure on your Desktop by running these commands:
Set-Location ~/Desktop
New-Item -ItemType Directory -Force clouderp-sentinel-phase2 | Out-Null
Set-Location clouderp-sentinel-phase2
New-Item -ItemType Directory -Force src, events | Out-Null
New-Item -ItemType File -Force template.yaml, src/app.py, src/requirements.txt, events/erp-export.csv | Out-Null
  • Confirm the files exist by running:
Get-ChildItem -Recurse | Select-Object FullName

You should see `template.yaml`, the `src` files, and `events/erp-export.csv` under `clouderp-sentinel-phase2`.

Files missing from the list?

Confirm that PowerShell is using your Desktop before rerunning the creation commands.

Help me find the missing project files.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open search.
  • Type `Visual Studio Code`, then press Enter.
  • Click File in the top menu.
  • Click Open Folder.
  • Select the `clouderp-sentinel-phase2` folder on your Desktop.
  • Click Select Folder.
Define the ingestion stack

The template starts with ingestion only. The dead-letter queue exists as a destination, but the source queue does not use it until Step 4.

  • Open `template.yaml` in the Explorer sidebar.
  • Paste this foundation into the empty file:
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: CloudERP Sentinel Phase 2 ingestion pipeline

Parameters:
  ErpBucketName:
    Type: String
    Description: Globally unique lowercase S3 bucket name

Resources:
  ProcessorLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub '/aws/lambda/${AWS::StackName}-risk-processor'
      RetentionInDays: 7

  DeadLetterQueue:
    Type: AWS::SQS::Queue
    Properties:
      MessageRetentionPeriod: 1209600

  ProcessingQueue:
    Type: AWS::SQS::Queue
    Properties:
      MessageRetentionPeriod: 345600
      VisibilityTimeout: 30

What does this foundation define?

  • The parameter receives the globally unique bucket name.
  • The log group keeps processor logs for seven days.
  • The source queue and dead-letter queue begin as separate resources.
  • Append the queue permission directly below `ProcessingQueue`:

  ProcessingQueuePolicy:
    Type: AWS::SQS::QueuePolicy
    Properties:
      Queues:
        - !Ref ProcessingQueue
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Sid: AllowS3EventNotifications
            Effect: Allow
            Principal:
              Service: s3.amazonaws.com
            Action: sqs:SendMessage
            Resource: !GetAtt ProcessingQueue.Arn
            Condition:
              ArnEquals:
                aws:SourceArn: !Sub 'arn:${AWS::Partition}:s3:::${ErpBucketName}'
              StringEquals:
                aws:SourceAccount: !Ref AWS::AccountId
  • Save `template.yaml`.
  • Validate the first two template chunks by running:
sam validate

You should see confirmation that `template.yaml` is valid.

First validation failed?

Check that both resource names align beneath `Resources:` and that every indentation level uses spaces.

Help me diagnose this template error.

  • Append the bucket and table resources at the end of `template.yaml`:

  ErpBucket:
    Type: AWS::S3::Bucket
    DependsOn: ProcessingQueuePolicy
    Properties:
      BucketName: !Ref ErpBucketName
      NotificationConfiguration:
        QueueConfigurations:
          - Event: s3:ObjectCreated:*
            Queue: !GetAtt ProcessingQueue.Arn
            Filter:
              S3Key:
                Rules:
                  - Name: prefix
                    Value: incoming/
                  - Name: suffix
                    Value: .csv

  RiskCasesTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: CaseId
          AttributeType: S
      KeySchema:
        - AttributeName: CaseId
          KeyType: HASH

How does the data path start?

Amazon S3 sends matching CSV object events to Amazon SQS. The table stores each qualifying transaction under a stable `CaseId`.

  • Append the function settings below `RiskCasesTable`:

  RiskProcessor:
    Type: AWS::Serverless::Function
    DependsOn: ProcessorLogGroup
    Properties:
      FunctionName: !Sub '${AWS::StackName}-risk-processor'
      CodeUri: src/
      Handler: app.lambda_handler
      Runtime: python3.14
      MemorySize: 256
      Timeout: 5
      Environment:
        Variables:
          TABLE_NAME: !Ref RiskCasesTable
  • Save `template.yaml`.
  • Validate the bucket, table, and function settings by running:
sam validate

You should see another valid-template result.

  • Append the execution permissions and queue event inside `RiskProcessor.Properties`:
      Policies:
        - Version: '2012-10-17'
          Statement:
            - Sid: PollProcessingQueue
              Effect: Allow
              Action:
                - sqs:ReceiveMessage
                - sqs:DeleteMessage
                - sqs:GetQueueAttributes
              Resource: !GetAtt ProcessingQueue.Arn
            - Sid: ReadErpExports
              Effect: Allow
              Action: s3:GetObject
              Resource: !Sub 'arn:${AWS::Partition}:s3:::${ErpBucketName}/*'
            - Sid: WriteRiskCases
              Effect: Allow
              Action: dynamodb:PutItem
              Resource: !GetAtt RiskCasesTable.Arn
      Events:
        QueueEvent:
          Type: SQS
          Properties:
            Queue: !GetAtt ProcessingQueue.Arn
            BatchSize: 1
            FunctionResponseTypes:
              - ReportBatchItemFailures

Why use a one-record batch?

One queued notification maps to one invocation and one retry counter. This keeps the failure path visible when you introduce a malformed export.

  • Append the stack outputs at the end of `template.yaml`:

Outputs:
  ErpBucketName:
    Value: !Ref ErpBucket
  RiskTableName:
    Value: !Ref RiskCasesTable
  ProcessingQueueUrl:
    Value: !Ref ProcessingQueue
  DeadLetterQueueUrl:
    Value: !Ref DeadLetterQueue
  DeadLetterQueueArn:
    Value: !GetAtt DeadLetterQueue.Arn
  RiskProcessorName:
    Value: !Ref RiskProcessor
  • Save `template.yaml`.
  • Validate the complete ingestion template by running:
sam validate

You should see confirmation that the complete template is valid.

✔️ Awesome, I've got everything!

Your saved template now defines ingestion without enabling redrive or alerts.

ⓧ I'd like to double check the full code

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: CloudERP Sentinel Phase 2 ingestion pipeline

Parameters:
  ErpBucketName:
    Type: String
    Description: Globally unique lowercase S3 bucket name

Resources:
  ProcessorLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub '/aws/lambda/${AWS::StackName}-risk-processor'
      RetentionInDays: 7

  DeadLetterQueue:
    Type: AWS::SQS::Queue
    Properties:
      MessageRetentionPeriod: 1209600

  ProcessingQueue:
    Type: AWS::SQS::Queue
    Properties:
      MessageRetentionPeriod: 345600
      VisibilityTimeout: 30

  ProcessingQueuePolicy:
    Type: AWS::SQS::QueuePolicy
    Properties:
      Queues:
        - !Ref ProcessingQueue
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Sid: AllowS3EventNotifications
            Effect: Allow
            Principal:
              Service: s3.amazonaws.com
            Action: sqs:SendMessage
            Resource: !GetAtt ProcessingQueue.Arn
            Condition:
              ArnEquals:
                aws:SourceArn: !Sub 'arn:${AWS::Partition}:s3:::${ErpBucketName}'
              StringEquals:
                aws:SourceAccount: !Ref AWS::AccountId

  ErpBucket:
    Type: AWS::S3::Bucket
    DependsOn: ProcessingQueuePolicy
    Properties:
      BucketName: !Ref ErpBucketName
      NotificationConfiguration:
        QueueConfigurations:
          - Event: s3:ObjectCreated:*
            Queue: !GetAtt ProcessingQueue.Arn
            Filter:
              S3Key:
                Rules:
                  - Name: prefix
                    Value: incoming/
                  - Name: suffix
                    Value: .csv

  RiskCasesTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: CaseId
          AttributeType: S
      KeySchema:
        - AttributeName: CaseId
          KeyType: HASH

  RiskProcessor:
    Type: AWS::Serverless::Function
    DependsOn: ProcessorLogGroup
    Properties:
      FunctionName: !Sub '${AWS::StackName}-risk-processor'
      CodeUri: src/
      Handler: app.lambda_handler
      Runtime: python3.14
      MemorySize: 256
      Timeout: 5
      Environment:
        Variables:
          TABLE_NAME: !Ref RiskCasesTable
      Policies:
        - Version: '2012-10-17'
          Statement:
            - Sid: PollProcessingQueue
              Effect: Allow
              Action:
                - sqs:ReceiveMessage
                - sqs:DeleteMessage
                - sqs:GetQueueAttributes
              Resource: !GetAtt ProcessingQueue.Arn
            - Sid: ReadErpExports
              Effect: Allow
              Action: s3:GetObject
              Resource: !Sub 'arn:${AWS::Partition}:s3:::${ErpBucketName}/*'
            - Sid: WriteRiskCases
              Effect: Allow
              Action: dynamodb:PutItem
              Resource: !GetAtt RiskCasesTable.Arn
      Events:
        QueueEvent:
          Type: SQS
          Properties:
            Queue: !GetAtt ProcessingQueue.Arn
            BatchSize: 1
            FunctionResponseTypes:
              - ReportBatchItemFailures

Outputs:
  ErpBucketName:
    Value: !Ref ErpBucket
  RiskTableName:
    Value: !Ref RiskCasesTable
  ProcessingQueueUrl:
    Value: !Ref ProcessingQueue
  DeadLetterQueueUrl:
    Value: !Ref DeadLetterQueue
  DeadLetterQueueArn:
    Value: !GetAtt DeadLetterQueue.Arn
  RiskProcessorName:
    Value: !Ref RiskProcessor
Build and deploy the processor

The processor writes each qualifying row immediately. A deterministic key prevents a repeated delivery from replacing an existing case.

  • Open `src/app.py`.
  • Paste the imports and configuration into the empty file:
# Load CSV, queue, and AWS SDK helpers.
import csv
import io
import json
import logging
import os
from decimal import Decimal, InvalidOperation
from urllib.parse import unquote_plus

import boto3
from botocore.exceptions import ClientError

# Create reusable service clients and table configuration.
LOGGER = logging.getLogger()
LOGGER.setLevel("INFO")
S3 = boto3.client("s3")
DYNAMODB = boto3.resource("dynamodb")
TABLE = DYNAMODB.Table(os.environ["TABLE_NAME"])
RISK_THRESHOLD = Decimal("70")
REQUIRED_COLUMNS = {
    "transaction_id", "vendor", "amount", "currency", "risk_score"
}
  • Append the Lambda handler below the configuration:

# Report only the queue records that fail processing.
def lambda_handler(event, context):
    failures = []
    for record in event.get("Records", []):
        message_id = record["messageId"]
        try:
            process_sqs_message(record)
        except Exception as error:
            LOGGER.exception("Message %s failed: %s", message_id, error)
            failures.append({"itemIdentifier": message_id})
    return {"batchItemFailures": failures}
  • Save `src/app.py`.
  • Check the first two Python chunks by running:
python -m py_compile .\src\app.py
if ($LASTEXITCODE -eq 0) { 'Python syntax valid' }

You should see `Python syntax valid`.

  • Append the queue-message parser:

# Resolve the S3 object referenced by each queue message.
def process_sqs_message(record):
    payload = json.loads(record["body"])
    if payload.get("Event") == "s3:TestEvent":
        LOGGER.info("Ignoring S3 test event")
        return

    s3_records = payload.get("Records")
    if not s3_records:
        raise ValueError("SQS message does not contain S3 Records")

    for s3_record in s3_records:
        bucket = s3_record["s3"]["bucket"]["name"]
        key = unquote_plus(s3_record["s3"]["object"]["key"])
        process_export(bucket, key)
  • Append the CSV reader:

# Validate the CSV shape before processing its rows.
def process_export(bucket, key):
    response = S3.get_object(Bucket=bucket, Key=key)
    csv_text = response["Body"].read().decode("utf-8-sig")
    reader = csv.DictReader(io.StringIO(csv_text))
    missing_columns = REQUIRED_COLUMNS - set(reader.fieldnames or [])
    if missing_columns:
        missing = ", ".join(sorted(missing_columns))
        raise ValueError(f"ERP export is missing columns: {missing}")

    for row_number, row in enumerate(reader, start=2):
        process_row(row, bucket, key, row_number)
  • Save `src/app.py`.
  • Check the parser and reader by running:
python -m py_compile .\src\app.py
if ($LASTEXITCODE -eq 0) { 'Python syntax valid' }

You should see `Python syntax valid` again.

  • Append the item builder:

# Build the durable table item from a validated row.
def build_item(row, bucket, key, amount, risk_score):
    transaction_id = row["transaction_id"].strip()
    return {
        "CaseId": f"ERP#{transaction_id}",
        "TransactionId": transaction_id,
        "Vendor": row["vendor"].strip(),
        "Amount": amount,
        "Currency": row["currency"].strip(),
        "RiskScore": risk_score,
        "Severity": "HIGH",
        "Status": "OPEN",
        "SourceBucket": bucket,
        "SourceKey": key,
    }
  • Append the row processor:

# Validate one row and protect its stable case identifier.
def process_row(row, bucket, key, row_number):
    required = [row["transaction_id"], row["vendor"], row["currency"]]
    if not all(value.strip() for value in required):
        raise ValueError(f"Row {row_number} contains an empty required value")

    try:
        amount = Decimal(row["amount"].strip())
        risk_score = Decimal(row["risk_score"].strip())
    except InvalidOperation as error:
        raise ValueError(f"Row {row_number} contains invalid numeric data") from error

    if risk_score < RISK_THRESHOLD:
        LOGGER.info("Skipped transaction %s", row["transaction_id"])
        return

    item = build_item(row, bucket, key, amount, risk_score)
    try:
        TABLE.put_item(
            Item=item,
            ConditionExpression="attribute_not_exists(CaseId)",
        )
    except ClientError as error:
        code = error.response.get("Error", {}).get("Code")
        if code == "ConditionalCheckFailedException":
            LOGGER.info("Suppressed duplicate risk case %s", item["CaseId"])
            return
        raise
    LOGGER.info("Created HIGH risk case %s", item["CaseId"])
  • Save `src/app.py`.
  • Check the complete processor by running:
python -m py_compile .\src\app.py
if ($LASTEXITCODE -eq 0) { 'Python syntax valid' }

You should see `Python syntax valid` for the complete file.

Seeing a Python syntax error?

Use the reported line number to check indentation and quotation marks in `src/app.py`.

Help me repair this syntax error.

✔️ Awesome, I've got everything!

Your processor now compiles and contains the idempotent write path.

ⓧ I'd like to double check the full code

# Load CSV, queue, and AWS SDK helpers.
import csv
import io
import json
import logging
import os
from decimal import Decimal, InvalidOperation
from urllib.parse import unquote_plus

import boto3
from botocore.exceptions import ClientError

# Create reusable service clients and table configuration.
LOGGER = logging.getLogger()
LOGGER.setLevel("INFO")
S3 = boto3.client("s3")
DYNAMODB = boto3.resource("dynamodb")
TABLE = DYNAMODB.Table(os.environ["TABLE_NAME"])
RISK_THRESHOLD = Decimal("70")
REQUIRED_COLUMNS = {
    "transaction_id", "vendor", "amount", "currency", "risk_score"
}

# Report only the queue records that fail processing.
def lambda_handler(event, context):
    failures = []
    for record in event.get("Records", []):
        message_id = record["messageId"]
        try:
            process_sqs_message(record)
        except Exception as error:
            LOGGER.exception("Message %s failed: %s", message_id, error)
            failures.append({"itemIdentifier": message_id})
    return {"batchItemFailures": failures}

# Resolve the S3 object referenced by each queue message.
def process_sqs_message(record):
    payload = json.loads(record["body"])
    if payload.get("Event") == "s3:TestEvent":
        LOGGER.info("Ignoring S3 test event")
        return

    s3_records = payload.get("Records")
    if not s3_records:
        raise ValueError("SQS message does not contain S3 Records")

    for s3_record in s3_records:
        bucket = s3_record["s3"]["bucket"]["name"]
        key = unquote_plus(s3_record["s3"]["object"]["key"])
        process_export(bucket, key)

# Validate the CSV shape before processing its rows.
def process_export(bucket, key):
    response = S3.get_object(Bucket=bucket, Key=key)
    csv_text = response["Body"].read().decode("utf-8-sig")
    reader = csv.DictReader(io.StringIO(csv_text))
    missing_columns = REQUIRED_COLUMNS - set(reader.fieldnames or [])
    if missing_columns:
        missing = ", ".join(sorted(missing_columns))
        raise ValueError(f"ERP export is missing columns: {missing}")

    for row_number, row in enumerate(reader, start=2):
        process_row(row, bucket, key, row_number)

# Build the durable table item from a validated row.
def build_item(row, bucket, key, amount, risk_score):
    transaction_id = row["transaction_id"].strip()
    return {
        "CaseId": f"ERP#{transaction_id}",
        "TransactionId": transaction_id,
        "Vendor": row["vendor"].strip(),
        "Amount": amount,
        "Currency": row["currency"].strip(),
        "RiskScore": risk_score,
        "Severity": "HIGH",
        "Status": "OPEN",
        "SourceBucket": bucket,
        "SourceKey": key,
    }

# Validate one row and protect its stable case identifier.
def process_row(row, bucket, key, row_number):
    required = [row["transaction_id"], row["vendor"], row["currency"]]
    if not all(value.strip() for value in required):
        raise ValueError(f"Row {row_number} contains an empty required value")

    try:
        amount = Decimal(row["amount"].strip())
        risk_score = Decimal(row["risk_score"].strip())
    except InvalidOperation as error:
        raise ValueError(f"Row {row_number} contains invalid numeric data") from error

    if risk_score < RISK_THRESHOLD:
        LOGGER.info("Skipped transaction %s", row["transaction_id"])
        return

    item = build_item(row, bucket, key, amount, risk_score)
    try:
        TABLE.put_item(
            Item=item,
            ConditionExpression="attribute_not_exists(CaseId)",
        )
    except ClientError as error:
        code = error.response.get("Error", {}).get("Code")
        if code == "ConditionalCheckFailedException":
            LOGGER.info("Suppressed duplicate risk case %s", item["CaseId"])
            return
        raise
    LOGGER.info("Created HIGH risk case %s", item["CaseId"])
  • Open `src/requirements.txt`.
  • Pin the deployment dependency by pasting:
boto3==1.43.106
  • Save `src/requirements.txt`.
  • Build the deployable application by running:
sam build

The first build may take a few minutes while AWS SAM installs the dependency. You should see `Build Succeeded`.

Build did not succeed?

Confirm that `template.yaml` points to `src/` and that `src/requirements.txt` contains the pinned Boto3 version.

Help me diagnose this SAM build.

  • Create a unique bucket name from your AWS account and Region by running:
$AccountId = aws sts get-caller-identity --query Account --output text
$Region = aws configure get region
$BucketName = "clouderp-sentinel-$AccountId-$Region"
$BucketName

You should see one lowercase bucket name containing your account number and Region.

  • Start the guided deployment by running:
sam deploy --guided
  • Enter `clouderp-sentinel-phase2` for the stack name.
  • Enter the Region shown in `$Region`.
  • Enter the value shown in `$BucketName` for `ErpBucketName`.
  • Approve IAM role creation.
  • Save the deployment settings to `samconfig.toml`.
  • Approve the proposed deployment.

The deployment can take several minutes. You should see the stack reach a successful completion state and print its outputs.

  • Copy the `RiskTableName` output here: your risk table name.
  • Store the table name for later checks by running:
$TableName = "[[TABLE_NAME="your risk table name"]]"
  • Open `events/erp-export.csv`.
  • Paste the first ERP export:
transaction_id,vendor,amount,currency,risk_score
TX-1001,Contoso,12500.00,USD,82
TX-1002,Northwind Traders,22000.00,USD,97
  • Save `events/erp-export.csv`.
  • Upload the export by running:
aws s3 cp .\events\erp-export.csv s3://$BucketName/incoming/erp-export.csv

You should see a successful upload result.

Before you check the table, do you expect both rows to pass the risk threshold?

  • Check the durable case count by running:
aws dynamodb scan --table-name $TableName --select COUNT --query Count --output text

You should see `2`. That is the first visible proof that one ERP export became two durable risk cases.

Your ingestion path works. Next, you will replay the valid file and expose how malformed data behaves under repeated delivery.

Prove Idempotency and Expose a Poison Message

Your first ERP upload already proved that the pipeline can turn one business file into two durable risk cases in DynamoDB. This step tests what happens when the same work arrives again.

Amazon SQS invokes AWS Lambda with at-least-once delivery, so a message can be processed more than once.

Idempotency keeps repeated work safe. A poison message exposes how malformed data triggers repeated failures.

In this step, get ready to:
  • Replay the valid ERP export and prove that duplicate cases are suppressed.
  • Upload a malformed ERP export that succeeds on one row before failing on another.
  • Watch the failed message retry while the DynamoDB count remains stable.
Replay the valid ERP export

Deterministic CaseId values give every transaction a stable identity. The conditional write accepts a case only when that identity does not already exist.

  • Return to the PowerShell terminal from the previous step.
  • Press the Up Arrow key until the earlier command that uploaded events/erp-export.csv appears.
  • Press Enter to replay the valid export at the same incoming/erp-export.csv key.

Amazon S3 accepts the overwrite at the same object key. Its object-created notification sends the replay into the pipeline.

  • Press the Up Arrow key until the DynamoDB count command from the previous step appears.

Before you run the count, predict whether the second delivery can create two more cases.

  • Press Enter to check the table count.

Good work. The count stays at 2, which proves the replay reached the table without replacing either case.

Why does the count stay at 2?

The processor calls TABLE.put_item with ConditionExpression="attribute_not_exists(CaseId)". DynamoDB rejects the write when the same CaseId already exists.

The processor handles that conditional rejection as a duplicate. The original case remains unchanged.

Did the count increase?

Check that you uploaded the valid export to the existing incoming/erp-export.csv key. A different file with new transaction IDs creates new cases.

Confirm that the command uses the existing table name stored in $TableName.

Help me diagnose an unexpected DynamoDB count after replaying my ERP export.

Create the malformed ERP export

A single export can contain valid business data beside an invalid value. This test file lets you observe which side effects survive when processing stops partway through the file.

  • Return to the Visual Studio Code workspace from the previous step.
  • Select the events folder in the Explorer sidebar.
  • Use the new-file control at the top of the Explorer sidebar to create erp-export-bad.csv.
  • Populate events/erp-export-bad.csv by pasting these rows:
transaction_id,vendor,amount,currency,risk_score
TX-2001,Wide World Importers,15000.00,USD,88
TX-2002,Fabrikam,not-a-number,USD,95

What does this test file contain?

  • Both rows have risk scores above the processing threshold.
  • The TX-2002 row contains not-a-number where the processor expects a numeric amount.
  • Press Ctrl+S to save events/erp-export-bad.csv.
  • Confirm that erp-export-bad.csv is listed beneath the events folder in the Explorer sidebar.

Does the CSV look different?

Confirm that the header has five comma-separated column names. Check that each data row also has five comma-separated values.

Keep not-a-number exactly as shown. That value creates the failure this step needs.

Help me compare my malformed ERP CSV with the required test file.

✔️ Awesome, I've got everything!

The malformed export now matches the test fixture.

ⓧ I'd like to double check the full code

Your complete file should match this reference.

transaction_id,vendor,amount,currency,risk_score
TX-2001,Wide World Importers,15000.00,USD,88
TX-2002,Fabrikam,not-a-number,USD,95

Before you upload the malformed export, predict whether any row can persist when the file contains invalid numeric data.

  • Return to the PowerShell terminal from earlier.
  • Press the Up Arrow key until the earlier valid-export upload command appears.
  • Replace both occurrences of erp-export.csv with erp-export-bad.csv.
  • Press Enter to upload the malformed export to incoming/erp-export-bad.csv.

PowerShell reports a successful upload. The upload itself only confirms that the new object reached the bucket.

  • Press the Up Arrow key until the earlier DynamoDB count command appears.

Before you run the count, predict whether the valid first row has already become durable.

  • Press Enter to check the table count.

You will see 3. The processor created ERP#TX-2001 before the malformed TX-2002 row stopped the delivery.

Why does partial work remain?

Each qualifying row is written as soon as the processor reaches it. The later numeric failure does not roll back the earlier DynamoDB write.

The function returns the SQS messageId inside batchItemFailures. That response keeps the failed message available for another attempt.

Watch the poison message retry

Amazon SQS hides the failed message during its 30-second visibility timeout. The same message becomes available again after that timeout.

This short pause is expected. The next invocation starts reading the same object from its first row.

  • Wait for at least 30 seconds.
  • Press the Up Arrow key until the DynamoDB count command appears.

Before you run the count again, predict whether retrying the first row can create a fourth case.

  • Press Enter to check the table after the retry.

The count remains 3. The retry encounters ERP#TX-2001 again without duplicating it.

Before you inspect the logs, predict which success and failure messages repeat after the visibility timeout.

  • Inspect the last ten minutes of processor logs by running this command:
sam logs -n RiskProcessor --stack-name clouderp-sentinel-phase2 -s '10min ago'

What does this log command show?

  • The -n RiskProcessor option selects the function by its SAM logical ID.
  • The --stack-name clouderp-sentinel-phase2 option selects the deployed stack.
  • The -s '10min ago' option limits the output to recent activity.

You will see duplicate-suppression entries for ERP#TX-1001 and ERP#TX-1002. These entries prove that replaying the valid export caused no new cases.

You will also see repeated failures containing Row 3 contains invalid numeric data. Duplicate suppression for ERP#TX-2001 proves that the malformed message reappeared without recreating its first case.

Only seeing one malformed-export failure?

The next retry may not have reached the logs yet. Wait through another 30-second visibility timeout before running the log command again.

Confirm that the log window still covers the time when you uploaded erp-export-bad.csv.

Help me diagnose missing or incomplete retry evidence in my Lambda logs.

You have proved that deterministic writes protect successful work during replay. Next, you will verify that the poison message leaves normal processing and becomes available for investigation.

Isolate Failures in a Dead-Letter Queue

Your malformed export now retries without duplicating `ERP#TX-2001`. The source queue still has no boundary that removes the poison message from normal processing.

You will connect the existing dead-letter queue to Amazon SQS. A redrive policy moves the message after five receives while the conditional table write protects earlier work.

In this step, get ready to:
  • Add a five-receive redrive boundary.
  • Deploy the changed queue configuration.
  • Verify one isolated message and three durable cases.
Add the redrive policy

The source queue currently retains every failed delivery for another attempt. The redrive policy gives repeated failures a durable investigation path.

  • Open `template.yaml` in Visual Studio Code.
  • Find this existing `ProcessingQueue` block:
  ProcessingQueue:
    Type: AWS::SQS::Queue
    Properties:
      MessageRetentionPeriod: 345600
      VisibilityTimeout: 30
  • Replace that block with this redrive-aware version:
  ProcessingQueue:
    Type: AWS::SQS::Queue
    Properties:
      MessageRetentionPeriod: 345600
      VisibilityTimeout: 30
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt DeadLetterQueue.Arn
        maxReceiveCount: 5

What does the policy change?

  • `maxReceiveCount: 5` sets the retry boundary.
  • `deadLetterTargetArn` identifies the queue that receives exhausted messages.
  • The 14-day DLQ retention remains longer than the four-day source retention.
  • Save `template.yaml`.
  • Validate the updated template by running:
sam validate

You should see confirmation that `template.yaml` is valid.

Redrive policy rejected?

Check that `RedrivePolicy` aligns with `VisibilityTimeout` and that its two properties sit one indentation level deeper.

Help me repair the redrive policy.

Deploy the queue change

The next deployment changes the existing source queue. It leaves the bucket, function, and table in place.

  • Build the updated application by running:
sam build

You should see `Build Succeeded`.

Before you deploy, do you expect AWS CloudFormation to replace the queue or update its policy in place?

  • Deploy the saved configuration by running:
sam deploy

The update can take several minutes while AWS CloudFormation changes the queue. You should see the stack update complete successfully.

Queue update failed?

Confirm that PowerShell is inside `clouderp-sentinel-phase2` and that the first guided deployment created `samconfig.toml`.

Help me diagnose this deployment update.

Verify the isolated message

The malformed message may need one more 30-second visibility cycle before SQS moves it. That short wait is the queue completing its retry boundary.

  • Open the AWS CloudFormation console.
  • Select the `clouderp-sentinel-phase2` stack.
  • Open the Outputs tab.
  • Copy `DeadLetterQueueUrl` here: your dead-letter queue URL.
  • Store the copied URL by running:
$DlqUrl = "[[DLQ_URL="your dead-letter queue URL"]]"
  • Wait for any remaining retry cycle to finish.

Before you check the two counts, do you expect retrying the first valid row to create another case?

  • Check the table and dead-letter queue by running:
aws dynamodb scan --table-name $TableName --select COUNT --query Count --output text
aws sqs get-queue-attributes --queue-url $DlqUrl --attribute-names ApproximateNumberOfMessages --query Attributes.ApproximateNumberOfMessages --output text

What should you see?

  • The DynamoDB count remains `3` because `ERP#TX-2001` already exists.
  • The dead-letter queue count becomes `1` because the malformed notification exhausted its retries.

Dead-letter queue still shows zero?

  • Wait through another 30-second visibility cycle.
  • Confirm that `$DlqUrl` contains the URL rather than the queue ARN.
  • Run the two checks again.

Help me trace the poison message.

The failure boundary now works as a real deployed change. Next, you will add notification resources and prove both business and operational alerts.

Publish Critical Alerts and Package the Evidence

Your dead-letter queue now contains one isolated notification. The table still holds the three valid cases created before and during the failure.

You will add Amazon SNS for email delivery and Amazon CloudWatch for queue monitoring. The notification channel comes first so it is confirmed before either alert can fire.

In this step, get ready to:
  • Deploy and confirm the email notification channel.
  • Trigger one queue alarm and one critical-case alert.
  • Package the verified pipeline in `README.md`.
Create and confirm the notification channel

SNS email subscriptions remain inactive until the recipient confirms them. Deploying this channel first prevents the later alarm from firing before your inbox can receive it.

  • Open `template.yaml`.
  • Add this parameter directly below `ErpBucketName`:
  AlertEmail:
    Type: String
    Description: Email address that confirms the SNS subscription
  • Add the notification resources below `RiskCasesTable`:
  AlertTopic:
    Type: AWS::SNS::Topic

  AlertEmailSubscription:
    Type: AWS::SNS::Subscription
    Properties:
      TopicArn: !Ref AlertTopic
      Protocol: email
      Endpoint: !Ref AlertEmail

How does the channel work?

The topic receives messages from the pipeline. The subscription connects that topic to the address supplied during deployment.

  • Add this output at the end of the existing `Outputs` section:
  AlertTopicArn:
    Value: !Ref AlertTopic
  • Save `template.yaml`.
  • Validate the channel resources by running:
sam validate

You should see confirmation that the template is valid.

  • Deploy the notification channel by running:
sam build
sam deploy --guided
  • Reuse `clouderp-sentinel-phase2` as the stack name.
  • Reuse the Region from the existing deployment.
  • Reuse the existing `ErpBucketName` value.
  • Enter the email address that should receive project alerts.
  • Save the updated values to `samconfig.toml`.
  • Approve the stack update.

The update can take several minutes. You should see the stack finish without a failed resource.

  • Open the inbox for the address supplied during deployment.
  • Open the AWS subscription confirmation message.
  • Select the confirmation link.

You should see confirmation that the subscription is active. The channel is now ready before either alert source exists.

Confirmation email missing?

Check the spam folder and confirm that the deployed address matches the inbox you opened.

Help me find the subscription message.

Connect the two alert sources

The function publishes business alerts for critical cases. The alarm publishes operational alerts when at least one message is visible in the dead-letter queue.

  • In `template.yaml`, find the current function environment block:
      Environment:
        Variables:
          TABLE_NAME: !Ref RiskCasesTable
  • Replace it with this topic-aware environment block:
      Environment:
        Variables:
          TABLE_NAME: !Ref RiskCasesTable
          TOPIC_ARN: !Ref AlertTopic
  • Add this statement after the `WriteRiskCases` statement in the function policy:
            - Sid: PublishCriticalAlerts
              Effect: Allow
              Action: sns:Publish
              Resource: !Ref AlertTopic
  • Add the queue alarm below `RiskProcessor`:
  DeadLetterQueueAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmDescription: A failed ERP event is waiting in the DLQ
      Namespace: AWS/SQS
      MetricName: ApproximateNumberOfMessagesVisible
      Dimensions:
        - Name: QueueName
          Value: !GetAtt DeadLetterQueue.QueueName
      Statistic: Maximum
      Period: 60
      EvaluationPeriods: 1
      Threshold: 1
      ComparisonOperator: GreaterThanOrEqualToThreshold
      TreatMissingData: notBreaching
      AlarmActions:
        - !Ref AlertTopic

What does the alarm measure?

One visible dead-letter queue message is enough to enter the alarm state. Empty periods remain healthy because missing data is treated as non-breaching.

  • Save `template.yaml`.
  • Validate the two alert sources by running:
sam validate

You should see a valid-template result.

✔️ Awesome, I've got everything!

Your final template now includes the confirmed channel, processor permission, and dead-letter queue alarm.

ⓧ I'd like to double check the full code

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: CloudERP Sentinel Phase 2 risk pipeline

Parameters:
  ErpBucketName:
    Type: String
    Description: Globally unique lowercase S3 bucket name
  AlertEmail:
    Type: String
    Description: Email address that confirms the SNS subscription

Resources:
  ProcessorLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub '/aws/lambda/${AWS::StackName}-risk-processor'
      RetentionInDays: 7

  DeadLetterQueue:
    Type: AWS::SQS::Queue
    Properties:
      MessageRetentionPeriod: 1209600

  ProcessingQueue:
    Type: AWS::SQS::Queue
    Properties:
      MessageRetentionPeriod: 345600
      VisibilityTimeout: 30
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt DeadLetterQueue.Arn
        maxReceiveCount: 5

  ProcessingQueuePolicy:
    Type: AWS::SQS::QueuePolicy
    Properties:
      Queues:
        - !Ref ProcessingQueue
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Sid: AllowS3EventNotifications
            Effect: Allow
            Principal:
              Service: s3.amazonaws.com
            Action: sqs:SendMessage
            Resource: !GetAtt ProcessingQueue.Arn
            Condition:
              ArnEquals:
                aws:SourceArn: !Sub 'arn:${AWS::Partition}:s3:::${ErpBucketName}'
              StringEquals:
                aws:SourceAccount: !Ref AWS::AccountId

  ErpBucket:
    Type: AWS::S3::Bucket
    DependsOn: ProcessingQueuePolicy
    Properties:
      BucketName: !Ref ErpBucketName
      NotificationConfiguration:
        QueueConfigurations:
          - Event: s3:ObjectCreated:*
            Queue: !GetAtt ProcessingQueue.Arn
            Filter:
              S3Key:
                Rules:
                  - Name: prefix
                    Value: incoming/
                  - Name: suffix
                    Value: .csv

  RiskCasesTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: CaseId
          AttributeType: S
      KeySchema:
        - AttributeName: CaseId
          KeyType: HASH

  AlertTopic:
    Type: AWS::SNS::Topic

  AlertEmailSubscription:
    Type: AWS::SNS::Subscription
    Properties:
      TopicArn: !Ref AlertTopic
      Protocol: email
      Endpoint: !Ref AlertEmail

  RiskProcessor:
    Type: AWS::Serverless::Function
    DependsOn: ProcessorLogGroup
    Properties:
      FunctionName: !Sub '${AWS::StackName}-risk-processor'
      CodeUri: src/
      Handler: app.lambda_handler
      Runtime: python3.14
      MemorySize: 256
      Timeout: 5
      Environment:
        Variables:
          TABLE_NAME: !Ref RiskCasesTable
          TOPIC_ARN: !Ref AlertTopic
      Policies:
        - Version: '2012-10-17'
          Statement:
            - Sid: PollProcessingQueue
              Effect: Allow
              Action:
                - sqs:ReceiveMessage
                - sqs:DeleteMessage
                - sqs:GetQueueAttributes
              Resource: !GetAtt ProcessingQueue.Arn
            - Sid: ReadErpExports
              Effect: Allow
              Action: s3:GetObject
              Resource: !Sub 'arn:${AWS::Partition}:s3:::${ErpBucketName}/*'
            - Sid: WriteRiskCases
              Effect: Allow
              Action: dynamodb:PutItem
              Resource: !GetAtt RiskCasesTable.Arn
            - Sid: PublishCriticalAlerts
              Effect: Allow
              Action: sns:Publish
              Resource: !Ref AlertTopic
      Events:
        QueueEvent:
          Type: SQS
          Properties:
            Queue: !GetAtt ProcessingQueue.Arn
            BatchSize: 1
            FunctionResponseTypes:
              - ReportBatchItemFailures

  DeadLetterQueueAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmDescription: A failed ERP event is waiting in the DLQ
      Namespace: AWS/SQS
      MetricName: ApproximateNumberOfMessagesVisible
      Dimensions:
        - Name: QueueName
          Value: !GetAtt DeadLetterQueue.QueueName
      Statistic: Maximum
      Period: 60
      EvaluationPeriods: 1
      Threshold: 1
      ComparisonOperator: GreaterThanOrEqualToThreshold
      TreatMissingData: notBreaching
      AlarmActions:
        - !Ref AlertTopic

Outputs:
  ErpBucketName:
    Value: !Ref ErpBucket
  RiskTableName:
    Value: !Ref RiskCasesTable
  ProcessingQueueUrl:
    Value: !Ref ProcessingQueue
  DeadLetterQueueUrl:
    Value: !Ref DeadLetterQueue
  DeadLetterQueueArn:
    Value: !GetAtt DeadLetterQueue.Arn
  RiskProcessorName:
    Value: !Ref RiskProcessor
  AlertTopicArn:
    Value: !Ref AlertTopic
  • In `src/app.py`, find this client and threshold block:
S3 = boto3.client("s3")
DYNAMODB = boto3.resource("dynamodb")
TABLE = DYNAMODB.Table(os.environ["TABLE_NAME"])
RISK_THRESHOLD = Decimal("70")
  • Replace it with this alert-aware configuration:
S3 = boto3.client("s3")
DYNAMODB = boto3.resource("dynamodb")
SNS = boto3.client("sns")
TABLE = DYNAMODB.Table(os.environ["TABLE_NAME"])
TOPIC_ARN = os.environ["TOPIC_ARN"]
RISK_THRESHOLD = Decimal("70")
CRITICAL_THRESHOLD = Decimal("90")
  • Find this item-building line in `process_row()`:
    item = build_item(row, bucket, key, amount, risk_score)
  • Replace it with these severity-aware lines:
    # Promote scores at or above 90 to critical severity.
    item = build_item(row, bucket, key, amount, risk_score)
    item["Severity"] = (
        "CRITICAL" if risk_score >= CRITICAL_THRESHOLD else "HIGH"
    )
  • Find the final `Created HIGH risk case` log line.
  • Replace that line with this logging and publishing block:
    LOGGER.info(
        "Created %s risk case %s", item["Severity"], item["CaseId"]
    )

    # Publish critical cases only after the durable write succeeds.
    if item["Severity"] == "CRITICAL":
        response = SNS.publish(
            TopicArn=TOPIC_ARN,
            Subject="CloudERP Sentinel critical risk case",
            Message=json.dumps(item, default=str, indent=2),
        )
        LOGGER.info(
            "Published critical alert %s for %s",
            response["MessageId"],
            item["CaseId"],
        )

Where is the reliability boundary?

The table write happens before the SNS call. A production design that must retry alerts independently would use an outbox or DynamoDB Streams.

  • Save `src/app.py`.
  • Check the updated Python syntax by running:
python -m py_compile .\src\app.py
if ($LASTEXITCODE -eq 0) { 'Python syntax valid' }

You should see `Python syntax valid`.

✔️ Awesome, I've got everything!

Your saved processor now writes durable cases and publishes critical alerts.

ⓧ I'd like to double check the full code

# Load CSV, queue, and AWS SDK helpers.
import csv
import io
import json
import logging
import os
from decimal import Decimal, InvalidOperation
from urllib.parse import unquote_plus

import boto3
from botocore.exceptions import ClientError

# Create reusable service clients and runtime configuration.
LOGGER = logging.getLogger()
LOGGER.setLevel("INFO")
S3 = boto3.client("s3")
DYNAMODB = boto3.resource("dynamodb")
SNS = boto3.client("sns")
TABLE = DYNAMODB.Table(os.environ["TABLE_NAME"])
TOPIC_ARN = os.environ["TOPIC_ARN"]
RISK_THRESHOLD = Decimal("70")
CRITICAL_THRESHOLD = Decimal("90")
REQUIRED_COLUMNS = {
    "transaction_id", "vendor", "amount", "currency", "risk_score"
}

# Report only the queue records that fail processing.
def lambda_handler(event, context):
    failures = []
    for record in event.get("Records", []):
        message_id = record["messageId"]
        try:
            process_sqs_message(record)
        except Exception as error:
            LOGGER.exception("Message %s failed: %s", message_id, error)
            failures.append({"itemIdentifier": message_id})
    return {"batchItemFailures": failures}

# Resolve the S3 object referenced by each queue message.
def process_sqs_message(record):
    payload = json.loads(record["body"])
    if payload.get("Event") == "s3:TestEvent":
        LOGGER.info("Ignoring S3 test event")
        return

    s3_records = payload.get("Records")
    if not s3_records:
        raise ValueError("SQS message does not contain S3 Records")

    for s3_record in s3_records:
        bucket = s3_record["s3"]["bucket"]["name"]
        key = unquote_plus(s3_record["s3"]["object"]["key"])
        process_export(bucket, key)

# Validate the CSV shape before processing its rows.
def process_export(bucket, key):
    response = S3.get_object(Bucket=bucket, Key=key)
    csv_text = response["Body"].read().decode("utf-8-sig")
    reader = csv.DictReader(io.StringIO(csv_text))
    missing_columns = REQUIRED_COLUMNS - set(reader.fieldnames or [])
    if missing_columns:
        missing = ", ".join(sorted(missing_columns))
        raise ValueError(f"ERP export is missing columns: {missing}")

    for row_number, row in enumerate(reader, start=2):
        process_row(row, bucket, key, row_number)

# Build the durable table item from a validated row.
def build_item(row, bucket, key, amount, risk_score):
    transaction_id = row["transaction_id"].strip()
    return {
        "CaseId": f"ERP#{transaction_id}",
        "TransactionId": transaction_id,
        "Vendor": row["vendor"].strip(),
        "Amount": amount,
        "Currency": row["currency"].strip(),
        "RiskScore": risk_score,
        "Severity": "HIGH",
        "Status": "OPEN",
        "SourceBucket": bucket,
        "SourceKey": key,
    }

# Validate one row and protect its stable case identifier.
def process_row(row, bucket, key, row_number):
    required = [row["transaction_id"], row["vendor"], row["currency"]]
    if not all(value.strip() for value in required):
        raise ValueError(f"Row {row_number} contains an empty required value")

    try:
        amount = Decimal(row["amount"].strip())
        risk_score = Decimal(row["risk_score"].strip())
    except InvalidOperation as error:
        raise ValueError(f"Row {row_number} contains invalid numeric data") from error

    if risk_score < RISK_THRESHOLD:
        LOGGER.info("Skipped transaction %s", row["transaction_id"])
        return

    # Promote scores at or above 90 to critical severity.
    item = build_item(row, bucket, key, amount, risk_score)
    item["Severity"] = (
        "CRITICAL" if risk_score >= CRITICAL_THRESHOLD else "HIGH"
    )
    try:
        TABLE.put_item(
            Item=item,
            ConditionExpression="attribute_not_exists(CaseId)",
        )
    except ClientError as error:
        code = error.response.get("Error", {}).get("Code")
        if code == "ConditionalCheckFailedException":
            LOGGER.info("Suppressed duplicate risk case %s", item["CaseId"])
            return
        raise

    LOGGER.info(
        "Created %s risk case %s", item["Severity"], item["CaseId"]
    )

    # Publish critical cases only after the durable write succeeds.
    if item["Severity"] == "CRITICAL":
        response = SNS.publish(
            TopicArn=TOPIC_ARN,
            Subject="CloudERP Sentinel critical risk case",
            Message=json.dumps(item, default=str, indent=2),
        )
        LOGGER.info(
            "Published critical alert %s for %s",
            response["MessageId"],
            item["CaseId"],
        )
  • Build and deploy both alert sources by running:
sam build
sam deploy

The deployment can take several minutes. You should see the stack update complete successfully.

The existing dead-letter queue already contains one message. CloudWatch may need one or two 60-second periods before the alarm email arrives.

  • Wait for an AWS alarm email that identifies the dead-letter queue alarm.
  • Select the `events` folder in the Visual Studio Code Explorer sidebar.
  • Use the new-file control at the top of the Explorer sidebar to create `critical-export.csv`.
  • Paste this critical export:
transaction_id,vendor,amount,currency,risk_score
TX-3001,Adventure Works,48000.00,USD,99
  • Save `events/critical-export.csv`.
  • Upload the critical export by running:
aws s3 cp .\events\critical-export.csv s3://$BucketName/incoming/critical-export.csv

Before you check the result, do you expect a score of `99` to produce both a table item and an email?

  • Check the table count by running:
aws dynamodb scan --table-name $TableName --select COUNT --query Count --output text

You should see `4`. Your inbox should also receive a critical-risk email for `ERP#TX-3001`.

What completes the alert proof?

  • The operational email reports the dead-letter queue alarm.
  • The business email reports the critical `ERP#TX-3001` case.
  • The table count of `4` proves the new case became durable.

One of the emails missing?

Allow another two minutes for the alarm evaluation or SNS email delivery. Confirm that the subscription remains active.

Help me trace the missing alert.

Package the evidence

A short project record connects the architecture to the checks you observed. It gives the counts, logs, and alerts enough context to stand on their own.

  • Select the `clouderp-sentinel-phase2` folder in the Visual Studio Code Explorer sidebar.
  • Use the new-file control at the top of the Explorer sidebar to create `README.md`.
  • Paste this project summary:
# CloudERP Sentinel: Phase 2 Risk Pipeline

## What it does

CloudERP Sentinel converts simulated ERP CSV exports into durable risk cases.

## Architecture

Amazon S3 sends CSV object events to Amazon SQS. AWS Lambda validates each row and writes qualifying cases to Amazon DynamoDB.

## Reliability evidence

- Replaying a valid export leaves the case count unchanged.
- A malformed export retries without duplicating its first valid row.
- The failed notification moves to a dead-letter queue after five receives.
- Amazon SNS delivers critical-case and queue-alarm emails.

## Verified result

The main project ends with four risk cases and one isolated failed notification.
  • Save `README.md`.
  • Confirm the evidence file exists by running:
Get-Item .\README.md | Select-Object Name, Length

You should see `README.md` with a non-zero length.

Strong finish. CloudERP Sentinel now ingests exports, suppresses duplicates, isolates poison messages, and sends two distinct operational signals.

Secret mission

Redrive Without Duplicating a Risk Case

A failed queue message points to an Amazon S3 object that you can correct in place. You will overwrite the bad export with fixed data. A redrive will prove that conditional writes keep the replay safe.

Clean Up Your Resources

Clean Up Your Resources

The deployed CloudERP Sentinel pipeline uses pay-as-you-go services from AWS. Decide whether to keep the resources running, pause new activity, or delete the cloud deployment and local files entirely.

Cost warning

Tutorial traffic may fit account credits or service allowances. Actual charges depend on your Region and existing usage.

Stored data can continue to incur charges. Deletion is the safest cost control once you have saved your evidence.

Resources you used:

  • Cloud deployment: the Amazon CloudFormation stack clouderp-sentinel-phase2, including its Amazon S3 bucket, two Amazon SQS queues, AWS Lambda function, Amazon DynamoDB table, Amazon SNS topic and subscription, Amazon CloudWatch alarm and log group, plus stack-managed IAM permissions.
  • Local files: the clouderp-sentinel-phase2 folder on your Desktop, including the source files, test exports, build output, deployment settings, and README.md.

Keep everything running

No action is needed. This option keeps the pipeline available for demonstrations or further testing.

  • Use the deployed pipeline only when you need to demonstrate its behavior.
  • Monitor your AWS charges while the stack remains deployed.
  • Keep the local clouderp-sentinel-phase2 folder as your evidence pack.

Pause - I'll come back to this later

Pausing this architecture means stopping new input while its cloud resources remain deployed.

  • Stop uploading CSV files to s3://$BucketName/incoming/.
  • Retain the deployed stack for your next project session.
  • Keep the local clouderp-sentinel-phase2 folder for future changes.

The S3 objects, DynamoDB items, queues, logs, alarm, and SNS topic remain provisioned. Storage or alarm charges may continue.

Delete - I don't want to use this again

Stack deletion removes the deployed pipeline. This option also removes the local `clouderp-sentinel-phase2` folder after you have saved any evidence you want to keep.

  • Empty the project S3 bucket by running this command:
aws s3 rm s3://$BucketName --recursive

What does this command do?

This command recursively removes the simulated ERP exports from the bucket. Clearing those objects prepares the bucket for stack deletion.

  • Confirm that the terminal lists the removed S3 objects.

Bucket cleanup failed?

Check that $BucketName still contains the stack output value. Confirm that your signed-in profile can delete objects from the project bucket.

Help me troubleshoot why the CloudERP Sentinel S3 cleanup command failed.

The next command removes the CloudFormation stack. The deletion flow asks for confirmation before it proceeds.

  • Delete the clouderp-sentinel-phase2 stack by running this command:
sam delete --stack-name clouderp-sentinel-phase2

What does this command do?

This deletes the CloudFormation stack and its stack-managed resources. The queues, function, table, topic, subscription, alarm, log group, and permissions are removed with the deployment.

  • Approve the deletion when the command requests confirmation.
  • Confirm that the terminal reports the `clouderp-sentinel-phase2` stack was deleted.

Stack deletion failed?

Check the deletion output for the resource that blocked the stack. Confirm that the project S3 bucket is empty before retrying the command.

Help me diagnose why the CloudERP Sentinel stack did not delete.

Deleting the local folder removes your source code and evidence pack permanently. Save any screenshots or notes you want to keep before continuing.

  • Remove the local `clouderp-sentinel-phase2` folder from your Desktop by running:
Set-Location ~/Desktop
Remove-Item -Recurse -Force .\clouderp-sentinel-phase2
  • Confirm the folder is gone by running:
Test-Path .\clouderp-sentinel-phase2

You should see `False`. The cloud deployment and local project files are now removed.

Local folder still in use?

Close Visual Studio Code if Windows reports that a project file is still in use. Then rerun the removal command from your Desktop.

Help me remove the local project folder.

Nice Work!

Nice Work!

You made it! CloudERP Sentinel now turns simulated ERP exports into durable risk cases through event-driven architecture.

You've learned how to:

  • Build an event-driven ingestion path that turns ERP exports into durable Amazon DynamoDB risk cases.
  • Apply idempotency with stable CaseId values and conditional writes.
  • Combine failure isolation and alerting with an Amazon SQS dead-letter queue, Amazon CloudWatch alarm, and Amazon SNS email channel.
  • Secret Mission: Correct the malformed export and redrive its failed event without duplicating either recovered risk case.

Ready to quiz yourself?