feat(catalog): auto-purge orphan materials when quotes/items release them

A material that ends up referenced by nothing (no line_items anywhere, no warehouse stock) was previously left behind forever, accumulating dead catalog rows as offers were edited or deleted.

- new quote_service.cleanup_orphan_materials(db, material_ids): for each candidate it checks usage across line_items (all quotes) and warehouse inventory; only when both are zero does it remove the auxiliary rows (supplier_materials, brand_materials, material_units, material_attribute_values) and finally the materials row; returns the purged ids
- replace_quote_children (edit modal): snapshots the quote's material ids before the children are replaced, diffs against the incoming items, and purges whatever was detached
- delete_quote: collects the quote's material ids + attachment keys first, deletes the quote, purges orphans inside the transaction, and only then (post-commit) removes the MinIO objects — now both the recorded attachment keys and a recursive remove_prefix(quote_id/) sweep, so stray objects under the quote folder cannot survive; the response's cascaded block reports purged_orphan_materials
- line_items: PATCH that switches a line item's material, and DELETE of a line item, both clean up the previously referenced material
- storage.remove_prefix(prefix): best-effort recursive object deletion under a prefix (sync MinIO client wrapped in asyncio.to_thread, same pattern as remove_object)

Safety: usage checks run against the whole system before any delete, so shared materials — including those held only by warehouse stock — are never touched. Verified: python compile + api image build clean.

backend/app/services/quote_service.py

@@ -290,12 +290,78 @@ async def quote_has_receipts(db: AsyncSession, quote_id: UUID) -> bool:
     return (count or 0) > 0
 
 
+async def cleanup_orphan_materials(db: AsyncSession, material_ids) -> list[UUID]:
+    """Check candidate material_ids and purge any that have zero references.
+
+    A material is considered an orphan if:
+    1. It is not referenced by ANY line_items across all quotes.
+    2. It is not referenced by ANY inventory stock items in warehouses.
+
+    If count == 0:
+    - Auxiliary records in supplier_materials, brand_materials, material_units,
+      and material_attribute_values are cleaned up.
+    - The material row is deleted from the central materials table.
+
+    Returns the list of purged material UUIDs.
+    """
+    purged: list[UUID] = []
+    unique_ids = {m for m in material_ids if m is not None}
+    if not unique_ids:
+        return purged
+
+    from app.models.catalog import (
+        BrandMaterial,
+        Material,
+        MaterialAttributeValue,
+        MaterialUnit,
+    )
+    from app.models.inventory import InventoryItem
+    from app.models.procurement import LineItem, SupplierMaterial
+
+    for mat_id in unique_ids:
+        # Check active usage in line_items across the whole system
+        line_item_count = await db.scalar(
+            select(func.count(LineItem.id)).where(LineItem.material_id == mat_id)
+        )
+        if (line_item_count or 0) > 0:
+            continue
+
+        # Check active usage in warehouse inventory
+        inv_count = await db.scalar(
+            select(func.count(InventoryItem.id)).where(InventoryItem.material_id == mat_id)
+        )
+        if (inv_count or 0) > 0:
+            continue
+
+        # Material is not used in ANY line item and not in ANY warehouse inventory: pure orphan!
+        await db.execute(
+            delete(SupplierMaterial).where(SupplierMaterial.material_id == mat_id)
+        )
+        await db.execute(
+            delete(BrandMaterial).where(BrandMaterial.material_id == mat_id)
+        )
+        await db.execute(
+            delete(MaterialUnit).where(MaterialUnit.material_id == mat_id)
+        )
+        await db.execute(
+            delete(MaterialAttributeValue).where(MaterialAttributeValue.material_id == mat_id)
+        )
+        await db.execute(
+            delete(Material).where(Material.id == mat_id)
+        )
+        purged.append(mat_id)
+        logger.info("Purged orphan material %s from central catalog", mat_id)
+
+    return purged
+
+
 async def replace_quote_children(
     db: AsyncSession, quote_id: UUID, payload: QuoteChildrenReplaceIn, actor: str | None = None
 ) -> Quote:
     """Full replacement of a quote's line items + termsheets (edit modal).
 
     Only allowed while the quote is still offered and has no warehouse receipts.
+    Automatically purges any materials detached during this replacement if they become orphans.
     """
     async with db.begin():
         quote = await get_quote_or_404(db, quote_id)
@@ -309,6 +375,11 @@ async def replace_quote_children(
                 "این پیشنهاد به‌دلیل ثبت رسید در انبار، غیرقابل ویرایش یا بازگشت است.",
                 field="status",
             )
+
+        # 1. Pre-update snapshot: record previous material_ids assigned to this quote's items
+        old_items = (await db.execute(select(LineItem).where(LineItem.quote_id == quote.id))).scalars().all()
+        old_material_ids = {li.material_id for li in old_items if li.material_id is not None}
+
         # children cascade via DDL: item_termsheets, item_decisions, item_receipts, schedule_steps
         await db.execute(
             delete(LineItem).where(LineItem.quote_id == quote.id).execution_options(synchronize_session=False)
@@ -317,10 +388,21 @@ async def replace_quote_children(
             delete(Termsheet).where(Termsheet.quote_id == quote.id).execution_options(synchronize_session=False)
         )
         await _insert_children(db, quote.id, quote.supplier_id, payload.line_items, payload.termsheets, payload.item_termsheets)
+        await db.flush()
+
+        # 2. Identify detached materials: previous materials no longer present in the updated line items
+        new_material_ids = {li.material_id for li in payload.line_items if li.material_id is not None}
+        detached_material_ids = old_material_ids - new_material_ids
+
+        # 3. Clean up any detached materials that are now orphans (0 usage)
+        if detached_material_ids:
+            await cleanup_orphan_materials(db, detached_material_ids)
+
         _add_event(db, quote.id, "edited", {"changed_fields": ["line_items", "termsheets"]}, actor)
     return quote
 
 
+
 async def _apply_status_transition(db: AsyncSession, quote: Quote, new_status: str) -> None:
     """Backward transitions only; reverts bookkeeping when going back to offered."""
     current = quote.status
@@ -721,30 +803,49 @@ async def delete_quote(db: AsyncSession, quote_id: UUID) -> dict:
         termsheets = (await db.execute(select(Termsheet).where(Termsheet.quote_id == quote.id))).scalars().all()
         events = (await db.execute(select(QuoteEvent).where(QuoteEvent.quote_id == quote.id))).scalars().all()
 
-        # clean up attachments in MinIO before cascade
+        # collect candidate material_ids for orphan cleanup before deleting quote
+        candidate_material_ids = [li.material_id for li in line_items if li.material_id is not None]
+
+        # collect attachment keys before deleting quote row
         from app.models.files import QuoteAttachment
         from app.services import storage
 
         attachments = (await db.execute(select(QuoteAttachment).where(QuoteAttachment.quote_id == quote.id))).scalars().all()
-        for att in attachments:
-            try:
-                await storage.remove_object(att.object_key)
-            except Exception:
-                logger.warning("Failed to remove MinIO object %s for attachment %s", att.object_key, att.id)
+        att_keys = [att.object_key for att in attachments]
+        quote_prefix = f"{quote.id}/"
 
         await db.delete(quote)
-        return {
-            "id": str(quote.id),
-            "deleted": True,
-            "cascaded": {
-                "line_items": len(line_items),
-                "termsheets": len(termsheets),
-                "events": len(events),
-                "decisions": len(decisions),
-                "receipts": len(receipts),
-                "attachments": len(attachments),
-            },
-        }
+        await db.flush()
+
+        # Purge any materials that no longer have any references in the system
+        purged_material_ids = await cleanup_orphan_materials(db, candidate_material_ids)
+
+    # Post-commit: clean up attachments in MinIO (best-effort; DB is already committed)
+    for key in att_keys:
+        try:
+            await storage.remove_object(key)
+        except Exception:
+            logger.warning("Failed to remove MinIO object %s for quote %s", key, quote_id)
+    try:
+        await storage.remove_prefix(quote_prefix)
+    except Exception:
+        logger.warning("Failed to remove MinIO prefix %s for quote %s", quote_prefix, quote_id)
+
+    return {
+        "id": str(quote_id),
+        "deleted": True,
+        "cascaded": {
+            "line_items": len(line_items),
+            "termsheets": len(termsheets),
+            "events": len(events),
+            "decisions": len(decisions),
+            "receipts": len(receipts),
+            "attachments": len(attachments),
+            "purged_orphan_materials": len(purged_material_ids),
+        },
+    }
+
+
 
 
 # ---------------------------------------------------------------------------