Beanie + MongoDB: Production Pitfalls

## Summary
Beanie's `.upsert(on_insert=...)` would keep the race, so the refactor uses one atomic server-side upsert backed by a unique index instead.

## Beanie's upsert API is not safe here
In Beanie 2.1, `find_one(...).upsert(..., on_insert=Contact(...))` runs an `update_one` without `upsert=True` and, if nothing matched, a separate `insert_one`. That is the same check-then-insert race as the original code. Webhook providers redeliver events and deliver them concurrently, so this path would still produce duplicates or `DuplicateKeyError`.

Instead the function uses the raw collection with `find_one_and_update(upsert=True)` and `$setOnInsert`: a single atomic operation that also returns the document in one round trip.

```python
from datetime import UTC, datetime

import pymongo
from beanie import Document, PydanticObjectId
from pymongo import ReturnDocument
from pymongo.errors import DuplicateKeyError

```python
class Contact(Document):
    organization_id: PydanticObjectId
    email: str
    name: str | None = None
    created_at: datetime

    class Settings:
        name = "contacts"
        indexes = [
```

            # The unique index is what makes the upsert atomic. Without it,
            # two concurrent upserts can both insert.
            pymongo.IndexModel(
                [("organization_id", 1), ("email", 1)],
                unique=True,
                name="contacts_org_email_unique",

```
            ),
        ]
```

async def get_or_create_contact(
    org_id: PydanticObjectId, email: str, name: str | None

```sql
) -> Contact:
    filter_ = {"organization_id": org_id, "email": email}
    update = {"$setOnInsert": {"name": name, "created_at": datetime.now(UTC)}}
    coll = Contact.get_pymongo_collection()

    for attempt in range(2):
        try:
            raw = await coll.find_one_and_update(
                filter_,
                update,
                upsert=True,
                return_document=ReturnDocument.AFTER,
            )
            break
        except DuplicateKeyError:
```

            # Lost a race on the unique index; the retry matches the winner's doc.

```
            if attempt:
                raise
    return Contact.model_validate(raw)
```

Requirements and behaviour

┌──────────────┬──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│              │                                                                                                                                                                  │
├──────────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ Unique index │ Required on (organization_id, email). Without it, two concurrent upserts can both insert.                                                                        │
├──────────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ Retry        │ The loser of a race can still get DuplicateKeyError; it retries once and matches the winner's document. It never catches and logs, which would hide lost writes. │
├──────────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ $setOnInsert │ name and created_at are written only on insert, so a repeat webhook doesn't overwrite an existing contact.                                                       │
├──────────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ Timestamps   │ datetime.now(UTC) replaces utcnow(); the client must use tz_aware=True.                                                                                          │
├──────────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ Tenant scope │ The filter includes organization_id.                                                                                                                             │
└──────────────┴──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘

Next steps

- Before deploying: the old code has probably created duplicates already. Dedupe contacts and confirm zero violations against production, or building the new unique index crashes the boot. Run scripts/check_indexes.py against prod to verify.
- Normalise emails (.strip().lower()) before calling, or different casings count as different contacts.
- Add a test that fails against the old code: force the interleaving, or assert the index exists and that a second raw insert raises. mongomock doesn't model concurrency, so a plain asyncio.gather test proves nothing.