feat(catalog): supplier catalog cleanup, fuzzy matching, and navigation fixes

- Cleaned 137+ fake engine-displacement models from supplier imports
  (v3/v4 scripts: Chevrolet, Ford, Chrysler, Dodge, Jeep, Nissan, etc.)
- Removed 1,251+ corrupted models (INT. prefixes, year-suffix, torque specs,
  empty names, trailing-year variants)
- Migrated supplier tables to master DB (supplier_catalog,
  supplier_catalog_compat, supplier_catalog_interchange)
- Fixed _get_mye_ids_with_parts() to query supplier_catalog_compat from
  master DB so supplier-only vehicles appear for all tenants
- Added fuzzy model matcher with parenthesis stripping, noise suffix removal,
  compact matching, prefix/substring fallback, model aliases, and ±3 year
  proximity
- Matched compat rows: KEEP GREEN +14,152, KNADIAN +3,021, VAZLO +127,500,
  LUK +477, RAYBESTOS +1,743
- Added KNADIAN catalog importer with year-range expansion and future-year
  filtering
- Added VAZLO catalog importer with position parsing and SKU-in-model cleanup
- Added Keep Green, LUK, Yokomitsu, Raybestos catalog importers
- Cache clearing after cleanups (_classify_cache_*, nexus:mye_ids:*,
  nexus:brand_mye_counts:*)

Final match rates:
- KEEP GREEN: 90.3%
- VAZLO: 93.6%
- YOKOMITSU: 100.0%
- KNADIAN: 57.4%
- LUK: 51.0%
- RAYBESTOS: 55.9%

pos/services/catalog_import_service.py

@@ -0,0 +1,157 @@
+"""Bulk catalog import service.
+
+Imports products into inventory with optional vehicle compatibilities
+and SKU aliases. Can auto-generate vehicle fitment via QWEN AI if
+compatibilities are not provided.
+"""
+
+import logging
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+def import_products(
+    tenant_conn,
+    products: list[dict],
+    branch_id: int,
+    auto_generate_compat: bool = False,
+    employee_id: Optional[int] = None,
+):
+    """Import a list of products into inventory.
+
+    Each product dict may contain:
+      - sku (str) *required
+      - name (str) *required
+      - brand (str)
+      - description (str)
+      - cost (float)
+      - price (float)
+      - stock (int)
+      - location (str)
+      - sku_aliases (list[dict]) [{"sku": str, "label": str}]
+      - vehicles (list[dict]) [{"make", "model", "year", "engine", "engine_code"}]
+
+    Returns {"imported": N, "failed": [{"sku": ..., "error": ...}], "compat_generated": M}
+    """
+    cur = tenant_conn.cursor()
+    imported = 0
+    failed = []
+    compat_generated = 0
+
+    for idx, p in enumerate(products):
+        sku = (p.get("sku") or "").strip()
+        name = (p.get("name") or "").strip()
+        if not sku or not name:
+            failed.append({"index": idx, "sku": sku, "error": "sku and name are required"})
+            continue
+
+        brand = (p.get("brand") or "").strip() or None
+        description = (p.get("description") or "").strip() or None
+        cost = float(p.get("cost") or 0)
+        price = float(p.get("price") or 0)
+        stock = int(p.get("stock") or 0)
+        location = (p.get("location") or "").strip() or None
+        barcode = (p.get("barcode") or "").strip() or None
+
+        try:
+            # Check for duplicate SKU in same branch
+            cur.execute(
+                "SELECT id FROM inventory WHERE part_number = %s AND branch_id = %s AND is_active = true",
+                (sku, branch_id),
+            )
+            if cur.fetchone():
+                # Update existing item instead of creating new
+                cur.execute(
+                    """
+                    UPDATE inventory
+                    SET name = %s, brand = %s, description = %s, cost = %s, price_1 = %s,
+                        location = %s, barcode = COALESCE(%s, barcode), updated_at = NOW()
+                    WHERE part_number = %s AND branch_id = %s AND is_active = true
+                    RETURNING id
+                    """,
+                    (name, brand, description, cost, price, location, barcode, sku, branch_id),
+                )
+                row = cur.fetchone()
+                item_id = row[0]
+            else:
+                # Insert new item
+                cur.execute(
+                    """
+                    INSERT INTO inventory
+                        (branch_id, part_number, barcode, name, description, brand,
+                         unit, cost, price_1, price_2, price_3, tax_rate,
+                         min_stock, max_stock, location, is_active)
+                    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, true)
+                    RETURNING id
+                    """,
+                    (
+                        branch_id, sku, barcode, name, description, brand,
+                        "PZA", cost, price, price, price, 0.16,
+                        0, 0, location,
+                    ),
+                )
+                row = cur.fetchone()
+                item_id = row[0]
+
+                # Record initial stock if provided
+                if stock > 0:
+                    from services.inventory_engine import record_initial
+                    record_initial(tenant_conn, item_id, branch_id, stock, cost)
+
+            # Insert SKU aliases
+            aliases = p.get("sku_aliases") or []
+            for alias in aliases:
+                alias_sku = (alias.get("sku") or "").strip()
+                label = (alias.get("label") or "").strip() or None
+                if alias_sku:
+                    cur.execute(
+                        """
+                        INSERT INTO inventory_sku_aliases (inventory_id, sku, label)
+                        VALUES (%s, %s, %s)
+                        ON CONFLICT (inventory_id, sku) DO UPDATE SET
+                            is_active = true, label = EXCLUDED.label
+                        """,
+                        (item_id, alias_sku, label),
+                    )
+
+            # Insert manual vehicle compatibilities
+            vehicles = p.get("vehicles") or []
+            for v in vehicles:
+                make = (v.get("make") or "").strip()
+                model = (v.get("model") or "").strip()
+                year = v.get("year")
+                engine = (v.get("engine") or "").strip() or None
+                engine_code = (v.get("engine_code") or "").strip() or None
+                if make and model and year:
+                    cur.execute(
+                        """
+                        INSERT INTO inventory_vehicle_compat
+                            (inventory_id, make, model, year, engine, engine_code, source, model_year_engine_id)
+                        VALUES (%s, %s, %s, %s, %s, %s, 'manual', NULL)
+                        ON CONFLICT DO NOTHING
+                        """,
+                        (item_id, make, model, year, engine, engine_code),
+                    )
+
+            tenant_conn.commit()
+            imported += 1
+
+            # Auto-generate compat via QWEN if requested and no vehicles provided
+            if auto_generate_compat and not vehicles:
+                try:
+                    from services.qwen_fitment import get_vehicle_fitment
+                    from services.inventory_vehicle_compat import save_qwen_fitment
+                    fitment = get_vehicle_fitment(sku, name, brand or "")
+                    inserted = save_qwen_fitment(tenant_conn, item_id, fitment)
+                    compat_generated += inserted
+                except Exception as qe:
+                    logger.warning("QWEN auto-match failed for %s: %s", sku, qe)
+
+        except Exception as e:
+            tenant_conn.rollback()
+            logger.warning("Import failed for sku=%s: %s", sku, e)
+            failed.append({"index": idx, "sku": sku, "error": str(e)})
+
+    cur.close()
+    return {"imported": imported, "failed": failed, "compat_generated": compat_generated}

pos/services/dropshipping_service.py

@@ -0,0 +1,168 @@
+"""Dropshipping integration service.
+
+Provides read-only inventory access for external dropshipping platforms
+and webhook dispatching on stock/price/sale events.
+"""
+
+import logging
+from services.inventory_engine import get_stock_bulk
+
+logger = logging.getLogger(__name__)
+
+
+def resolve_tenant_by_api_key(master_conn, api_key: str):
+    """Find tenant_id and db_name for a given dropshipping API key.
+
+    Returns (tenant_id, db_name) or (None, None) if invalid.
+    """
+    if not api_key:
+        return None, None
+    cur = master_conn.cursor()
+    # tenant_config lives in each tenant DB, so we need to scan tenants
+    cur.execute("SELECT id, db_name FROM tenants WHERE is_active = true")
+    tenants = cur.fetchall()
+    for tid, db_name in tenants:
+        try:
+            tcur = master_conn.cursor()
+            # Use dblink or connect to tenant DB? Simpler: the blueprint
+            # will pass tenant_conn directly after resolution.
+            # Instead, we store a mapping in master DB for speed.
+            # For now, return all candidates and let caller validate.
+            pass
+        except Exception:
+            continue
+    cur.close()
+    return None, None
+
+
+def _get_dropshipping_key(tenant_conn):
+    cur = tenant_conn.cursor()
+    cur.execute("SELECT value FROM tenant_config WHERE key = 'dropshipping_api_key'")
+    row = cur.fetchone()
+    cur.close()
+    return row[0] if row else None
+
+
+def validate_api_key(tenant_conn, api_key: str) -> bool:
+    """Check if the provided API key matches the tenant's configured key."""
+    if not api_key:
+        return False
+    expected = _get_dropshipping_key(tenant_conn)
+    return expected is not None and expected == api_key
+
+
+def get_inventory_list(tenant_conn, search: str = None, page: int = 1, per_page: int = 50):
+    """Return inventory items with stock and price for dropshipping."""
+    offset = (max(page, 1) - 1) * per_page
+    stock_map = get_stock_bulk(tenant_conn, branch_id=None)
+
+    cur = tenant_conn.cursor()
+    params = []
+    where = "WHERE is_active = true"
+    if search:
+        where += " AND (name ILIKE %s OR part_number ILIKE %s)"
+        params.extend([f"%{search}%", f"%{search}%"])
+
+    cur.execute(
+        f"""
+        SELECT id, part_number, name, brand, price_1, price_2, price_3,
+               image_url, unit, description
+        FROM inventory
+        {where}
+        ORDER BY id DESC
+        LIMIT %s OFFSET %s
+        """,
+        params + [per_page, offset],
+    )
+    rows = cur.fetchall()
+
+    # Count total
+    cur.execute(f"SELECT COUNT(*) FROM inventory {where}", params)
+    total = cur.fetchone()[0]
+    cur.close()
+
+    items = []
+    for r in rows:
+        inv_id = r[0]
+        items.append({
+            "id": inv_id,
+            "sku": r[1],
+            "name": r[2],
+            "brand": r[3],
+            "price_1": float(r[4]) if r[4] else None,
+            "price_2": float(r[5]) if r[5] else None,
+            "price_3": float(r[6]) if r[6] else None,
+            "stock": stock_map.get(inv_id, 0),
+            "image_url": r[7],
+            "unit": r[8],
+            "description": r[9],
+        })
+    return {"items": items, "page": page, "per_page": per_page, "total": total}
+
+
+def get_inventory_by_sku(tenant_conn, sku: str):
+    """Return a single inventory item by SKU/part_number."""
+    stock_map = get_stock_bulk(tenant_conn, branch_id=None)
+    cur = tenant_conn.cursor()
+    cur.execute(
+        """
+        SELECT id, part_number, name, brand, price_1, price_2, price_3,
+               image_url, unit, description
+        FROM inventory
+        WHERE part_number = %s AND is_active = true
+        LIMIT 1
+        """,
+        (sku,),
+    )
+    row = cur.fetchone()
+    cur.close()
+    if not row:
+        return None
+    inv_id = row[0]
+    return {
+        "id": inv_id,
+        "sku": row[1],
+        "name": row[2],
+        "brand": row[3],
+        "price_1": float(row[4]) if row[4] else None,
+        "price_2": float(row[5]) if row[5] else None,
+        "price_3": float(row[6]) if row[6] else None,
+        "stock": stock_map.get(inv_id, 0),
+        "image_url": row[7],
+        "unit": row[8],
+        "description": row[9],
+    }
+
+
+def get_stock_by_skus(tenant_conn, skus: list[str]) -> dict:
+    """Return stock levels for a list of SKUs."""
+    stock_map = get_stock_bulk(tenant_conn, branch_id=None)
+    cur = tenant_conn.cursor()
+    cur.execute(
+        """
+        SELECT id, part_number FROM inventory
+        WHERE part_number = ANY(%s) AND is_active = true
+        """,
+        (skus,),
+    )
+    rows = cur.fetchall()
+    cur.close()
+    result = {}
+    for inv_id, sku in rows:
+        result[sku] = stock_map.get(inv_id, 0)
+    return result
+
+
+def get_webhook_targets(tenant_conn, event_type: str) -> list[str]:
+    """Return active webhook URLs for a given event type."""
+    cur = tenant_conn.cursor()
+    cur.execute(
+        """
+        SELECT target_url FROM dropshipping_webhooks
+        WHERE event_type = %s AND is_active = true
+        """,
+        (event_type,),
+    )
+    urls = [r[0] for r in cur.fetchall()]
+    cur.close()
+    return urls

pos/services/marketplace_external_service.py

@@ -9,6 +9,7 @@ Depends on:
 
 import json
 import logging
+import urllib.parse
 from typing import Optional
 from decimal import Decimal
 
@@ -19,6 +20,19 @@ from services.inventory_engine import get_stock, get_stock_bulk
 logger = logging.getLogger(__name__)
 
 
+def _resolve_image_urls(images: list[str], base_url: str | None = None) -> list[str]:
+    """Convert relative image paths to absolute URLs."""
+    resolved = []
+    for url in images:
+        if not url:
+            continue
+        if base_url and url.startswith("/"):
+            resolved.append(urllib.parse.urljoin(base_url.rstrip("/") + "/", url.lstrip("/")))
+        else:
+            resolved.append(url)
+    return resolved
+
+
 # ═══════════════════════════════════════════════════════════════════════════
 # CONFIG HELPERS
 # ═══════════════════════════════════════════════════════════════════════════
@@ -145,6 +159,7 @@ def build_item_payload(
     custom_title: str = None,
     extra_attributes: list = None,
     shipping_cost: float = None,
+    base_url: str = None,
 ) -> dict:
     """Convert a Nexus inventory row into a MercadoLibre item payload."""
     title = custom_title or f"{inventory_row['name']} {inventory_row['brand'] or ''} {inventory_row['part_number'] or ''}".strip()
@@ -167,17 +182,20 @@ def build_item_payload(
         "buying_mode": "buy_it_now",
         "listing_type_id": listing_type_id,
         "condition": "new",
-        "pictures": [{"source": url} for url in images if url],
+        "pictures": [{"source": url} for url in _resolve_image_urls(images, base_url) if url],
         "shipping": shipping_payload,
         "attributes": [],
     }
 
-    if inventory_row.get("brand"):
+    # Collect extra attribute IDs to avoid duplicates
+    extra_attr_ids = {a.get("id") for a in (extra_attributes or []) if a.get("id")}
+
+    if inventory_row.get("brand") and "BRAND" not in extra_attr_ids:
         payload["attributes"].append(
             {"id": "BRAND", "value_name": inventory_row["brand"]}
         )
 
-    if inventory_row.get("part_number"):
+    if inventory_row.get("part_number") and "PART_NUMBER" not in extra_attr_ids:
         payload["attributes"].append(
             {"id": "PART_NUMBER", "value_name": inventory_row["part_number"]}
         )
@@ -193,11 +211,11 @@ def build_item_payload(
         if isinstance(vehicle_compat, list) and vehicle_compat:
             first = vehicle_compat[0]
             if isinstance(first, dict):
-                if first.get("brand"):
+                if first.get("brand") and "VEHICLE_MODEL" not in extra_attr_ids:
                     payload["attributes"].append(
                         {"id": "VEHICLE_MODEL", "value_name": first["brand"]}
                     )
-                if first.get("model"):
+                if first.get("model") and "VEHICLE_MODEL_NAME" not in extra_attr_ids:
                     payload["attributes"].append(
                         {"id": "VEHICLE_MODEL_NAME", "value_name": first["model"]}
                     )
@@ -243,7 +261,7 @@ def check_meli_shipping_config(svc: MeliService, cfg: dict) -> dict:
 # LISTINGS CRUD
 # ═══════════════════════════════════════════════════════════════════════════
 
-def check_inventory_ml_status(tenant_conn, inventory_ids: list[int]) -> dict:
+def check_inventory_ml_status(tenant_conn, inventory_ids: list[int], base_url: str = None) -> dict:
     """Check local pre-flight status for ML publishing.
 
     Returns per-item dict with checks: has_image, has_stock, has_price,
@@ -298,7 +316,7 @@ def check_inventory_ml_status(tenant_conn, inventory_ids: list[int]) -> dict:
             "has_price": price > 0,
             "price": price,
             "stock": stock,
-            "image_url": image_url,
+            "image_url": _resolve_image_urls([image_url], base_url)[0] if image_url else None,
             "already_published": existing is not None,
             "existing_listing": existing,
         })
@@ -312,6 +330,7 @@ def validate_items(
     listing_type_id: str = "gold_special",
     shipping_mode: str = "me2",
     custom_data: dict = None,
+    base_url: str = None,
 ) -> dict:
     """Validate items against ML /items/validate without creating them.
 
@@ -370,6 +389,7 @@ def validate_items(
         images = []
         if inv.get("image_url"):
             images.append(inv["image_url"])
+        images = _resolve_image_urls(images, base_url)
         if not images:
             results["invalid"].append({"inventory_id": inv_id, "error": "El producto no tiene imagen"})
             continue
@@ -385,6 +405,7 @@ def validate_items(
             shipping_mode=shipping_mode, listing_type_id=listing_type_id,
             custom_title=title, extra_attributes=extra_attrs,
             shipping_cost=shipping_cost,
+            base_url=base_url,
         )
 
         try:
@@ -412,6 +433,7 @@ def publish_items(
     listing_type_id: str = "gold_special",
     shipping_mode: str = "me2",
     custom_data: dict = None,
+    base_url: str = None,
 ) -> dict:
     """Publish one or more inventory items to MercadoLibre.
 
@@ -476,6 +498,7 @@ def publish_items(
         images = []
         if inv.get("image_url"):
             images.append(inv["image_url"])
+        images = _resolve_image_urls(images, base_url)
 
         if not images:
             results["failed"].append({"inventory_id": inv_id, "error": "El producto no tiene imagen. ML requiere imagen para publicar."})
@@ -492,6 +515,7 @@ def publish_items(
             shipping_mode=shipping_mode, listing_type_id=listing_type_id,
             custom_title=title, extra_attributes=extra_attrs,
             shipping_cost=shipping_cost,
+            base_url=base_url,
         )
 
         try:
@@ -740,6 +764,239 @@ def close_listing(tenant_conn, listing_id: int) -> dict:
     return {"ok": True, "status": "closed"}
 
 
+def delete_listing_permanently(tenant_conn, listing_id: int) -> dict:
+    """Hard-delete a closed listing from the local DB.
+
+    Sets listing_id = NULL on marketplace_order_items to avoid FK errors,
+    then deletes the marketplace_listings row.
+    """
+    cur = tenant_conn.cursor()
+    cur.execute(
+        "SELECT id, external_status FROM marketplace_listings WHERE id = %s",
+        (listing_id,),
+    )
+    row = cur.fetchone()
+    if not row:
+        cur.close()
+        raise ValueError("Listing not found")
+
+    # Clear FK references so we can delete safely
+    cur.execute(
+        "UPDATE marketplace_order_items SET listing_id = NULL WHERE listing_id = %s",
+        (listing_id,),
+    )
+    cur.execute(
+        "DELETE FROM marketplace_listings WHERE id = %s",
+        (listing_id,),
+    )
+    tenant_conn.commit()
+    cur.close()
+    return {"ok": True, "deleted": True}
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# QUESTIONS & ANSWERS
+# ═══════════════════════════════════════════════════════════════════════════
+
+def _upsert_question(cur, q_data: dict, listing_id_map: dict):
+    """Upsert a single question from ML API into marketplace_questions."""
+    external_qid = str(q_data.get("id"))
+    external_item_id = str(q_data.get("item_id"))
+    text = q_data.get("text", "")
+    status = q_data.get("status", "unanswered")
+    answer = q_data.get("answer", {})
+    answer_text = answer.get("text") if answer else None
+    answer_date = None
+    if answer and answer.get("date_created"):
+        answer_date = answer["date_created"]
+    from_user = q_data.get("from", {})
+    buyer_id = str(from_user.get("id")) if from_user else None
+    buyer_nickname = from_user.get("nickname") if from_user else None
+    question_date = q_data.get("date_created")
+    listing_id = listing_id_map.get(external_item_id)
+
+    cur.execute(
+        """
+        INSERT INTO marketplace_questions
+            (listing_id, external_question_id, external_item_id, question_text,
+             answer_text, status, buyer_id, buyer_nickname, question_date,
+             answer_date, raw_json, updated_at)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())
+        ON CONFLICT (external_question_id)
+        DO UPDATE SET
+            question_text = EXCLUDED.question_text,
+            answer_text = EXCLUDED.answer_text,
+            status = EXCLUDED.status,
+            buyer_id = EXCLUDED.buyer_id,
+            buyer_nickname = EXCLUDED.buyer_nickname,
+            question_date = EXCLUDED.question_date,
+            answer_date = EXCLUDED.answer_date,
+            raw_json = EXCLUDED.raw_json,
+            updated_at = NOW()
+        """,
+        (
+            listing_id,
+            external_qid,
+            external_item_id,
+            text,
+            answer_text,
+            status,
+            buyer_id,
+            buyer_nickname,
+            question_date,
+            answer_date,
+            json.dumps(q_data),
+        ),
+    )
+
+
+def sync_questions(tenant_conn) -> dict:
+    """Fetch questions from ML for all active listings and upsert locally.
+
+    Returns {"synced": N, "items": [...]}.
+    """
+    cfg = get_meli_config(tenant_conn)
+    svc = _get_meli_service(cfg)
+    if not svc:
+        raise ValueError("MercadoLibre not configured")
+
+    cur = tenant_conn.cursor()
+    cur.execute(
+        """
+        SELECT id, external_item_id
+        FROM marketplace_listings
+        WHERE channel = 'mercadolibre' AND is_active = true
+        """
+    )
+    listings = {r[1]: r[0] for r in cur.fetchall()}
+    cur.close()
+
+    if not listings:
+        return {"synced": 0, "items": []}
+
+    total_synced = 0
+    for item_id in listings:
+        try:
+            resp = svc.get_questions(item_id, limit=50)
+            questions = resp.get("questions", [])
+            if not questions:
+                continue
+            cur = tenant_conn.cursor()
+            for q in questions:
+                _upsert_question(cur, q, listings)
+            tenant_conn.commit()
+            cur.close()
+            total_synced += len(questions)
+        except Exception as e:
+            logger.warning("Failed to sync questions for item %s: %s", item_id, e)
+
+    return {"synced": total_synced}
+
+
+def fetch_question_from_ml(tenant_conn, external_question_id: str) -> dict:
+    """Fetch a single question from ML API and upsert locally."""
+    cfg = get_meli_config(tenant_conn)
+    svc = _get_meli_service(cfg)
+    if not svc:
+        raise ValueError("MercadoLibre not configured")
+
+    q_data = svc.get_question(external_question_id)
+    cur = tenant_conn.cursor()
+    cur.execute(
+        "SELECT id, external_item_id FROM marketplace_listings WHERE external_item_id = %s",
+        (str(q_data.get("item_id")),),
+    )
+    row = cur.fetchone()
+    listing_id_map = {row[1]: row[0]} if row else {}
+    _upsert_question(cur, q_data, listing_id_map)
+    tenant_conn.commit()
+    cur.close()
+    return q_data
+
+
+def answer_question(tenant_conn, local_question_id: int, text: str) -> dict:
+    """Answer a question via ML API and update local status."""
+    cfg = get_meli_config(tenant_conn)
+    svc = _get_meli_service(cfg)
+    if not svc:
+        raise ValueError("MercadoLibre not configured")
+
+    cur = tenant_conn.cursor()
+    cur.execute(
+        "SELECT external_question_id FROM marketplace_questions WHERE id = %s",
+        (local_question_id,),
+    )
+    row = cur.fetchone()
+    if not row:
+        cur.close()
+        raise ValueError("Question not found")
+    external_qid = row[0]
+    cur.close()
+
+    resp = svc.answer_question(external_qid, text)
+
+    cur = tenant_conn.cursor()
+    cur.execute(
+        """
+        UPDATE marketplace_questions
+        SET answer_text = %s, status = 'answered', answer_date = NOW(), updated_at = NOW()
+        WHERE id = %s
+        """,
+        (text, local_question_id),
+    )
+    tenant_conn.commit()
+    cur.close()
+    return {"ok": True, "ml_response": resp}
+
+
+def list_local_questions(tenant_conn, status: str = None) -> list:
+    """Return questions from local DB, optionally filtered by status."""
+    cur = tenant_conn.cursor()
+    if status:
+        cur.execute(
+            """
+            SELECT q.id, q.external_question_id, q.external_item_id, q.question_text,
+                   q.answer_text, q.status, q.buyer_nickname, q.question_date,
+                   q.answer_date, q.created_at, l.title, l.external_permalink
+            FROM marketplace_questions q
+            LEFT JOIN marketplace_listings l ON l.id = q.listing_id
+            WHERE q.status = %s
+            ORDER BY q.question_date DESC
+            """,
+            (status,),
+        )
+    else:
+        cur.execute(
+            """
+            SELECT q.id, q.external_question_id, q.external_item_id, q.question_text,
+                   q.answer_text, q.status, q.buyer_nickname, q.question_date,
+                   q.answer_date, q.created_at, l.title, l.external_permalink
+            FROM marketplace_questions q
+            LEFT JOIN marketplace_listings l ON l.id = q.listing_id
+            ORDER BY q.question_date DESC
+            """
+        )
+    rows = cur.fetchall()
+    cur.close()
+    results = []
+    for r in rows:
+        results.append({
+            "id": r[0],
+            "external_question_id": r[1],
+            "external_item_id": r[2],
+            "question_text": r[3],
+            "answer_text": r[4],
+            "status": r[5],
+            "buyer_nickname": r[6],
+            "question_date": r[7],
+            "answer_date": r[8],
+            "created_at": r[9],
+            "listing_title": r[10],
+            "listing_permalink": r[11],
+        })
+    return results
+
+
 # ═══════════════════════════════════════════════════════════════════════════
 # ORDERS
 # ═══════════════════════════════════════════════════════════════════════════

pos/services/meli_service.py

@@ -167,6 +167,23 @@ class MeliService:
     def close_item(self, item_id: str) -> dict:
         return self.update_item(item_id, {"status": "closed"})
 
+    # ─── Questions & Answers ─────────────────────────────────────────────
+
+    def get_questions(self, item_id: str, status: str = None, offset: int = 0, limit: int = 50) -> dict:
+        params = {"item_id": item_id, "offset": offset, "limit": limit}
+        if status:
+            params["status"] = status
+        return self._request("GET", "/questions/search", params=params)
+
+    def get_question(self, question_id: str) -> dict:
+        return self._request("GET", f"/questions/{question_id}")
+
+    def answer_question(self, question_id: str, text: str) -> dict:
+        return self._request("POST", "/answers", json_payload={"question_id": question_id, "text": text})
+
+    def delete_question(self, question_id: str) -> dict:
+        return self._request("DELETE", f"/questions/{question_id}")
+
     # ─── Categories ──────────────────────────────────────────────────────
 
     def get_category(self, category_id: str) -> dict:

pos/services/pos_engine.py

@@ -447,6 +447,35 @@ def process_sale(conn, sale_data):
     except Exception:
         pass  # Learning errors never block sales
 
+    # Dropshipping webhook hook (non-blocking)
+    try:
+        from services import dropshipping_service as ds_svc
+        from services.webhook_service import dispatch_webhooks_bulk
+        webhook_urls = ds_svc.get_webhook_targets(conn, 'sale_made')
+        if webhook_urls:
+            payload_items = []
+            for item in enriched_items:
+                remaining = item['stock_before'] - item['quantity']
+                payload_items.append({
+                    'sku': item['part_number'],
+                    'name': item['name'],
+                    'quantity_sold': item['quantity'],
+                    'stock_remaining': remaining,
+                    'unit_price': item['unit_price'],
+                })
+            threading.Thread(
+                target=dispatch_webhooks_bulk,
+                args=(webhook_urls, 'sale_made', {
+                    'sale_id': sale_id,
+                    'items': payload_items,
+                    'total': totals['total'],
+                    'created_at': str(created_at),
+                }),
+                daemon=True
+            ).start()
+    except Exception:
+        pass  # Webhook errors never block sales
+
     return {
         'id': sale_id,
         'branch_id': branch_id,

pos/services/webhook_service.py

@@ -0,0 +1,65 @@
+"""Webhook dispatch service for dropshipping and external integrations.
+
+Sends POST requests to configured target URLs with retry logic.
+Can be called synchronously or enqueued via Celery.
+"""
+
+import json
+import logging
+import requests
+import threading
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+def _send_post(url: str, payload: dict, headers: Optional[dict] = None, timeout: int = 10):
+    """Send a POST request and return (success, status_code, response_text)."""
+    default_headers = {"Content-Type": "application/json"}
+    if headers:
+        default_headers.update(headers)
+    try:
+        resp = requests.post(url, json=payload, headers=default_headers, timeout=timeout)
+        success = 200 <= resp.status_code < 300
+        if not success:
+            logger.warning("Webhook %s returned %s: %s", url, resp.status_code, resp.text[:200])
+        return success, resp.status_code, resp.text
+    except requests.exceptions.Timeout:
+        logger.warning("Webhook %s timed out after %ss", url, timeout)
+        return False, 0, "timeout"
+    except Exception as e:
+        logger.warning("Webhook %s failed: %s", url, e)
+        return False, 0, str(e)
+
+
+def dispatch_webhook_sync(target_url: str, event_type: str, payload: dict, secret: Optional[str] = None):
+    """Send webhook synchronously (use inside Celery tasks for async)."""
+    full_payload = {
+        "event": event_type,
+        "data": payload,
+    }
+    headers = {}
+    if secret:
+        headers["X-Webhook-Secret"] = secret
+    success, status, body = _send_post(target_url, full_payload, headers=headers)
+    return {"success": success, "status": status, "body": body[:500]}
+
+
+def dispatch_webhooks_bulk(target_urls: list[str], event_type: str, payload: dict, secret: Optional[str] = None):
+    """Dispatch to multiple URLs concurrently using threads."""
+    results = []
+    threads = []
+
+    def _send(url):
+        result = dispatch_webhook_sync(url, event_type, payload, secret=secret)
+        results.append({"url": url, **result})
+
+    for url in target_urls:
+        t = threading.Thread(target=_send, args=(url,))
+        t.start()
+        threads.append(t)
+
+    for t in threads:
+        t.join(timeout=15)
+
+    return results