Publishing events

publish_event()

libs/morpheus-common/morpheus_common/events/publisher.py — the shared publisher for the two FIFO queues.

def publish_event(
    *,
    queue_url: str,
    event: BaseModel,
    group_id: str,
    dedup_id: str | None = None,
) -> None:
Argument Meaning

queue_url

Full SQS queue URL, from the function’s Settings

event

Any Pydantic model — serialised with model_dump_json()

group_id

FIFO MessageGroupId. Messages in one group are ordered and deduplicated within a five-minute window.

dedup_id

FIFO MessageDeduplicationId. Defaults to sha256 of the serialised body.

The boto3 SQS client is a module-level singleton pinned to us-east-1. Publishing logs at DEBUG with the event type, group id, and queue — never the body.

Usage:

from morpheus_common.events import ContactUpdateEvent, publish_event

publish_event(
    queue_url=settings.crm_sync_queue_url,
    event=ContactUpdateEvent(
        subscription_id=str(settings.subscription_id),
        hubspot_contact_id=hubspot_cid,
        properties={"call_list_date": today, "call_priority": "4"},
    ),
    group_id=hubspot_cid,
)

Because event_type defaults on the model, constructing the model is enough — the routing key cannot drift from the registry key without someone editing the default.

Choosing a group_id

FIFO ordering is per group, and a group is processed serially. The choice is therefore a throughput decision as much as a correctness one.

Producer group_id Why

enrichment-scraper → SellerRecord.Received

seller_id

One group per seller. Grouping by marketplace would cap the consumer at two concurrent invocations regardless of queue depth.

enrichment-scorer → Seller.CrmSync

seller_id

Two syncs for one seller must not interleave; different sellers are independent.

phone-agent → Contact.Update (line intel)

target_phone_number_id

One group per phone row.

phone-agent → Contact.Update (campaign)

hubspot_contact_id

Serialises updates to one contact; parallel across contacts.

phone-agent → FollowUpTask.Sync

task_id

One group per task.

The rule of thumb: group by the entity being mutated. That gives ordering where it matters and parallelism everywhere else.

Choosing a dedup_id

Omitting it hashes the serialised body, which deduplicates identical messages. That is right when payload equality means "the same event".

Supply one explicitly when identity is narrower than the payload:

publish_event(
    queue_url=settings.crm_sync_queue_url,
    event=crm_event,
    group_id=seller_id,
    dedup_id=hashlib.sha256(f"crm-sync-{seller_id}".encode()).hexdigest(),
)

Here the dedup id keys on seller identity alone, so two enrichment passes that produced slightly different results inside the five-minute window collapse to one CRM sync. The alternative — a body hash — would let both through.

ContentBasedDeduplication is false on both FIFO queues, so an explicit or default-computed id is always required.

QueueService in the scraper builds its own entries for send_message_batch rather than using publish_event(), because batching is what makes 10 messages one API call. It uses sha256(seller_id + marketplace). New producers should reach for publish_events() instead of hand-rolling batch entries.

publish_events()

The batched counterpart, for producers with more than a handful of messages. It chunks into SendMessageBatch calls of ten (the SQS maximum), so a thousand events cost a hundred API calls rather than a thousand.

def publish_events(
    *,
    queue_url: str,
    events: Sequence[OutboundEvent],
) -> list[str]:

OutboundEvent is a NamedTuple of group_id, event and an optional dedup_id, so the per-message semantics match publish_event() exactly — including the default body hash.

Failures are returned, not raised: the result is the group_id of every event that did not make it, whether SQS rejected the individual entry or the whole chunk errored. One bad chunk therefore cannot abort the rest of the batch. Callers should log the count and move on.

from morpheus_common.events import ContactUpdateEvent, OutboundEvent, publish_events

outbound = [
    OutboundEvent(
        group_id=hubspot_cid,
        event=ContactUpdateEvent(...),
    )
    for hubspot_cid in targets
]

failed = publish_events(queue_url=settings.crm_sync_queue_url, events=outbound)
logger.info("published=%d failed=%d", len(outbound) - len(failed), len(failed))

Prefer publish_events() over calling publish_event() in a loop whenever the row count is unbounded. A scheduled Lambda publishing per row spends one network round trip per record, which is a common cause of hitting the function timeout part-way through a batch — silently dropping whichever records happen to sort last.

Standard queues

EmailAgentQueue is a standard queue, so MessageGroupId and MessageDeduplicationId are invalid and would be rejected. phone-agent has a separate publisher for it:

from services.sqs_publisher import publish_to_standard_queue

publish_to_standard_queue(
    queue_url=settings.email_agent_queue_url,
    body={"event_type": "Email.SendInternal", "task_id": str(task.id), …},
    event_type="Email.SendInternal",
)

It sets the event_type message attribute, which is where the email-agent dispatcher reads the routing key.

Which queue a producer targets determines which publisher to use. Using publish_event() against EmailAgentQueue fails; using publish_to_standard_queue() against a FIFO queue fails for the opposite reason.

Message attribute versus body field

Queue Routing key location

EnrichmentPipelineQueue.fifo

event_type message attribute

EmailAgentQueue

event_type message attribute

CrmSyncQueue.fifo

event_type field inside the body

CrmSyncQueue reads from the body, which is what allows a message body to be a JSON array of events dispatched individually — an array has no single attribute to route on.

Producers should set both where practical: the models carry event_type regardless, so a message published with publish_event() always has it in the body.

Fire-and-forget publishing

Where a publish is not the point of the operation, it is best-effort:

try:
    publish_event(queue_url=..., event=event, group_id=seller_id, dedup_id=...)
except Exception:
    logger.warning(
        "Failed to publish CRM sync event sellerId=%r — enrichment DB write was committed",
        seller_id, exc_info=True,
    )

The database write already committed; failing the message would repeat expensive AI work to retry a cheap publish. Campaign re-engagement gets the same tolerance from publish_events(), which reports failed group ids instead of raising, so one rejected entry does not stop the rest of the batch.

Contrast that with Phone.LineIntelPending, which returns early with an ERROR when CRM_SYNC_QUEUE_URL is unset — there the publish is the whole point of the job.

Publishing from outside this repository

The Payability web application publishes most CrmSyncQueue events. It must match the same contract:

  • event_type in the body, matching a key in lambdas/morpheus_hubspot/registry.py#event_registry.

  • A body that validates against the corresponding Pydantic model in morpheus_common.events.models.

  • subscription_id matching the consumer’s configured tenant, where the model has one.

  • A MessageGroupId — the entity being mutated — and a MessageDeduplicationId.

Queue URLs come from the SAM stack outputs (CrmSyncQueueUrl). The EmailAgentQueue outputs are commented out, so read that URL with aws sqs get-queue-url --queue-name morpheus-sam-email-agent.