Designing a Security-Findings Aggregator with DDD and Kafka

Security teams run several cloud security posture management (CSPM) and scanning tools, and each one reports problems in its own format. An aggregator pulls those findings into one place, removes duplicates, ranks them, and opens remediation work for the owning team.

That sounds like a simple data pipeline. It gets complicated because each tool has its own idea of what a "finding" or "severity" is, findings arrive constantly, and every mistake costs money. A lost finding is an open hole nobody is fixing. A duplicated finding is a second ticket that wastes someone's afternoon and teaches the team to ignore the queue.

One bounded context per job, not per database table. A bounded context is a boundary inside which a word has one exact meaning. Here, "severity" means three different things depending on where you stand:

• To a vendor tool, it's that tool's own label or score. AWS Security Hub's ASFF format, for example, carries both a label (INFORMATIONAL to CRITICAL) and a normalized 0–100 number.
• To the risk team, it's our own scale, adjusted for how exposed the asset is.
• To the remediation team, it's a due date.

Forcing one severity field to mean all three is how these systems rot. So the design has four contexts:

Figure 1: Four bounded contexts, connected only by events.

The contexts never share tables and never call into each other's internals. Each one owns its data, and the arrows are the only coupling.

Keeping vendor models out: the anti-corruption layer

Each tool gets an adapter that translates its payload into our language at the edge, and nothing past the adapter ever sees the vendor's fields. This is the anti-corruption layer (ACL) pattern. When a vendor renames a field or ships a new schema version, one adapter changes and the rest of the system doesn't notice.

from pydantic import BaseModel, ConfigDict


class NormalizedFinding(BaseModel):
    """Our model of a finding. Frozen: once translated, it never changes."""

    model_config = ConfigDict(frozen=True)

    source: str
    source_id: str
    resource_arn: str
    rule: str
    severity: Severity  # value object, defined in the next section


class SecurityHubAdapter:
    """Translates an ASFF finding into our model. Nothing else imports ASFF."""

    _LABELS = {"INFORMATIONAL": 0, "LOW": 1, "MEDIUM": 2, "HIGH": 3, "CRITICAL": 4}

    def translate(self, raw: dict) -> NormalizedFinding:
        return NormalizedFinding(
            source="securityhub",
            source_id=raw["Id"],
            resource_arn=raw["Resources"][0]["Id"],
            rule=raw["GeneratorId"],
            severity=Severity(level=self._LABELS[raw["Severity"]["Label"]]),
        )

Value objects for anything defined only by its values, entities for things with an identity that lasts, and aggregates for whatever must stay consistent at the same instant. These are the tactical building blocks of DDD, and getting them right is what keeps business rules in one place.

from datetime import datetime, timezone
from uuid import UUID, uuid4

from pydantic import BaseModel, ConfigDict, Field


class Severity(BaseModel):
    """Value object: immutable, validated once, equal by value."""

    model_config = ConfigDict(frozen=True)

    level: int = Field(ge=0, le=4)  # 0 = informational ... 4 = critical


class RemediationCaseClosed(BaseModel):
    """Domain event raised by the aggregate."""

    model_config = ConfigDict(frozen=True)

    event_id: UUID = Field(default_factory=uuid4)
    case_id: str
    occurred_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))


class RemediationCase(BaseModel):
    """Aggregate root: the only way to change a case and its tasks."""

    model_config = ConfigDict(validate_assignment=True)

    case_id: str
    finding_key: str  # reference other aggregates by ID, never by object
    open_tasks: set[str] = Field(default_factory=set)
    closed: bool = False
    events: list[RemediationCaseClosed] = Field(default_factory=list)

    def close(self) -> None:
        if self.open_tasks:
            raise ValueError("cannot close a case with open tasks")
        self.closed = True
        self.events.append(RemediationCaseClosed(case_id=self.case_id))

A few things Pydantic v2 gives a domain model for free:

• frozen=True makes a real value object. Severity(level=3) == Severity(level=3) is True, instances are hashable, and assigning to level raises an error.
• Field(ge=0, le=4) puts the rule in the type. Severity(level=9) fails at construction, so an invalid severity can't exist anywhere in the system.
• validate_assignment=True on the aggregate re-checks every field you change, not just the ones set at creation.

Two rules keep aggregates healthy:

• Keep them small. A RemediationCase holds its own tasks, not the finding, the asset and the owning team. A big aggregate means many writers fight over one row.
• One transaction changes one aggregate. When closing a case should update the finding's status, the case raises a RemediationCaseClosed event and the Risk context updates the finding in its own transaction. The two are briefly out of step, and that's an accepted trade-off.

That second rule is where event-driven design comes in.

A domain event records something that happened inside a context. An integration event is the public, versioned version of it that other contexts are allowed to depend on. Keeping these separate lets a context refactor its internals without breaking every consumer.

FindingTriaged is published to Kafka with a stable schema: finding key, our severity, the owning team, and an event ID. The internal Finding class can change shape freely as long as that contract holds. With Pydantic, the contract is a model that both sides import:

from typing import Literal
from uuid import UUID, uuid4

from pydantic import BaseModel, ConfigDict, Field


class FindingTriaged(BaseModel):
    """Integration event: the public contract on the findings.triaged topic."""

    model_config = ConfigDict(frozen=True)

    schema_version: Literal[1] = 1
    event_id: UUID = Field(default_factory=uuid4)
    finding_key: str
    severity: Severity
    owner_team: str


triaged = FindingTriaged(finding_key="k-1", severity=Severity(level=4), owner_team="payments")
payload = triaged.model_dump_json()                     # producer: write to the outbox
received = FindingTriaged.model_validate_json(payload)  # consumer: validate on read

A message that fails model_validate_json, such as one with an unknown schema_version, is rejected before it touches any business logic. It goes straight to the dead-letter topic instead of retrying forever.

Two choices shape the event flow:

• Event-carried state, not bare notifications. The event includes the severity and owner, so Remediation doesn't have to call back into Risk to look them up. That callback would quietly couple the two contexts again.
• Choreography for the main path. Each context reacts to the events it cares about and nobody orchestrates the whole flow. When a workflow needs a timeout or a compensating step, such as reopening a case if a re-scan finds the issue again, a small orchestrator owns just that workflow. I covered the same trade-off for AI agents in the multi-agent coordination tax: mesh vs mediator.

Because the database and the broker can't commit together, and Kafka delivers at least once. That gives two gaps, and a crash can land in either one.

1. The dual-write gap (producer side). The service commits the finding to its database, then crashes before publishing FindingIngested. The finding exists, but nothing downstream ever hears about it. That's a lost event.
2. The redelivery gap (consumer side). The consumer opens a remediation task, then crashes before committing its Kafka offset. On restart it reads the same message again and opens a second task. That's a duplicate.

The fix: outbox on the way out, inbox on the way in

• Transactional outbox. The service writes the finding and an outbox row in one database transaction. A separate relay reads unpublished outbox rows and publishes them. If the relay crashes after publishing but before marking a row as sent, it publishes that row again. Nothing is lost, and duplicates are now expected.
• Inbox (idempotent consumer). The consumer records each event ID in an inbox table with a primary key, in the same transaction as its side effect. A redelivered event finds its ID already there and is skipped.

Switch between the two versions to see where the crash lands and what it does:

Figure 2: The same crash in the same place, with and without the outbox and inbox.

with db:  # one transaction: dedupe + side effect
    seen = db.execute("SELECT 1 FROM inbox WHERE event_id = ?", (eid,)).fetchone()
    if not seen:
        db.execute("INSERT INTO inbox VALUES (?)", (eid,))
        db.execute("INSERT INTO task(finding_id) VALUES (?)", (fid,))
# commit the Kafka offset only after this transaction succeeds

The outbox makes delivery at least once. The inbox makes processing effectively once. Neither one works alone.

I wrote a small simulation in Python with SQLite and ran it: 10,000 findings, a 2% chance of a crash at each gap, three random seeds. The "Kafka topic" is an in-memory log consumed at least once, so this measures the logic of the patterns, not Kafka's performance.

Ranges are across the three seeds. Three things stood out:

• The naive pipeline's total was nearly correct. Lost events and duplicate tasks roughly cancelled out, so the total stayed within 0.3% of 10,000. A dashboard that only checks "findings in vs tasks out" would have passed it, while about 4% of findings were handled wrong: roughly 2% had no task at all and roughly 2% had two. Reconcile on IDs, not counts.
• The safe pipeline wasn't duplicate-free on the wire. The outbox relay published 10,206–10,233 events for 10,000 findings, and the consumer handled 10,420–10,451 deliveries. The inbox absorbed every one of those extras. Duplicates still happen; they just stop mattering.
• The fix costs one table and one rule. The inbox is a single-column table, and the rule is "side effect and inbox insert in the same transaction, offset commit after."

What about messages that keep failing?

Retry a few times with backoff, then move the message to a dead-letter topic and keep going. A finding with a malformed resource ARN will fail every time. If the consumer retries it forever, every finding behind it on that partition waits. A dead-letter topic (DLQ) holds poison messages so a person can look at them, with the error and the original payload attached.

What about ordering?

Kafka only guarantees order within a partition, so key every event by the thing whose order matters. Here that's the finding key. Every event about one finding lands on one partition in order, so "triaged" can't overtake "ingested". Events about different findings have no ordering promise, and the design doesn't need one.

REST is the synchronous front door: engineers, dashboards and ticketing tools read findings and update cases over HTTP, while Kafka carries everything between contexts. The Reporting context builds a read model from events and serves it, so queries never touch the write side.

Making POST safe to retry

GET, PUT and DELETE are idempotent by definition in HTTP (RFC 9110); POST is not, so give it an Idempotency-Key header. The client sends a unique key with each import. The server stores the key with the response, and a retry with the same key returns the stored response instead of importing twice. It's the same idea as the inbox table, applied at the HTTP edge. The IETF draft for the header standardizes what payment APIs have done for years.

The read model also means the API's data is a little behind the write side. The API should say so: return a last_updated timestamp, and after a client changes a case, return the new state in the response rather than making it re-query a read model that may not have caught up. The repository and data-access layers underneath are covered in three layers between your code and your database, and choosing the stores themselves in the databases field guide.

Follow one finding from left to right, and each architecture style shows up as a layer running underneath the whole path. The columns are the stages a finding moves through. The rows are the four styles, showing what each one does at each stage.

Figure 3: The whole path of one finding, with the four architecture styles as layers. On a narrow screen the rows stack, one style at a time.

Each style solves a different problem, and they meet at clear seams:

Key takeaways:

• Draw boundaries around meaning, not tables. If one word means three things, you have three contexts.
• Translate vendor models at the edge, so a schema change upstream touches one adapter.
• Put each business rule in one aggregate and let events update the rest.
• The outbox stops lost events, and the inbox stops duplicate effects. You need both.
• Check correctness by reconciling IDs. Counts can match while the data is wrong.

• Azure Architecture Center — Using domain analysis to model microservices
• Azure Architecture Center — Event-driven architecture style
• AWS Prescriptive Guidance — Transactional outbox pattern
• RFC 9110 — HTTP Semantics