EnrichmentPipelineQueue events

Queue: morpheus-sam-enrichment-pipeline.fifo. Producer: enrichment-scraper. Consumer: enrichment-scorer.

Routing key: the event_type SQS message attribute, set to SellerRecord.Received.

The SQS trigger on enrichment-scorer is currently commented out in lambdas/template.yaml, so published messages accumulate on the queue (retained four days) until it is enabled. See EventBridge schedules.

SellerRecord.Received

Model: SellerRecord — defined independently in both functions (enrichment_scraper/models/seller.py and enrichment_scorer/models/seller.py), since the units may not import each other.

class Marketplace(StrEnum):
    AMAZON = "amazon"
    WALMART = "walmart"

class DataSource(StrEnum):
    SMARTSCOUT = "smartscout"
    JUNGLE_SCOUT = "jungle_scout"
    WALMART_SCRAPER = "walmart_scraper"

class SellerRecord(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True, frozen=True)

    seller_id: str                       # external marketplace id, e.g. ATVPDKIKX0DER
    marketplace: Marketplace
    business_name: str
    storefront_url: str
    estimated_monthly_revenue: float     # validator: must be >= 0
    asin_count: int                      # validator: must be >= 0
    primary_category: str
    seller_since: date | None
    percent_fba: float | None
    mom_growth: float | None
    num_brands: int | None
    state: str | None
    country: str | None
    source: DataSource
    raw: dict                            # the full provider payload

frozen=True makes the record immutable once validated: a handler cannot accidentally mutate the event it is processing. str_strip_whitespace=True normalises every string field on the way in.

raw carries the entire SmartScout row, so a future consumer can extract a field that was not mapped without re-fetching from the provider.

Example

{
  "seller_id": "ATVPDKIKX0DER",
  "marketplace": "amazon",
  "business_name": "Example Trading Co",
  "storefront_url": "https://www.amazon.com/sp?seller=ATVPDKIKX0DER",
  "estimated_monthly_revenue": 42000.0,
  "asin_count": 87,
  "primary_category": "Home & Kitchen",
  "seller_since": "2022-03-14",
  "percent_fba": 0.92,
  "mom_growth": 8.4,
  "num_brands": 3,
  "state": "DE",
  "country": "US",
  "source": "smartscout",
  "raw": { "amazonSellerId": "ATVPDKIKX0DER", "…": "…" }
}

SQS message construction

QueueService.publish() sends send_message_batch calls of 10:

Field Value

MessageBody

record.model_dump_json()

MessageGroupId

record.seller_id

MessageDeduplicationId

sha256(seller_id + marketplace)

MessageAttributes.event_type

{"DataType": "String", "StringValue": "SellerRecord.Received"}

Why one group per seller

Grouping by marketplace would serialise the entire pipeline behind two FIFO groups, capping the scorer at two concurrent invocations regardless of queue depth. One group per seller preserves ordering for that seller — which is all that matters, since the consumer upserts by sellerId — while allowing wide parallelism.

Deduplication

The id is derived from seller identity, not from message content, so the same seller discovered twice inside SQS’s five-minute deduplication window is enqueued once. ContentBasedDeduplication is false on the queue, so the explicit id is required.

Consumer behaviour

SellerReceivedHandler runs four steps, only the first of which is allowed to fail the message:

  1. _upsert_seller() — ON CONFLICT (sellerId) DO UPDATE. Commits.

  2. _enrich() — AI enrichment, skipped when enrichedAt is set. Best-effort.

  3. _apply_enrichment() — write results, publish Seller.CrmSync. Best-effort.

  4. _score() + _apply_score() — payout-pain score, skipped when already set. Best-effort.

Full detail: Seller enrichment.

Backward compatibility

The consumer defaults its routing key to SellerRecord.Received when the event_type message attribute is absent (_DEFAULT_SQS_EVENT_TYPE in handler.py), for messages published before the attribute was added.

Failure handling

maxReceiveCount: 1 — a handler failure sends the message straight to morpheus-sam-enrichment-pipeline-dlq.fifo, retained 14 days.

Failure Fate

Malformed JSON body

DLQ

ValidationError (for example negative revenue)

DLQ

Unknown event_type

deleted — retrying cannot route it

Step 1 raised

DLQ

Steps 2–4 raised

message acked — the seller row is committed and the failure logged

Queue VisibilityTimeout is 180 s against the consumer’s 120 s timeout.