feat(marketplace): urgent/featured leads for customers, correct request cap

New paid feature per product clarification: a customer can pay 50
Tracecoins to mark their own open requirement ("lead", professional-
facing) urgent. Urgent leads:
  - sort first in the professional-facing open-leads feed (new --
    there was previously no way for a professional to discover a
    requirement other than being handed its id directly, so this adds
    GET /marketplace alongside the existing GET /marketplace/{id})
  - get a raised per-lead request cap: 20 instead of the normal 10
    (was hardcoded to 20 for everyone -- corrected to the actual
    intended normal cap while adding the urgent tier)
  - trigger an immediate best-effort notification to approved
    professionals in the same profession + location

No separate expiry: urgent status rides the existing 7-day
expires_at, it doesn't extend anything -- it only changes ordering,
cap, and notification while the lead is already live.

TracecoinWalletRepository::try_debit_balance is a new one-shot debit
primitive (SELECT...FOR UPDATE, idempotent on reference_type +
reference_id) for spends that don't need reserve/confirm's two-step
hold -- mark_requirement_urgent debits before flagging the lead, so a
failed/insufficient-funds debit leaves no state to compensate, and a
retry (or two concurrent requests for the same lead) can't double-
charge.

apps/customers/src/handlers.rs

@@ -25,6 +25,7 @@ pub fn router() -> Router<AppState> {
         .route("/requirements", get(list_requirements).post(create_requirement))
         .route("/requirements/{id}", get(get_requirement).patch(update_requirement))
         .route("/requirements/{id}/submit", post(submit_requirement))
+        .route("/requirements/{id}/urgent", post(mark_requirement_urgent))
         .route("/requests", get(list_requests))
         .route("/requests/{lead_id}/approve", post(approve_request))
         .route("/requests/{lead_id}/reject", post(reject_request))
@@ -306,6 +307,119 @@ async fn submit_requirement(
     }
 }
 
+/// TraceCoins a customer pays to mark their own open requirement urgent.
+/// Urgent leads sort first in the professional-facing feed, get a raised
+/// request cap (see MAX_REQUESTS_PER_LEAD_URGENT in profession_shared.rs),
+/// and trigger an immediate notification to matching professionals. No
+/// separate expiry -- urgent status lasts exactly as long as the lead
+/// itself (the existing 7-day expires_at).
+const URGENT_UPGRADE_COST_TRACECOINS: i32 = 50;
+
+async fn mark_requirement_urgent(
+    State(state): State<AppState>,
+    Path(id): Path<Uuid>,
+    auth: AuthUser,
+) -> impl IntoResponse {
+    let req = match RequirementRepository::get_by_id(&state.pool, id).await {
+        Ok(Some(r)) => r,
+        Ok(None) => return (StatusCode::NOT_FOUND, "Requirement not found").into_response(),
+        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
+    };
+
+    if req.created_by_user_id != Some(auth.user_id) {
+        return (StatusCode::FORBIDDEN, "You do not own this requirement").into_response();
+    }
+
+    if req.status != "OPEN" {
+        return (StatusCode::BAD_REQUEST, "Only an open (approved) requirement can be marked urgent").into_response();
+    }
+
+    if req.is_urgent {
+        return (StatusCode::OK, Json(req)).into_response();
+    }
+
+    // Debit before flagging, not after: try_debit_balance is idempotent on
+    // (reference_type, reference_id), keyed to this lead's own id, so a
+    // retry -- or two concurrent requests for the same lead -- can never
+    // double-charge. Charging first also means a failed/insufficient-funds
+    // debit leaves nothing to compensate: the lead is simply never flagged.
+    let debited = match TracecoinWalletRepository::try_debit_balance(
+        &state.pool,
+        auth.user_id,
+        URGENT_UPGRADE_COST_TRACECOINS,
+        "LEAD_URGENT_UPGRADE",
+        req.id,
+    ).await {
+        Ok(true) => true,
+        Ok(false) => return (StatusCode::PAYMENT_REQUIRED, Json(serde_json::json!({
+            "error": "Insufficient Tracecoins to mark this requirement urgent",
+            "code": "INSUFFICIENT_FUNDS",
+            "cost": URGENT_UPGRADE_COST_TRACECOINS,
+        }))).into_response(),
+        Err(e) => {
+            tracing::error!("mark_requirement_urgent debit error: {}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response();
+        }
+    };
+    debug_assert!(debited);
+
+    let updated = match sqlx::query_as::<_, db::models::requirement::Requirement>(
+        r#"
+        UPDATE leads
+        SET is_urgent = true, urgent_at = NOW(), updated_at = NOW()
+        WHERE id = $1 AND is_urgent = false
+        RETURNING *
+        "#,
+    )
+    .bind(req.id)
+    .fetch_optional(&state.pool)
+    .await
+    {
+        Ok(Some(updated)) => updated,
+        // Already urgent (a concurrent request for the same lead won the
+        // race) -- the debit above was a no-op for it too, so nothing to
+        // undo; just report the current state.
+        Ok(None) => req,
+        Err(e) => {
+            tracing::error!("mark_requirement_urgent failed to flag lead {} after successful debit: {}", req.id, e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, "Payment succeeded but failed to mark the requirement urgent -- contact support").into_response();
+        }
+    };
+
+    // Best-effort instant notification to approved professionals in the
+    // same profession + location -- not wrapped in the transaction above,
+    // and never blocks the response.
+    let notify = sqlx::query_scalar::<_, Uuid>(
+        r#"
+        SELECT user_id FROM user_role_profiles
+        WHERE role_key = $1 AND approval_status = 'APPROVED' AND location ILIKE $2
+        LIMIT 500
+        "#,
+    )
+    .bind(&updated.profession_key)
+    .bind(format!("%{}%", updated.location))
+    .fetch_all(&state.pool)
+    .await
+    .unwrap_or_default();
+
+    for professional_user_id in notify {
+        let _ = sqlx::query(
+            r#"
+            INSERT INTO notifications (user_id, title, body, type, reference_id)
+            VALUES ($1, $2, $3, 'LEAD_URGENT', $4)
+            "#,
+        )
+        .bind(professional_user_id)
+        .bind("Urgent lead available")
+        .bind(format!("A new urgent {} lead was just posted in {}.", updated.profession_key, updated.location))
+        .bind(updated.id)
+        .execute(&state.pool)
+        .await;
+    }
+
+    (StatusCode::OK, Json(updated)).into_response()
+}
+
 async fn list_requests(
     State(state): State<AppState>,
     auth: AuthUser,

crates/contracts/src/profession_shared.rs

@@ -36,6 +36,21 @@ pub struct LeadRequestPayload {
     pub message: Option<String>,
 }
 
+#[derive(Deserialize)]
+pub struct OpenLeadsQuery {
+    pub location: Option<String>,
+    pub page: Option<i64>,
+    pub limit: Option<i64>,
+}
+
+/// Standard per-lead request cap -- how many professionals can send a
+/// request before a requirement stops accepting new ones.
+const MAX_REQUESTS_PER_LEAD: i32 = 10;
+/// Raised cap for a customer-paid urgent lead (see mark_requirement_urgent
+/// in apps/customers/src/handlers.rs) -- more competing requests is the
+/// point of paying to be urgent.
+const MAX_REQUESTS_PER_LEAD_URGENT: i32 = 20;
+
 /// Build the shared Router that every profession service merges into its own Router.
 /// `profession_key` must be a `'static str` matching the role key, e.g. `"PHOTOGRAPHER"`.
 pub fn shared_routes(profession_key: &'static str) -> Router<ProfessionState> {
@@ -81,6 +96,13 @@ pub fn shared_routes(profession_key: &'static str) -> Router<ProfessionState> {
             }),
         )
         .route("/marketplace/{id}", get(get_requirement))
+        .route(
+            "/marketplace",
+            get({
+                let pk = profession_key;
+                move |state, auth, query| list_open_leads(state, auth, query, pk)
+            }),
+        )
         // ── Lead Requests ────────────────────────────────────────────────────
         .route(
             "/leads/request",
@@ -130,6 +152,48 @@ async fn get_requirement(
     }
 }
 
+/// Browsable feed of open leads matching this profession -- there was
+/// previously no way for a professional to discover a requirement other
+/// than being handed its id directly. Urgent (customer-paid) leads sort
+/// first, then newest first; expired/closed/rejected leads never appear.
+async fn list_open_leads(
+    State(state): State<ProfessionState>,
+    _auth: AuthUser,
+    Query(query): Query<OpenLeadsQuery>,
+    profession_key: &'static str,
+) -> impl IntoResponse {
+    let limit = query.limit.unwrap_or(20).clamp(1, 100);
+    let page = query.page.unwrap_or(1).max(1);
+    let offset = (page - 1) * limit;
+    let location_filter = query.location.map(|l| format!("%{}%", l));
+
+    let rows = sqlx::query_as::<_, db::models::requirement::Requirement>(
+        r#"
+        SELECT * FROM leads
+        WHERE profession_key = $1
+          AND status = 'OPEN'
+          AND (expires_at IS NULL OR expires_at > NOW())
+          AND ($2::TEXT IS NULL OR location ILIKE $2)
+        ORDER BY is_urgent DESC, created_at DESC
+        LIMIT $3 OFFSET $4
+        "#,
+    )
+    .bind(profession_key)
+    .bind(&location_filter)
+    .bind(limit)
+    .bind(offset)
+    .fetch_all(&state.pool)
+    .await;
+
+    match rows {
+        Ok(rows) => (StatusCode::OK, Json(serde_json::json!({
+            "data": rows,
+            "pagination": { "page": page, "limit": limit }
+        }))).into_response(),
+        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
+    }
+}
+
 async fn send_lead_request(
     State(state): State<ProfessionState>,
     auth: AuthUser,
@@ -186,7 +250,10 @@ async fn send_lead_request(
         _           => return (StatusCode::NOT_FOUND, "Requirement not found").into_response(),
     };
 
-    if req.request_count >= 20 {
+    // Urgent (customer-paid) leads get a raised request cap -- more
+    // professionals competing for it is the point of paying to be urgent.
+    let request_cap = if req.is_urgent { MAX_REQUESTS_PER_LEAD_URGENT } else { MAX_REQUESTS_PER_LEAD };
+    if req.request_count >= request_cap {
         return (StatusCode::CONFLICT, "Requirement reached max requests").into_response();
     }
 

crates/db/migrations/20260818000003_leads_urgent_flag.down.sql

@@ -0,0 +1,9 @@
+BEGIN;
+
+DROP INDEX IF EXISTS idx_leads_open_feed;
+
+ALTER TABLE leads
+    DROP COLUMN IF EXISTS is_urgent,
+    DROP COLUMN IF EXISTS urgent_at;
+
+COMMIT;

crates/db/migrations/20260818000003_leads_urgent_flag.up.sql

@@ -0,0 +1,21 @@
+-- New paid feature: a customer can pay 50 TraceCoins to mark their own
+-- open requirement ("lead", in professional-facing vocabulary) as urgent.
+-- Urgent leads: sort first in the professional-facing feed, get a raised
+-- request cap (20 vs the normal 10), and trigger an immediate notification
+-- to matching professionals -- see apps/customers/src/handlers.rs
+-- (mark_requirement_urgent) and crates/contracts/src/profession_shared.rs
+-- (the new open-leads feed + send_lead_request's cap check).
+--
+-- No separate expiry: urgent status lasts exactly as long as the lead
+-- itself (the existing 7-day expires_at) -- it doesn't extend anything,
+-- it just changes how the lead is surfaced and capped while it's live.
+BEGIN;
+
+ALTER TABLE leads
+    ADD COLUMN IF NOT EXISTS is_urgent BOOLEAN NOT NULL DEFAULT false,
+    ADD COLUMN IF NOT EXISTS urgent_at TIMESTAMPTZ;
+
+CREATE INDEX IF NOT EXISTS idx_leads_open_feed
+    ON leads (profession_key, status, is_urgent DESC, created_at DESC);
+
+COMMIT;

crates/db/src/models/requirement.rs

@@ -17,6 +17,8 @@ pub struct Requirement {
     pub rejection_reason: Option<String>,
     pub request_count: i32,
     pub accepted_count: i32,
+    pub is_urgent: bool,
+    pub urgent_at: Option<DateTime<Utc>>,
     pub expires_at: Option<DateTime<Utc>>,
     pub approved_at: Option<DateTime<Utc>>,
     pub approved_by: Option<Uuid>,

crates/db/src/models/tracecoin_wallet.rs

@@ -110,6 +110,84 @@ impl TracecoinWalletRepository {
         Ok(true)
     }
 
+    /// Debits `amount` directly from `balance` (not via reserve/confirm) --
+    /// for one-shot spends like the urgent-lead upgrade, where there's no
+    /// multi-step hold to manage: check funds, spend, done. Idempotent on
+    /// `reference_id` + `reference_type`: a retry with the same reference
+    /// finds the existing ledger row and returns `true` without debiting
+    /// twice, mirroring the FOR-UPDATE-gated pattern the reserve/debit/
+    /// release trio above already uses.
+    pub async fn try_debit_balance(
+        pool: &PgPool,
+        user_id: Uuid,
+        amount: i32,
+        reference_type: &str,
+        reference_id: Uuid,
+    ) -> Result<bool, sqlx::Error> {
+        let mut tx = pool.begin().await?;
+
+        sqlx::query(
+            r#"
+            INSERT INTO tracecoin_wallets (user_id, balance, reserved)
+            VALUES ($1, 0, 0)
+            ON CONFLICT (user_id) DO NOTHING
+            "#,
+        )
+        .bind(user_id)
+        .execute(&mut *tx)
+        .await?;
+
+        let wallet = sqlx::query_as::<_, Wallet>(
+            "SELECT * FROM tracecoin_wallets WHERE user_id = $1 FOR UPDATE",
+        )
+        .bind(user_id)
+        .fetch_one(&mut *tx)
+        .await?;
+
+        let already: Option<Uuid> = sqlx::query_scalar(
+            "SELECT id FROM tracecoin_ledger WHERE wallet_id = $1 AND reference_id = $2 AND reference_type = $3",
+        )
+        .bind(wallet.id)
+        .bind(reference_id)
+        .bind(reference_type)
+        .fetch_optional(&mut *tx)
+        .await?;
+
+        if already.is_some() {
+            tx.commit().await?;
+            return Ok(true);
+        }
+
+        if wallet.balance < amount {
+            tx.rollback().await?;
+            return Ok(false);
+        }
+
+        sqlx::query(
+            "UPDATE tracecoin_wallets SET balance = balance - $1, updated_at = NOW() WHERE id = $2",
+        )
+        .bind(amount)
+        .bind(wallet.id)
+        .execute(&mut *tx)
+        .await?;
+
+        sqlx::query(
+            r#"
+            INSERT INTO tracecoin_ledger (wallet_id, transaction_type, amount, reference_type, reference_id)
+            VALUES ($1, 'DEBIT', $2, $3, $4)
+            "#,
+        )
+        .bind(wallet.id)
+        .bind(amount)
+        .bind(reference_type)
+        .bind(reference_id)
+        .execute(&mut *tx)
+        .await?;
+
+        tx.commit().await?;
+        Ok(true)
+    }
+
     pub async fn try_debit_reserved_tracecoins(
         pool: &PgPool,
         user_id: Uuid,