Initial duplicate detector

Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.11-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN python3 -m pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 7860
+
+CMD ["python3", "-m", "uvicorn", "src.api:app", "--host", "0.0.0.0", "--port", "7860"]
+

README.md

@@ -0,0 +1,100 @@
+Auto Expense Categorization – Duplicate Detection
+=================================================
+
+This mini-service connects to the `expense` MongoDB database and surfaces *soft* merge suggestions whenever two or more expense entries look like the same purchase. The rules currently implemented are the ones requested:
+
+* Amount difference no more than ±1 %
+* Timestamp difference within a configurable ±N minutes window (default: 10 min)
+* Merchant names that are either identical once normalised or mapped through a merchant-alias table
+
+Instead of destroying or editing any expense rows, the service writes a merge suggestion into the `merge_suggestions` collection so that an operator (or another automation) can perform the actual merge later on.
+
+Quick Start
+-----------
+
+1. Create a virtual environment and install dependencies:
+
+   ```
+   python3 -m venv .venv
+   .\.venv\Scripts\activate
+   python3 -m pip install -r requirements.txt
+   ```
+
+2. Copy `.env.example` to `.env` and set the Mongo connection string if you do not want to rely on the baked-in default.
+
+3. Run the detector (the default config scans the last 48 h of data and writes suggestions only):
+
+   ```
+   python3 -m src.main --minutes 12 --lookback-hours 72
+   ```
+
+   You will see log lines such as:
+
+   ```
+   INFO DuplicateDetector Identified 2 duplicates, suggestion 673a...
+   ```
+
+API Server
+----------
+
+Run the HTTP service with FastAPI/uvicorn:
+
+```
+python3 -m uvicorn src.api:app --reload
+```
+
+Endpoints:
+
+* `GET /health` – readiness probe.
+* `POST /duplicates/detect` – kicks off a scan (body can override `lookback_hours`, `limit`, `amount_pct`, `minutes`).
+* `GET /suggestions?limit=50` – lists recent merge suggestions so the UI can ask “These seem similar. Would you like to merge them?”.
+
+Collections
+-----------
+
+* `expenses`: source data. The detector expects fields such as `_id`, `amount`, `currency`, `merchant`, `expense_time`.
+* `merchant_aliases`: optional alias definitions (`name`, `aliases`).
+* `merge_suggestions`: the service writes documents shaped as:
+
+  ```
+  {
+      "_id": ObjectId(...),
+      "candidate_ids": [...],
+      "message": "These seem similar. Would you like to merge them?",
+      "details": {
+          "amount_delta_pct": 0.53,
+          "time_delta_minutes": 4.2,
+          "merchant_match_rule": "alias"
+      },
+      "audit": {
+          "generated_by": "duplicate-detector",
+          "generated_at": ISODate(...)
+      },
+      "status": "pending"
+  }
+  ```
+
+Configuration
+-------------
+
+All tunables live in `src/config.py`. Environment variables take precedence, so you can tune tolerances per deployment without editing code.
+
+| Variable | Description | Default |
+| --- | --- | --- |
+| `MONGO_URI` | Mongo connection string | Provided URI |
+| `MONGO_DB` | Database name | `expense` |
+| `MONGO_EXPENSE_COLLECTION` | Expenses collection | `expenses` |
+| `MONGO_ALIAS_COLLECTION` | Merchant alias collection | `merchant_aliases` |
+| `MONGO_SUGGESTION_COLLECTION` | Merge-suggestion collection | `merge_suggestions` |
+| `AMOUNT_TOLERANCE_PCT` | Amount delta percentage | `1.0` |
+| `TIME_TOLERANCE_MINUTES` | Time delta minutes | `10` |
+| `DEFAULT_LOOKBACK_HOURS` | How far back to scan | `48` |
+
+Next Steps
+----------
+
+* Wire this module into your ingestion pipeline so suggestions are generated immediately after a new expense is stored.
+* Surface the `merge_suggestions` collection in your UI to show prompts such as “These seem similar. Would you like to merge them?”
+* Extend `MerchantAliasResolver` to sync aliases from your upstream ERP or ML model.
+
+

requirements.txt

@@ -0,0 +1,5 @@
+pymongo>=4.8.0
+python-dotenv>=1.0.1
+fastapi>=0.115.0
+uvicorn>=0.30.6
+

src/__init__.py

@@ -0,0 +1,10 @@
+"""Duplicate detection service package."""
+
+__all__ = [
+    "config",
+    "models",
+    "merchant_alias",
+    "repositories",
+    "duplicate_detector",
+]
+

src/api.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import logging
+from typing import Any, List
+
+from fastapi import FastAPI, HTTPException, Query
+from pydantic import BaseModel, ConfigDict, Field
+
+from .config import settings
+from .duplicate_detector import DuplicateDetector
+from .merchant_alias import MerchantAliasResolver
+from .repositories import (
+    ExpenseRepository,
+    MerchantAliasRepository,
+    MergeSuggestionRepository,
+    build_client,
+)
+
+logger = logging.getLogger("DuplicateAPI")
+
+app = FastAPI(
+    title="Expense Duplicate Service",
+    version="1.0.0",
+    description="Detects near-duplicate expense entries and proposes merge suggestions.",
+)
+
+mongo_client = build_client()
+expense_repository = ExpenseRepository.from_client(mongo_client)
+alias_repository = MerchantAliasRepository.from_client(mongo_client)
+suggestion_repository = MergeSuggestionRepository.from_client(mongo_client)
+
+
+class DetectRequest(BaseModel):
+    lookback_hours: int | None = Field(
+        default=None,
+        ge=1,
+        description="Hours to look back when scanning expenses.",
+    )
+    limit: int | None = Field(
+        default=None,
+        ge=1,
+        le=settings.max_batch_size,
+        description="Maximum number of expenses to evaluate.",
+    )
+    amount_pct: float | None = Field(
+        default=None,
+        gt=0,
+        description="Override amount tolerance percentage (default 1%%).",
+    )
+    minutes: int | None = Field(
+        default=None,
+        gt=0,
+        description="Override time tolerance minutes (default 10).",
+    )
+
+
+class DetectResponse(BaseModel):
+    expenses_scanned: int
+    cluster_count: int
+    suggestion_ids: List[str]
+    message: str
+
+
+class Suggestion(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    id: str = Field(alias="_id")
+    candidate_ids: List[str]
+    message: str
+    details: dict[str, Any]
+    audit: dict[str, Any]
+    status: str
+
+
+@app.get("/health")
+async def health() -> dict[str, str]:
+    return {"status": "ok"}
+
+
+@app.post("/duplicates/detect", response_model=DetectResponse)
+async def detect_duplicates(payload: DetectRequest) -> DetectResponse:
+    lookback_hours = payload.lookback_hours or settings.default_lookback_hours
+    limit = payload.limit or settings.max_batch_size
+    amount_pct = payload.amount_pct or float(settings.amount_tolerance_pct)
+    minutes = payload.minutes or settings.time_tolerance_minutes
+
+    alias_resolver = MerchantAliasResolver()
+    alias_resolver.load_from_cursor(alias_repository.fetch_all())
+    detector = DuplicateDetector(
+        alias_resolver=alias_resolver,
+        suggestions_repo=suggestion_repository,
+        amount_tolerance_pct=amount_pct,
+        time_tolerance_minutes=minutes,
+    )
+
+    expenses = expense_repository.fetch_recent(lookback_hours, limit)
+    if not expenses:
+        return DetectResponse(
+            expenses_scanned=0,
+            cluster_count=0,
+            suggestion_ids=[],
+            message="No expenses found for the requested window.",
+        )
+
+    clusters = detector.find_clusters(expenses)
+    suggestion_ids = detector.persist_suggestions(clusters) if clusters else []
+    message = (
+        "These seem similar. Would you like to merge them?"
+        if suggestion_ids
+        else "No duplicate clusters detected."
+    )
+    return DetectResponse(
+        expenses_scanned=len(expenses),
+        cluster_count=len(clusters),
+        suggestion_ids=suggestion_ids,
+        message=message,
+    )
+
+
+@app.get("/suggestions", response_model=List[Suggestion])
+async def list_suggestions(limit: int = Query(50, ge=1, le=500)) -> List[Suggestion]:
+    docs = suggestion_repository.fetch_recent(limit)
+    return [Suggestion.model_validate(doc) for doc in docs]
+

src/config.py

@@ -0,0 +1,49 @@
+"""Centralised configuration for the duplicate detector."""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from decimal import Decimal
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+DEFAULT_MONGO_URI = (
+    "mongodb://expenseuser:Kem_6o%3F%3F@165.227.69.221:27017/expense?authSource=admin"
+)
+
+
+def _get_decimal(env_key: str, default: str) -> Decimal:
+    raw_value = os.getenv(env_key, default)
+    try:
+        return Decimal(raw_value)
+    except Exception as exc:  # pragma: no cover - defensive logging
+        raise ValueError(f"Invalid decimal for {env_key}: {raw_value}") from exc
+
+
+def _get_int(env_key: str, default: str) -> int:
+    raw_value = os.getenv(env_key, default)
+    try:
+        return int(raw_value)
+    except Exception as exc:  # pragma: no cover - defensive logging
+        raise ValueError(f"Invalid int for {env_key}: {raw_value}") from exc
+
+
+@dataclass(frozen=True)
+class Settings:
+    mongo_uri: str = os.getenv("MONGO_URI", DEFAULT_MONGO_URI)
+    mongo_db: str = os.getenv("MONGO_DB", "expense")
+    expense_collection: str = os.getenv("MONGO_EXPENSE_COLLECTION", "expenses")
+    alias_collection: str = os.getenv("MONGO_ALIAS_COLLECTION", "merchant_aliases")
+    suggestion_collection: str = os.getenv("MONGO_SUGGESTION_COLLECTION", "merge_suggestions")
+    amount_tolerance_pct: Decimal = _get_decimal("AMOUNT_TOLERANCE_PCT", "1.0")
+    time_tolerance_minutes: int = _get_int("TIME_TOLERANCE_MINUTES", "10")
+    default_lookback_hours: int = _get_int("DEFAULT_LOOKBACK_HOURS", "48")
+    service_name: str = os.getenv("SERVICE_NAME", "duplicate-detector")
+    max_batch_size: int = _get_int("MAX_BATCH_SIZE", "5000")
+
+
+settings = Settings()
+

src/duplicate_detector.py

@@ -0,0 +1,144 @@
+"""Core duplicate-detection logic."""
+
+from __future__ import annotations
+
+import logging
+from collections import defaultdict, deque
+from datetime import datetime
+from decimal import Decimal
+from typing import Dict, Iterable, List, Sequence, Set, Tuple
+
+from .config import settings
+from .merchant_alias import MerchantAliasResolver, normalize_merchant
+from .models import DuplicateCluster, Expense, MergeSuggestion
+from .repositories import MergeSuggestionRepository
+
+logger = logging.getLogger("DuplicateDetector")
+
+
+def _pct_delta(a: Decimal, b: Decimal) -> float:
+    if a == 0:
+        return float("inf")
+    return abs(float((a - b) / a * Decimal(100)))
+
+
+def _minutes_delta(a: datetime, b: datetime) -> float:
+    return abs((a - b).total_seconds() / 60)
+
+
+class DuplicateDetector:
+    def __init__(
+        self,
+        *,
+        alias_resolver: MerchantAliasResolver,
+        suggestions_repo: MergeSuggestionRepository,
+        amount_tolerance_pct: float | None = None,
+        time_tolerance_minutes: int | None = None,
+    ) -> None:
+        self.alias_resolver = alias_resolver
+        self.suggestions_repo = suggestions_repo
+        self.amount_tolerance_pct = amount_tolerance_pct or float(settings.amount_tolerance_pct)
+        self.time_tolerance_minutes = time_tolerance_minutes or settings.time_tolerance_minutes
+
+    def _build_graph(self, expenses: Sequence[Expense]) -> Dict[int, Set[int]]:
+        adjacency: Dict[int, Set[int]] = defaultdict(set)
+        for i, exp_a in enumerate(expenses):
+            for j in range(i + 1, len(expenses)):
+                exp_b = expenses[j]
+                delta_minutes = _minutes_delta(exp_a.expense_time, exp_b.expense_time)
+                if delta_minutes > self.time_tolerance_minutes:
+                    break
+                amount_delta_pct = _pct_delta(exp_a.amount, exp_b.amount)
+                if amount_delta_pct > self.amount_tolerance_pct:
+                    continue
+                alias_match, alias_rule = self.alias_resolver.are_aliases(
+                    exp_a.merchant,
+                    exp_b.merchant,
+                )
+                if alias_match:
+                    adjacency[i].add(j)
+                    adjacency[j].add(i)
+        return adjacency
+
+    def _clusters_from_graph(
+        self,
+        adjacency: Dict[int, Set[int]],
+        expenses: Sequence[Expense],
+    ) -> List[DuplicateCluster]:
+        visited: Set[int] = set()
+        clusters: List[DuplicateCluster] = []
+        for node in range(len(expenses)):
+            if node in visited or node not in adjacency:
+                continue
+            component_nodes: List[int] = []
+            queue: deque[int] = deque([node])
+            while queue:
+                current = queue.popleft()
+                if current in visited:
+                    continue
+                visited.add(current)
+                component_nodes.append(current)
+                for neighbor in adjacency[current]:
+                    if neighbor not in visited:
+                        queue.append(neighbor)
+            if len(component_nodes) <= 1:
+                continue
+            component_nodes.sort()
+            component_expenses = [expenses[idx] for idx in component_nodes]
+            amounts = [exp.amount for exp in component_expenses]
+            times = [exp.expense_time for exp in component_expenses]
+            amount_delta_pct = _pct_delta(min(amounts), max(amounts))
+            time_delta_minutes = _minutes_delta(min(times), max(times))
+            merchant_rule = self._merchant_rule(component_expenses)
+            clusters.append(
+                DuplicateCluster(
+                    expenses=component_expenses,
+                    amount_delta_pct=amount_delta_pct,
+                    time_delta_minutes=time_delta_minutes,
+                    merchant_rule=merchant_rule,
+                ),
+            )
+        return clusters
+
+    def _merchant_rule(self, expenses: Sequence[Expense]) -> str:
+        normalized = {normalize_merchant(exp.merchant) for exp in expenses}
+        if len(normalized) == 1:
+            return "exact"
+        return "alias"
+
+    def find_clusters(self, expenses: Sequence[Expense]) -> List[DuplicateCluster]:
+        if not expenses:
+            return []
+        sorted_expenses = sorted(expenses, key=lambda e: e.expense_time)
+        graph = self._build_graph(sorted_expenses)
+        clusters = self._clusters_from_graph(graph, sorted_expenses)
+        logger.info("Evaluated %d expenses, found %d clusters", len(expenses), len(clusters))
+        return clusters
+
+    def persist_suggestions(self, clusters: Iterable[DuplicateCluster]) -> List[str]:
+        suggestion_ids: List[str] = []
+        for cluster in clusters:
+            candidate_ids = [expense.expense_id for expense in cluster.expenses]
+            tie_breaker = "same purchase?" if len(candidate_ids) > 2 else None
+            details = cluster.to_details()
+            if tie_breaker:
+                details["tie_breaker"] = tie_breaker
+            suggestion = MergeSuggestion(
+                candidate_ids=candidate_ids,
+                message="These seem similar. Would you like to merge them?",
+                details=details,
+                audit={
+                    "generated_by": settings.service_name,
+                    "generated_at": datetime.utcnow(),
+                    "rule_version": "v1.0",
+                },
+            )
+            suggestion_id = self.suggestions_repo.insert_soft_merge(suggestion)
+            suggestion_ids.append(suggestion_id)
+            logger.info(
+                "Recorded merge suggestion %s for candidates %s",
+                suggestion_id,
+                candidate_ids,
+            )
+        return suggestion_ids
+

src/main.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+
+from .config import settings
+from .duplicate_detector import DuplicateDetector
+from .merchant_alias import MerchantAliasResolver
+from .repositories import (
+    ExpenseRepository,
+    MerchantAliasRepository,
+    MergeSuggestionRepository,
+    build_client,
+)
+
+
+def configure_logging(verbose: bool) -> None:
+    level = logging.DEBUG if verbose else logging.INFO
+    logging.basicConfig(
+        level=level,
+        format="%(asctime)s %(levelname)s %(message)s",
+    )
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Detect near-duplicate expenses and write merge suggestions.",
+    )
+    parser.add_argument(
+        "--minutes",
+        dest="minutes",
+        type=int,
+        default=settings.time_tolerance_minutes,
+        help="Time tolerance in minutes for comparing expenses (default: %(default)s).",
+    )
+    parser.add_argument(
+        "--amount-pct",
+        dest="amount_pct",
+        type=float,
+        default=float(settings.amount_tolerance_pct),
+        help="Amount tolerance percentage (default: %(default)s).",
+    )
+    parser.add_argument(
+        "--lookback-hours",
+        dest="lookback_hours",
+        type=int,
+        default=settings.default_lookback_hours,
+        help="How far back to fetch expenses (default: %(default)s).",
+    )
+    parser.add_argument(
+        "--limit",
+        dest="limit",
+        type=int,
+        default=settings.max_batch_size,
+        help="Maximum number of expenses to scan (default: %(default)s).",
+    )
+    parser.add_argument(
+        "--verbose",
+        action="store_true",
+        help="Enable debug logging.",
+    )
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    configure_logging(args.verbose)
+
+    client = build_client()
+    alias_repo = MerchantAliasRepository.from_client(client)
+    alias_resolver = MerchantAliasResolver()
+    alias_resolver.load_from_cursor(alias_repo.fetch_all())
+
+    expense_repo = ExpenseRepository.from_client(client)
+    expenses = expense_repo.fetch_recent(args.lookback_hours, args.limit)
+
+    if not expenses:
+        logging.info("No expenses found for lookback window")
+        return 0
+
+    suggestion_repo = MergeSuggestionRepository.from_client(client)
+    detector = DuplicateDetector(
+        alias_resolver=alias_resolver,
+        suggestions_repo=suggestion_repo,
+        amount_tolerance_pct=args.amount_pct,
+        time_tolerance_minutes=args.minutes,
+    )
+
+    clusters = detector.find_clusters(expenses)
+    if not clusters:
+        logging.info("No duplicate clusters detected")
+        return 0
+
+    suggestion_ids = detector.persist_suggestions(clusters)
+    logging.info(
+        "Finished writing %d suggestions. Example message: %s",
+        len(suggestion_ids),
+        "These seem similar. Would you like to merge them?",
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
+

src/merchant_alias.py

@@ -0,0 +1,47 @@
+"""Utilities to resolve merchants that are aliases of each other."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from typing import Dict, Iterable, Set
+
+
+MERCHANT_CLEAN_RE = re.compile(r"[^a-z0-9]+")
+
+
+def normalize_merchant(name: str) -> str:
+    """Lowercase, strip and remove punctuation for comparisons."""
+    cleaned = MERCHANT_CLEAN_RE.sub("", name.strip().lower())
+    return cleaned
+
+
+@dataclass
+class MerchantAliasResolver:
+    """A simple in-memory alias graph.
+
+    The resolver can hydrate itself from Mongo (through a repository) or
+    fall back to a small bootstrapped dictionary.
+    """
+
+    alias_sets: Dict[str, Set[str]] = field(default_factory=dict)
+
+    def load_from_cursor(self, alias_documents: Iterable[dict]) -> None:
+        for doc in alias_documents:
+            canonical = normalize_merchant(doc.get("name", ""))
+            aliases = {normalize_merchant(alias) for alias in doc.get("aliases", [])}
+            aliases.add(canonical)
+            self.alias_sets[canonical] = aliases
+
+    def are_aliases(self, a: str, b: str) -> tuple[bool, str]:
+        norm_a = normalize_merchant(a)
+        norm_b = normalize_merchant(b)
+        if not norm_a or not norm_b:
+            return False, "blank"
+        if norm_a == norm_b:
+            return True, "exact"
+        for root, alias_group in self.alias_sets.items():
+            if norm_a in alias_group and norm_b in alias_group:
+                return True, "alias"
+        return False, "none"
+

src/models.py

@@ -0,0 +1,70 @@
+"""Domain models for expenses and merge suggestions."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from decimal import Decimal
+from typing import List, Mapping, Sequence
+
+
+@dataclass(frozen=True)
+class Expense:
+    expense_id: str
+    amount: Decimal
+    currency: str
+    merchant: str
+    expense_time: datetime
+    source: str | None = None
+    metadata: Mapping[str, object] | None = None
+
+    @staticmethod
+    def from_document(doc: Mapping[str, object]) -> "Expense":
+        try:
+            amount_value = Decimal(str(doc["amount"]))
+        except KeyError as exc:
+            raise ValueError("Expense document missing 'amount'") from exc
+        return Expense(
+            expense_id=str(doc.get("_id")),
+            amount=amount_value,
+            currency=str(doc.get("currency", "INR")),
+            merchant=str(doc.get("merchant", "")).strip(),
+            expense_time=doc["expense_time"],
+            source=doc.get("source"),
+            metadata=doc.get("metadata") or {},
+        )
+
+
+@dataclass
+class MergeSuggestion:
+    candidate_ids: Sequence[str]
+    message: str
+    details: Mapping[str, object]
+    audit: Mapping[str, object]
+    status: str = "pending"
+    _id: str | None = None
+
+    def to_document(self) -> Mapping[str, object]:
+        return {
+            "candidate_ids": list(self.candidate_ids),
+            "message": self.message,
+            "details": dict(self.details),
+            "audit": dict(self.audit),
+            "status": self.status,
+        }
+
+
+@dataclass
+class DuplicateCluster:
+    expenses: List[Expense] = field(default_factory=list)
+    amount_delta_pct: float = 0.0
+    time_delta_minutes: float = 0.0
+    merchant_rule: str = "exact"
+
+    def to_details(self) -> Mapping[str, object]:
+        return {
+            "amount_delta_pct": self.amount_delta_pct,
+            "time_delta_minutes": self.time_delta_minutes,
+            "merchant_match_rule": self.merchant_rule,
+        }
+

src/repositories.py

@@ -0,0 +1,87 @@
+"""Mongo repositories used by the duplicate detector."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from typing import Iterable, List, Mapping
+
+from pymongo import MongoClient
+from pymongo.collection import Collection
+from pymongo.errors import DuplicateKeyError
+
+from . import config
+from .models import Expense, MergeSuggestion
+
+
+def build_client() -> MongoClient:
+    return MongoClient(config.settings.mongo_uri)
+
+
+class ExpenseRepository:
+    def __init__(self, collection: Collection):
+        self._collection = collection
+
+    @classmethod
+    def from_client(cls, client: MongoClient) -> "ExpenseRepository":
+        return cls(
+            client[config.settings.mongo_db][config.settings.expense_collection],
+        )
+
+    def fetch_recent(self, lookback_hours: int, limit: int) -> List[Expense]:
+        since = datetime.now(tz=timezone.utc) - timedelta(hours=lookback_hours)
+        cursor = (
+            self._collection.find(
+                {"expense_time": {"$gte": since}},
+                sort=[("expense_time", 1)],
+                limit=limit,
+            )
+            or []
+        )
+        return [Expense.from_document(doc) for doc in cursor]
+
+
+class MerchantAliasRepository:
+    def __init__(self, collection: Collection):
+        self._collection = collection
+
+    @classmethod
+    def from_client(cls, client: MongoClient) -> "MerchantAliasRepository":
+        return cls(
+            client[config.settings.mongo_db][config.settings.alias_collection],
+        )
+
+    def fetch_all(self) -> Iterable[dict]:
+        return self._collection.find({}, projection={"name": 1, "aliases": 1})
+
+
+class MergeSuggestionRepository:
+    def __init__(self, collection: Collection):
+        self._collection = collection
+
+    @classmethod
+    def from_client(cls, client: MongoClient) -> "MergeSuggestionRepository":
+        return cls(
+            client[config.settings.mongo_db][config.settings.suggestion_collection],
+        )
+
+    def insert_soft_merge(self, suggestion: MergeSuggestion) -> str:
+        candidate_ids = sorted(suggestion.candidate_ids)
+        existing = self._collection.find_one({"candidate_ids": candidate_ids}, {"_id": 1})
+        if existing:
+            return str(existing["_id"])
+        doc = suggestion.to_document()
+        doc["candidate_ids"] = candidate_ids
+        result = self._collection.insert_one(doc)
+        return str(result.inserted_id)
+
+    def fetch_recent(self, limit: int = 50) -> List[Mapping[str, object]]:
+        cursor = (
+            self._collection.find({}, sort=[("audit.generated_at", -1)], limit=limit)
+            or []
+        )
+        suggestions: List[Mapping[str, object]] = []
+        for doc in cursor:
+            doc["_id"] = str(doc.get("_id"))
+            suggestions.append(doc)
+        return suggestions
+