Autoparts-DB/pos/services/b2b_service.py

# /home/Autopartes/pos/services/b2b_service.py
"""Nexus B2B yonke marketplace: wholesale used parts between deshuesaderos.

A yonke (seller) publishes its used/rebuild pieces to a shared catalog; another
yonke (buyer) places an order. Listings live in the MASTER DB (nexus_autoparts)
so they are visible cross-tenant; the pieces themselves stay in each seller's
tenant DB.

Tables (master DB):
  b2b_listings          — published pieces (UNIQUE seller_tenant_id+inventory_id)
  marketplace_orders    — order header (buyer/seller tenant ids, total, status)
  marketplace_order_items — order lines (+ listing_id link)

Order lifecycle:
  pending → confirmed → shipped → received → completed
  pending/confirmed → cancelled

Actor permissions per transition:
  buyer  : cancelled (from pending), received (from confirmed/shipped)
  seller : confirmed (from pending), shipped (from pending/confirmed),
           cancelled (from pending/confirmed)
  both   : completed (from received)

The source states are deliberately permissive (the UI exposes "Marcar
enviado"/"Recibido" a step early) so a valid actor never hits a dead end.

Conventions:
  * Master reads/writes go through the `master_conn` passed by the caller
    (the blueprint owns commit/rollback), EXCEPT `update_order_status`, which
    commits the master transaction itself: on `received` it must persist the
    status BEFORE running the cross-tenant side effects, so the order state can
    never be rolled back while pieces were already created in another tenant.
  * Tenant connections are opened/committed/closed inside this module, because
    one order touches two tenants (buyer + seller) at once.
  * Everything degrades: side effects at `received` run inside SAVEPOINTs and
    never abort the order on failure — failures are returned as `warnings`.
"""

from decimal import Decimal

from psycopg2 import errors as pg_errors
from psycopg2.extras import RealDictCursor
from tenant_db import get_tenant_conn

# ─── Domain constants ──────────────────────────────────────────────────────

LISTING_STATUSES = ('available', 'reserved', 'sold', 'removed')
ORDER_STATUSES = ('pending', 'confirmed', 'shipped', 'received', 'completed', 'cancelled')

# status -> {new_status: (allowed actor roles)}
ORDER_TRANSITIONS = {
    'pending':   {'confirmed': ('seller',),
                  'shipped':   ('seller',),
                  'cancelled': ('buyer', 'seller')},
    'confirmed': {'shipped':   ('seller',),
                  'received':  ('buyer',),
                  'cancelled': ('seller',)},
    'shipped':   {'received': ('buyer',)},
    'received':  {'completed': ('buyer', 'seller')},
    'completed': {},
    'cancelled': {},
}

USED_CONDITIONS = ('usado', 'reconstruido')


class B2BError(Exception):
    """Base class for all B2B errors (mapped to HTTP by the blueprint)."""


class B2BValidationError(B2BError):
    """Bad input / illegal state transition -> HTTP 400."""


class B2BForbidden(B2BError):
    """Caller is not allowed -> HTTP 403."""


class B2BNotFound(B2BError):
    """Entity missing -> HTTP 404."""


# ─── Small helpers ─────────────────────────────────────────────────────────

def _to_int(val, default=None):
    if val is None or val == '':
        return default
    try:
        return int(float(str(val).replace(',', '')))
    except (TypeError, ValueError):
        return default


def _to_float(val, default=0.0):
    if val is None or val == '':
        return default
    try:
        return float(str(val).replace(',', ''))
    except (TypeError, ValueError):
        return default


def _jval(v):
    """Make a psycopg2 value JSON-safe (Decimal -> float, date -> isoformat)."""
    if isinstance(v, Decimal):
        return float(v)
    if hasattr(v, 'isoformat'):
        return v.isoformat()
    return v


def _clean(d):
    return {k: _jval(v) for k, v in (d or {}).items()}


def _cur(conn):
    return conn.cursor(cursor_factory=RealDictCursor)


def _tenant_name(master_conn, tenant_id):
    """Best-effort tenant display name from the master DB (never raises)."""
    if not tenant_id:
        return None
    try:
        cur = _cur(master_conn)
        cur.execute("SELECT name FROM tenants WHERE id = %s", (tenant_id,))
        row = cur.fetchone()
        cur.close()
        return row['name'] if row else None
    except Exception:
        try:
            master_conn.rollback()
        except Exception:
            pass
        return None


def is_seller(master_conn, tenant_id):
    """True when the tenant has the seller flag enabled in the master DB."""
    try:
        cur = _cur(master_conn)
        cur.execute("SELECT COALESCE(is_seller, false) AS is_seller FROM tenants WHERE id = %s",
                    (tenant_id,))
        row = cur.fetchone()
        cur.close()
        return bool(row['is_seller']) if row else False
    except Exception:
        try:
            master_conn.rollback()
        except Exception:
            pass
        return False


def _audit_best_effort(conn, action, entity_type=None, entity_id=None, new_value=None):
    try:
        from services.audit import log_action
        log_action(conn, action, entity_type, entity_id, new_value=new_value)
    except Exception:
        pass


# ─── Listings: serialization ───────────────────────────────────────────────

_LISTING_FIELDS = (
    'id', 'seller_tenant_id', 'seller_inventory_id', 'part_number', 'name',
    'condition', 'condition_grade', 'price_wholesale', 'donor_make',
    'donor_model', 'donor_year', 'image_url', 'status', 'notes',
    'created_at', 'updated_at',
)
_LISTING_COLS = ', '.join(_LISTING_FIELDS)
_LISTING_COLS_L = ', '.join('l.' + f for f in _LISTING_FIELDS)


def _listing_dict(row, seller_name=None, include_private=False):
    """Map a b2b_listings row to the API dict.

    `inventory_id` is the public alias of `seller_inventory_id` (the contract
    field). Extra fields (condition, notes, seller_tenant_id, timestamps) are
    additive and harmless for the UI.
    """
    d = _clean(dict(row))
    d['inventory_id'] = d.get('seller_inventory_id')
    d['price_wholesale'] = float(d.get('price_wholesale') or 0)
    if seller_name is not None:
        d['seller_name'] = seller_name
    if not include_private:
        d.pop('seller_tenant_id', None)
    return d


# ─── Seller: my listings ───────────────────────────────────────────────────

def list_my_listings(master_conn, tenant_id, status=None):
    """Listings published by `tenant_id` (removed ones hidden by default)."""
    cur = _cur(master_conn)
    where = ["seller_tenant_id = %s"]
    params = [tenant_id]
    if status:
        status = (status or '').strip().lower()
        if status not in LISTING_STATUSES:
            raise B2BValidationError(f"status inválido: {status}")
        where.append("status = %s")
        params.append(status)
    else:
        where.append("status <> 'removed'")
    cur.execute(
        f"SELECT {_LISTING_COLS} FROM b2b_listings "
        f"WHERE {' AND '.join(where)} ORDER BY created_at DESC, id DESC",
        params,
    )
    rows = cur.fetchall()
    cur.close()
    return [_listing_dict(r, include_private=True) for r in rows]


def get_listing(master_conn, listing_id):
    """Single listing + seller_name, or None."""
    cur = _cur(master_conn)
    cur.execute(f"SELECT {_LISTING_COLS} FROM b2b_listings WHERE id = %s", (listing_id,))
    row = cur.fetchone()
    cur.close()
    if not row:
        return None
    return _listing_dict(row, seller_name=_tenant_name(master_conn, row['seller_tenant_id']))


# ─── Seller: publish / edit / remove ───────────────────────────────────────

def _fetch_piece(tenant_id, inventory_id):
    """Read the source piece from the seller's tenant DB (+ primary image)."""
    conn = get_tenant_conn(tenant_id)
    try:
        cur = conn.cursor(cursor_factory=RealDictCursor)
        cur.execute("""
            SELECT id, part_number, name, condition, condition_grade,
                   donor_make, donor_model, donor_year, image_url, is_active
            FROM inventory WHERE id = %s
        """, (inventory_id,))
        row = cur.fetchone()
        if row:
            cur.execute("""
                SELECT url FROM inventory_images
                WHERE inventory_id = %s
                ORDER BY is_primary DESC, sort_order ASC, id ASC
                LIMIT 1
            """, (inventory_id,))
            img = cur.fetchone()
            if img and img.get('url'):
                row = dict(row)
                row['image_url'] = img['url']
        cur.close()
        return dict(row) if row else None
    finally:
        conn.close()


def publish_listing(master_conn, tenant_id, data):
    """Create or refresh a listing for one of the tenant's used pieces.

    Raises B2BForbidden when the tenant is not a seller, B2BValidationError on
    bad input / ineligible piece, B2BNotFound when the piece does not exist.
    """
    data = data or {}
    if not is_seller(master_conn, tenant_id):
        raise B2BForbidden(
            'Tu cuenta no está habilitada como vendedor (is_seller). '
            'Contacta al administrador para activar el modo mayoreo.'
        )

    inventory_id = _to_int(data.get('inventory_id'))
    if not inventory_id:
        raise B2BValidationError('inventory_id requerido')

    if 'price_wholesale' not in data or data.get('price_wholesale') in (None, ''):
        raise B2BValidationError('price_wholesale requerido')
    price = _to_float(data.get('price_wholesale'), None)
    if price is None or price < 0:
        raise B2BValidationError('price_wholesale inválido')

    piece = _fetch_piece(tenant_id, inventory_id)
    if not piece:
        raise B2BNotFound('Pieza no encontrada')
    condition = (piece.get('condition') or '').strip().lower()
    if condition not in USED_CONDITIONS:
        raise B2BValidationError(
            "Solo se pueden publicar piezas usadas o reconstruidas "
            f"(condition actual: {condition or 'desconocida'})"
        )
    if piece.get('is_active') is False:
        raise B2BValidationError('La pieza no está activa en inventario')

    notes = (data.get('notes') or '').strip() or None
    cur = _cur(master_conn)
    cur.execute("""
        INSERT INTO b2b_listings
            (seller_tenant_id, seller_inventory_id, part_number, name, condition,
             condition_grade, price_wholesale, donor_make, donor_model, donor_year,
             image_url, status, notes)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'available', %s)
        ON CONFLICT (seller_tenant_id, seller_inventory_id) DO UPDATE SET
            part_number     = EXCLUDED.part_number,
            name            = EXCLUDED.name,
            condition       = EXCLUDED.condition,
            condition_grade = EXCLUDED.condition_grade,
            price_wholesale = EXCLUDED.price_wholesale,
            donor_make      = EXCLUDED.donor_make,
            donor_model     = EXCLUDED.donor_model,
            donor_year      = EXCLUDED.donor_year,
            image_url       = EXCLUDED.image_url,
            notes           = COALESCE(EXCLUDED.notes, b2b_listings.notes),
            status          = 'available',
            updated_at      = NOW()
        RETURNING id
    """, (
        tenant_id, inventory_id, piece.get('part_number'), piece.get('name'),
        condition or 'usado', piece.get('condition_grade'), price,
        piece.get('donor_make'), piece.get('donor_model'), piece.get('donor_year'),
        piece.get('image_url'), notes,
    ))
    listing_id = cur.fetchone()['id']
    cur.close()

    listing = get_listing(master_conn, listing_id)
    _audit_publish_best_effort(tenant_id, listing)
    return listing


def _audit_publish_best_effort(tenant_id, listing):
    try:
        conn = get_tenant_conn(tenant_id)
        try:
            _audit_best_effort(conn, 'B2B_LISTING_PUBLISH', 'b2b_listing',
                               (listing or {}).get('id'),
                               {'price_wholesale': (listing or {}).get('price_wholesale')})
            conn.commit()
        finally:
            conn.close()
    except Exception:
        pass


def update_listing(master_conn, tenant_id, listing_id, data):
    """Update price/status of a listing (owner only). Returns the listing."""
    data = data or {}
    sets = []
    params = []
    if 'price_wholesale' in data:
        price = _to_float(data.get('price_wholesale'), None)
        if price is None or price < 0:
            raise B2BValidationError('price_wholesale inválido')
        sets.append('price_wholesale = %s')
        params.append(price)
    if 'status' in data:
        status = (data.get('status') or '').strip().lower()
        if status not in LISTING_STATUSES:
            raise B2BValidationError(f"status inválido: {status}")
        sets.append('status = %s')
        params.append(status)
    if 'notes' in data:
        sets.append('notes = %s')
        params.append((data.get('notes') or '').strip() or None)
    if not sets:
        raise B2BValidationError('Nada que actualizar (price_wholesale, status, notes)')

    cur = _cur(master_conn)
    cur.execute("SELECT seller_tenant_id FROM b2b_listings WHERE id = %s", (listing_id,))
    row = cur.fetchone()
    if not row:
        cur.close()
        raise B2BNotFound('Listing no encontrado')
    if row['seller_tenant_id'] != tenant_id:
        cur.close()
        raise B2BForbidden('Este listing pertenece a otro vendedor')

    sets.append('updated_at = NOW()')
    params.append(listing_id)
    cur.execute(f"UPDATE b2b_listings SET {', '.join(sets)} WHERE id = %s", params)
    cur.close()
    return get_listing(master_conn, listing_id)


def remove_listing(master_conn, tenant_id, listing_id):
    """Soft-delete a listing (status='removed'). Owner only. Returns bool."""
    cur = _cur(master_conn)
    cur.execute("SELECT seller_tenant_id FROM b2b_listings WHERE id = %s", (listing_id,))
    row = cur.fetchone()
    if not row:
        cur.close()
        raise B2BNotFound('Listing no encontrado')
    if row['seller_tenant_id'] != tenant_id:
        cur.close()
        raise B2BForbidden('Este listing pertenece a otro vendedor')
    cur.execute(
        "UPDATE b2b_listings SET status = 'removed', updated_at = NOW() WHERE id = %s",
        (listing_id,),
    )
    cur.close()
    return True


# ─── Buyer: catalog ────────────────────────────────────────────────────────

def list_catalog(master_conn, buyer_tenant_id, filters=None, page=1, per_page=12):
    """Available listings from OTHER tenants, with seller_name and pagination."""
    filters = filters or {}
    page = max(_to_int(page, 1) or 1, 1)
    per_page = min(max(_to_int(per_page, 12) or 12, 1), 100)

    where = ["l.status = 'available'", 'l.seller_tenant_id <> %s']
    params = [buyer_tenant_id]

    q = (filters.get('q') or '').strip()
    if q:
        where.append("(l.name ILIKE %s OR l.part_number ILIKE %s "
                     "OR l.donor_make ILIKE %s OR l.donor_model ILIKE %s)")
        params.extend([f'%{q}%'] * 4)

    for field, col in (('make', 'l.donor_make'), ('model', 'l.donor_model')):
        val = (filters.get(field) or '').strip()
        if val:
            where.append(f"{col} ILIKE %s")
            params.append(f'%{val}%')

    year = _to_int(filters.get('year'))
    if year:
        where.append("l.donor_year = %s")
        params.append(year)

    grade = (filters.get('grade') or '').strip().upper()
    if grade:
        where.append("UPPER(l.condition_grade) = %s")
        params.append(grade)

    where_sql = ' AND '.join(where)
    cur = _cur(master_conn)
    cur.execute(f"""
        SELECT COUNT(*) AS total
        FROM b2b_listings l
        WHERE {where_sql}
    """, params)
    total = cur.fetchone()['total'] or 0

    cur.execute(f"""
        SELECT {_LISTING_COLS_L}, t.name AS seller_name
        FROM b2b_listings l
        LEFT JOIN tenants t ON t.id = l.seller_tenant_id
        WHERE {where_sql}
        ORDER BY l.created_at DESC, l.id DESC
        LIMIT %s OFFSET %s
    """, params + [per_page, (page - 1) * per_page])
    rows = cur.fetchall()
    cur.close()

    items = [_listing_dict(r, seller_name=r.get('seller_name')) for r in rows]
    pagination = {
        'page': page, 'per_page': per_page, 'total': total,
        'total_pages': (total + per_page - 1) // per_page,
    }
    return items, pagination


# ─── Orders ────────────────────────────────────────────────────────────────

_ORDER_COLS = (
    'id, buyer_tenant_id, seller_tenant_id, buyer_name, seller_name, total, '
    'status, notes, created_at, updated_at'
)


def _order_items(master_conn, order_ids):
    """Bulk-load order lines grouped by order_id."""
    if not order_ids:
        return {}
    cur = _cur(master_conn)
    cur.execute("""
        SELECT id, order_id, listing_id, part_number, part_name, quantity,
               unit_price, subtotal
        FROM marketplace_order_items
        WHERE order_id = ANY(%s)
        ORDER BY order_id, id
    """, (list(order_ids),))
    rows = cur.fetchall()
    cur.close()
    out = {}
    for r in rows:
        item = _clean(dict(r))
        item['name'] = item.get('part_name')  # UI-friendly alias
        out.setdefault(r['order_id'], []).append(item)
    return out


def _order_dict(row, items, role):
    d = _clean(dict(row))
    d['total'] = float(d.get('total') or 0)
    d['role'] = role
    if role == 'buyer':
        d['other_party'] = d.get('seller_name')
        d['other_tenant_id'] = d.get('seller_tenant_id')
    else:
        d['other_party'] = d.get('buyer_name')
        d['other_tenant_id'] = d.get('buyer_tenant_id')
    d['items'] = items or []
    return d


def create_order(master_conn, buyer_tenant_id, items, notes=None):
    """Create a pending order + items and reserve the listings.

    All lines must reference available listings from ONE other tenant.
    """
    if not isinstance(items, list) or not items:
        raise B2BValidationError('items requerido (lista no vacía)')

    wanted = []
    for it in items:
        if not isinstance(it, dict):
            raise B2BValidationError('cada item debe ser un objeto {listing_id, quantity}')
        listing_id = _to_int(it.get('listing_id'))
        qty = _to_int(it.get('quantity'), 1)
        if not listing_id:
            raise B2BValidationError('listing_id requerido en cada item')
        if qty is None or qty < 1:
            raise B2BValidationError('quantity debe ser >= 1')
        wanted.append({'listing_id': listing_id, 'quantity': qty})

    listing_ids = [w['listing_id'] for w in wanted]
    cur = _cur(master_conn)
    cur.execute(
        f"SELECT {_LISTING_COLS} FROM b2b_listings WHERE id = ANY(%s) FOR UPDATE",
        (listing_ids,),
    )
    rows = {r['id']: dict(r) for r in cur.fetchall()}
    cur.close()

    sellers = set()
    resolved = []
    for w in wanted:
        listing = rows.get(w['listing_id'])
        if not listing:
            raise B2BNotFound(f"Listing {w['listing_id']} no encontrado")
        if listing['status'] != 'available':
            raise B2BValidationError(
                f"Listing {w['listing_id']} no está disponible (status: {listing['status']})")
        if listing['seller_tenant_id'] == buyer_tenant_id:
            raise B2BValidationError('No puedes comprar tus propias piezas')
        sellers.add(listing['seller_tenant_id'])
        resolved.append((listing, w['quantity']))

    if len(sellers) > 1:
        raise B2BValidationError(
            'Una orden solo puede tener un vendedor; crea una orden por vendedor')
    seller_tenant_id = sellers.pop()

    total = round(sum(
        _to_float(listing['price_wholesale'], 0) * qty
        for listing, qty in resolved), 2)
    buyer_name = _tenant_name(master_conn, buyer_tenant_id) or ''
    seller_name = _tenant_name(master_conn, seller_tenant_id) or ''

    cur = _cur(master_conn)
    cur.execute("""
        INSERT INTO marketplace_orders
            (buyer_tenant_id, seller_tenant_id, buyer_name, seller_name,
             total, status, notes)
        VALUES (%s, %s, %s, %s, %s, 'pending', %s)
        RETURNING id
    """, (buyer_tenant_id, seller_tenant_id, buyer_name, seller_name,
          total, (notes or '').strip() or None))
    order_id = cur.fetchone()['id']

    order_items = []
    for listing, qty in resolved:
        price = _to_float(listing['price_wholesale'], 0)
        subtotal = round(price * qty, 2)
        cur.execute("""
            INSERT INTO marketplace_order_items
                (order_id, listing_id, part_number, part_name, quantity, unit_price, subtotal)
            VALUES (%s, %s, %s, %s, %s, %s, %s)
        """, (order_id, listing['id'], listing.get('part_number'), listing.get('name'),
              qty, price, subtotal))
        order_items.append({
            'listing_id': listing['id'],
            'part_number': listing.get('part_number'),
            'part_name': listing.get('name'),
            'name': listing.get('name'),
            'quantity': qty,
            'unit_price': price,
            'subtotal': subtotal,
        })

    cur.execute("""
        UPDATE b2b_listings SET status = 'reserved', updated_at = NOW()
        WHERE id = ANY(%s)
    """, ([listing['id'] for listing, _ in resolved],))
    cur.close()

    return {
        'id': order_id,
        'total': total,
        'status': 'pending',
        'seller_tenant_id': seller_tenant_id,
        'buyer_tenant_id': buyer_tenant_id,
        'seller_name': seller_name,
        'buyer_name': buyer_name,
        'items': order_items,
    }


def list_orders(master_conn, tenant_id, role):
    """Orders where `tenant_id` is buyer ('buyer') or seller ('seller')."""
    col = 'buyer_tenant_id' if role == 'buyer' else 'seller_tenant_id'
    cur = _cur(master_conn)
    cur.execute(
        f"SELECT {_ORDER_COLS} FROM marketplace_orders WHERE {col} = %s "
        f"ORDER BY created_at DESC, id DESC",
        (tenant_id,),
    )
    rows = cur.fetchall()
    cur.close()
    items = _order_items(master_conn, [r['id'] for r in rows])
    return [_order_dict(r, items.get(r['id']), role) for r in rows]


def get_order(master_conn, tenant_id, order_id):
    """Order detail; only participants. Raises B2BNotFound / B2BForbidden."""
    cur = _cur(master_conn)
    cur.execute(f"SELECT {_ORDER_COLS} FROM marketplace_orders WHERE id = %s", (order_id,))
    row = cur.fetchone()
    cur.close()
    if not row:
        raise B2BNotFound('Orden no encontrada')
    if row['buyer_tenant_id'] == tenant_id:
        role = 'buyer'
    elif row['seller_tenant_id'] == tenant_id:
        role = 'seller'
    else:
        raise B2BForbidden('No participas en esta orden')
    items = _order_items(master_conn, [order_id])
    return _order_dict(row, items.get(order_id), role)


def update_order_status(master_conn, tenant_id, order_id, new_status, branch_id=None):
    """Apply an order status transition (role-checked).

    Commits the master transaction itself: when moving to `received` the status
    must be durable before the cross-tenant side effects (marking the seller's
    piece sold and creating the piece in the buyer's inventory) run.
    """
    new_status = (new_status or '').strip().lower()
    if new_status not in ORDER_STATUSES:
        raise B2BValidationError(
            f"status inválido: {new_status} (válidos: {', '.join(ORDER_STATUSES)})")

    cur = _cur(master_conn)
    cur.execute(
        f"SELECT {_ORDER_COLS} FROM marketplace_orders WHERE id = %s FOR UPDATE",
        (order_id,),
    )
    row = cur.fetchone()
    if not row:
        cur.close()
        raise B2BNotFound('Orden no encontrada')

    if row['buyer_tenant_id'] == tenant_id:
        role = 'buyer'
    elif row['seller_tenant_id'] == tenant_id:
        role = 'seller'
    else:
        cur.close()
        raise B2BForbidden('No participas en esta orden')

    current = row['status'] or 'pending'
    allowed = ORDER_TRANSITIONS.get(current, {}).get(new_status)
    if not allowed:
        cur.close()
        raise B2BValidationError(f"No se puede pasar de '{current}' a '{new_status}'")
    if role not in allowed:
        cur.close()
        raise B2BForbidden(
            f"Rol '{role}' no puede cambiar la orden de '{current}' a '{new_status}'")

    cur.execute(
        "UPDATE marketplace_orders SET status = %s, updated_at = NOW() WHERE id = %s",
        (new_status, order_id),
    )
    cur.close()
    master_conn.commit()

    result = get_order(master_conn, tenant_id, order_id)
    result['previous_status'] = current

    if new_status == 'received':
        extras = _finalize_received(master_conn, dict(row), branch_id=branch_id)
        result.update(extras)
    return result


# ─── Received: cross-tenant side effects (best-effort) ─────────────────────

def _resolve_branch(conn, branch_id=None):
    if branch_id:
        return branch_id
    try:
        cur = conn.cursor()
        cur.execute("SELECT id FROM branches ORDER BY id LIMIT 1")
        r = cur.fetchone()
        cur.close()
        return r[0] if r else None
    except Exception:
        return None


def _mark_seller_piece_sold(conn, inventory_id):
    """Mark the seller's inventory row as sold (SAVEPOINT, never raises).

    `inventory` has no `status` column in the standard tenant schema, so we set
    `is_active = false` (piece no longer available) when it is absent; if a
    tenant ever adds `inventory.status`, we write 'sold' there instead.
    """
    cur = conn.cursor()
    cur.execute("SAVEPOINT b2b_sold")
    try:
        cur.execute("""
            SELECT 1 FROM information_schema.columns
            WHERE table_name = 'inventory' AND column_name = 'status' LIMIT 1
        """)
        has_status = cur.fetchone() is not None
        if has_status:
            cur.execute("UPDATE inventory SET status = 'sold' WHERE id = %s", (inventory_id,))
        else:
            cur.execute("UPDATE inventory SET is_active = false WHERE id = %s", (inventory_id,))
        cur.execute("RELEASE SAVEPOINT b2b_sold")
        return True
    except Exception:
        try:
            cur.execute("ROLLBACK TO SAVEPOINT b2b_sold")
        except Exception:
            pass
        return False
    finally:
        cur.close()


def _insert_buyer_inventory(conn, item, listing, unit_price, warnings):
    """Insert the purchased piece into the buyer's inventory (SAVEPOINT)."""
    from services.yonke_service import generate_unique_sku

    name = (listing.get('name') or item.get('part_name') or item.get('part_number')
            or 'Pieza usada')
    candidate = (listing.get('part_number') or item.get('part_number') or '').strip() or None
    grade = listing.get('condition_grade')

    cur = conn.cursor()
    inv_id = None
    for _attempt in range(5):
        if candidate is None:
            try:
                candidate = generate_unique_sku(conn, None)
            except Exception:
                candidate = None
        cur.execute("SAVEPOINT b2b_inv")
        try:
            cur.execute("""
                INSERT INTO inventory
                    (part_number, name, unit, cost, price_1, price_2, price_3, tax_rate,
                     condition, condition_grade, is_unique_piece,
                     donor_make, donor_model, donor_year, ownership, image_url, is_active)
                VALUES (%s, %s, 'PZA', %s, %s, %s, %s, %s,
                        'usado', %s, true,
                        %s, %s, %s, 'propio', %s, true)
                RETURNING id
            """, (
                candidate, name, unit_price, unit_price, unit_price, unit_price, 0.16,
                grade, listing.get('donor_make'), listing.get('donor_model'),
                listing.get('donor_year'), listing.get('image_url'),
            ))
            inv_id = cur.fetchone()[0]
            cur.execute("RELEASE SAVEPOINT b2b_inv")
            break
        except pg_errors.UniqueViolation:
            cur.execute("ROLLBACK TO SAVEPOINT b2b_inv")
            candidate = None
        except Exception as e:
            try:
                cur.execute("ROLLBACK TO SAVEPOINT b2b_inv")
            except Exception:
                pass
            warnings.append(f"no se pudo crear la pieza en el inventario del comprador: {e}")
            cur.close()
            return None
    cur.close()
    return inv_id


def _seed_stock(conn, inv_id, branch_id, qty, cost, warnings):
    """Seed initial stock for a freshly created buyer piece (SAVEPOINT)."""
    if not branch_id or qty <= 0:
        return
    from services.inventory_engine import record_initial

    cur = conn.cursor()
    cur.execute("SAVEPOINT b2b_stock")
    try:
        record_initial(conn, inv_id, branch_id, qty, cost=cost)
        cur.execute("RELEASE SAVEPOINT b2b_stock")
    except Exception as e:
        try:
            cur.execute("ROLLBACK TO SAVEPOINT b2b_stock")
        except Exception:
            pass
        warnings.append(f"no se registró la operación de stock: {e}")
    # Idempotent safety net: the trigger normally creates this row already.
    try:
        cur.execute("""
            INSERT INTO inventory_stock (inventory_id, branch_id, stock)
            VALUES (%s, %s, %s)
            ON CONFLICT (inventory_id, branch_id) DO UPDATE SET stock = EXCLUDED.stock
        """, (inv_id, branch_id, qty))
    except Exception as e:
        warnings.append(f"no se pudo fijar el stock inicial: {e}")
    cur.close()


def _finalize_received(master_conn, order, branch_id=None):
    """On `received`: sell the listings/seller pieces and stock the buyer.

    Best-effort by design: every step runs in its own SAVEPOINT / connection,
    failures are collected in `warnings`, and the order is never aborted.
    """
    warnings = []
    extras = {'listings_sold': 0, 'inventory_created': 0, 'warnings': warnings}
    order_id = order['id']
    seller_id = order['seller_tenant_id']
    buyer_id = order['buyer_tenant_id']

    cur = _cur(master_conn)
    cur.execute("""
        SELECT id, listing_id, part_number, part_name, quantity, unit_price
        FROM marketplace_order_items WHERE order_id = %s ORDER BY id
    """, (order_id,))
    items = [dict(r) for r in cur.fetchall()]

    listing_ids = [it['listing_id'] for it in items if it['listing_id']]
    listings = {}
    if listing_ids:
        cur.execute(f"SELECT {_LISTING_COLS} FROM b2b_listings WHERE id = ANY(%s)",
                    (listing_ids,))
        for r in cur.fetchall():
            listings[r['id']] = dict(r)

    # Legacy rows may lack listing_id: fall back to (seller, part_number).
    for it in items:
        if it['listing_id'] in listings:
            continue
        cur.execute(f"""
            SELECT {_LISTING_COLS} FROM b2b_listings
            WHERE seller_tenant_id = %s AND part_number = %s
            ORDER BY (status = 'reserved') DESC, id DESC LIMIT 1
        """, (seller_id, it.get('part_number')))
        r = cur.fetchone()
        if r:
            listings[it['listing_id']] = dict(r)

    # 1) Mark listings sold (master DB) ------------------------------------
    sold = 0
    for it in items:
        listing = listings.get(it['listing_id']) if it['listing_id'] else None
        if not listing and it['listing_id']:
            continue
        try:
            if listing:
                cur.execute(
                    "UPDATE b2b_listings SET status = 'sold', updated_at = NOW() WHERE id = %s",
                    (listing['id'],))
            else:
                cur.execute("""
                    UPDATE b2b_listings SET status = 'sold', updated_at = NOW()
                    WHERE seller_tenant_id = %s AND part_number = %s AND status = 'reserved'
                """, (seller_id, it.get('part_number')))
            sold += cur.rowcount
        except Exception as e:
            master_conn.rollback()
            warnings.append(f"no se pudo marcar el listing como vendido: {e}")
            cur = _cur(master_conn)
    try:
        master_conn.commit()
    except Exception as e:
        warnings.append(f"no se pudo confirmar el estado de los listings: {e}")
    extras['listings_sold'] = sold
    cur.close()

    # 2) Mark the seller's pieces sold (seller tenant DB) ------------------
    seller_conn = None
    try:
        seller_conn = get_tenant_conn(seller_id)
        for it in items:
            listing = listings.get(it['listing_id'])
            inv_id = _to_int((listing or {}).get('seller_inventory_id'))
            if inv_id:
                _mark_seller_piece_sold(seller_conn, inv_id)
        seller_conn.commit()
    except Exception as e:
        warnings.append(f"no se pudieron marcar las piezas del vendedor: {e}")
        if seller_conn is not None:
            try:
                seller_conn.rollback()
            except Exception:
                pass
    finally:
        if seller_conn is not None:
            try:
                seller_conn.close()
            except Exception:
                pass

    # 3) Create the purchased pieces in the buyer's inventory --------------
    buyer_conn = None
    try:
        buyer_conn = get_tenant_conn(buyer_id)
        buyer_branch = _resolve_branch(buyer_conn, branch_id)
        created = 0
        for it in items:
            listing = listings.get(it['listing_id']) or {}
            qty = max(_to_int(it.get('quantity'), 1) or 1, 1)
            price = _to_float(it.get('unit_price'), 0)
            inv_id = _insert_buyer_inventory(buyer_conn, it, listing, price, warnings)
            if inv_id:
                _seed_stock(buyer_conn, inv_id, buyer_branch, qty, price, warnings)
                created += 1
        buyer_conn.commit()
        extras['inventory_created'] = created
        try:
            _audit_best_effort(buyer_conn, 'B2B_ORDER_RECEIVED', 'marketplace_order',
                               order_id, {'inventory_created': created})
            buyer_conn.commit()
        except Exception:
            pass
    except Exception as e:
        warnings.append(f"no se pudo crear el inventario del comprador: {e}")
        if buyer_conn is not None:
            try:
                buyer_conn.rollback()
            except Exception:
                pass
    finally:
        if buyer_conn is not None:
            try:
                buyer_conn.close()
            except Exception:
                pass

    return extras


# ─── Seller status (UI badge) ──────────────────────────────────────────────

def get_seller_status(master_conn, tenant_id):
    """{is_seller, listings_count, incoming_orders} for the current tenant."""
    seller = is_seller(master_conn, tenant_id)
    cur = _cur(master_conn)
    cur.execute("""
        SELECT COUNT(*) AS c FROM b2b_listings
        WHERE seller_tenant_id = %s AND status <> 'removed'
    """, (tenant_id,))
    listings_count = cur.fetchone()['c'] or 0

    cur.execute("""
        SELECT COUNT(*) AS c FROM marketplace_orders
        WHERE seller_tenant_id = %s AND status IN ('pending', 'confirmed', 'shipped')
    """, (tenant_id,))
    incoming = cur.fetchone()['c'] or 0

    cur.execute("""
        SELECT COUNT(*) AS c FROM marketplace_orders WHERE seller_tenant_id = %s
    """, (tenant_id,))
    total_incoming = cur.fetchone()['c'] or 0
    cur.close()

    return {
        'is_seller': seller,
        'listings_count': listings_count,
        'incoming_orders': incoming,
        'incoming_orders_total': total_incoming,
    }