diff --git a/backend/app/main.py b/backend/app/main.py index 7362ae3..cb15c2c 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -30,6 +30,15 @@ settings = get_settings() inbound_server = InboundSMTPServer() +def _index_option_matches(index: dict, options: dict) -> bool: + return ( + bool(index.get("unique")) == bool(options.get("unique")) + and bool(index.get("sparse")) == bool(options.get("sparse")) + and dict(index.get("partialFilterExpression") or {}) + == dict(options.get("partialFilterExpression") or {}) + ) + + async def ensure_mongo_index(collection, keys, **options) -> None: """Create an index, replacing any same-name or same-key index with different options.""" key_list = [(field, direction) for field, direction in keys] @@ -39,9 +48,6 @@ async def ensure_mongo_index(collection, keys, **options) -> None: name = "_".join(f"{field}_{direction}" for field, direction in key_list) options["name"] = name - desired_unique = bool(options.get("unique")) - desired_sparse = bool(options.get("sparse")) - async with await collection.list_indexes() as cursor: async for index in cursor: if index["name"] == "_id_": @@ -51,17 +57,25 @@ async def ensure_mongo_index(collection, keys, **options) -> None: same_name = index["name"] == name if not same_key and not same_name: continue - same_options = ( - bool(index.get("unique")) == desired_unique - and bool(index.get("sparse")) == desired_sparse - ) - if same_key and same_name and same_options: + if same_key and same_name and _index_option_matches(index, options): return await collection.drop_index(index["name"]) await collection.create_index(keys, **options) +async def cleanup_incomplete_handoff_links(database) -> None: + """Remove leftover links that cannot participate in the unique source key.""" + await database.handoff_message_links.delete_many( + { + "$or": [ + {"source_message_id": None}, + {"source_message_id": {"$exists": False}}, + ] + } + ) + + async def backfill_missing_thread_ids(database) -> None: cursor = database.mailbox_entries.find({"thread_id": None}) async for entry in cursor: @@ -302,6 +316,7 @@ async def lifespan(_: FastAPI): [("account_id", 1), ("handoff_thread_id", 1)], unique=True, ) + await cleanup_incomplete_handoff_links(mongo_db) await ensure_mongo_index( mongo_db.handoff_message_links, [("handoff_id", 1), ("created_at", -1)], @@ -310,7 +325,9 @@ async def lifespan(_: FastAPI): mongo_db.handoff_message_links, [("handoff_id", 1), ("source_message_id", 1), ("relationship", 1)], unique=True, - sparse=True, + partialFilterExpression={ + "source_message_id": {"$exists": True, "$type": "objectId"} + }, ) await ensure_mongo_index( mongo_db.handoff_delivery_jobs,