WIP — merchant profile foundation (multi-provider payment model, part 1)

Lays the schema + types + resolution layer for the merchant-profile-aware
multi-provider model documented in plans/multi-provider-payment-model.md.
Does NOT yet migrate any existing call site — legacy `state.payment_provider()`
and the singleton config tables continue to work via deprecation shims so
the daemon keeps running unchanged on this checkpoint.

This commit is intentionally a WIP foundation, not a shippable release —
no version bump, no release notes, no admin UI, no call-site migration.
A follow-up cycle ports purchase / subscriptions / reconcile / upgrade /
tipping to the new resolution layer, rebuilds the BTCPay + Zaprite connect
flows around merchant_profile_id, refactors webhook URLs to
/v1/{kind}/webhook/{provider_id}, ships the Merchant Profiles admin UI
section, wires the tier-cap, and bumps to :52 with the one-way migration
release notes.

What landed:

migrations/0020_merchant_profiles.sql
  Full schema + data port + DROP of the singleton tables. Creates
  merchant_profiles, payment_providers (FK to profile, unique per
  (profile, kind)), merchant_profile_rail_preferences (tie-breaker
  when a profile has 2 providers serving the same rail). Adds
  merchant_profile_id to products + (merchant_profile_id, payment_provider_id)
  to subscriptions for the snapshot-on-create semantics. Ports
  btcpay_config + zaprite_config + active_payment_provider setting
  into the new tables, then drops them. Master operator post-migration
  step: update the Zaprite webhook URL on the Zaprite dashboard to
  the new /v1/zaprite/webhook/{provider-id} form (or click Reconnect
  Zaprite in the new UI once it ships).

src/merchant_profiles.rs (new module)
  MerchantProfile struct + NewMerchantProfile + MerchantProfileUpdate
  input types. Business-logic CRUD helpers: create, get, get_default,
  require_default, list, update, set_default, delete, for_product.
  Delete refuses if products or active subs are attached or if it's
  the default profile. Tier-cap check stubbed with a TODO for the
  next chunk's tier.rs wire-up.

src/db/repo.rs (+469 lines)
  Repo helpers: create/get_by_id/get_default/get_for_product/list/
  update/set_default/delete for merchant_profiles + count helpers
  for products/active_subscriptions per profile. PaymentProviderRow
  struct + create/get/list_for_profile/list_all/delete. RailPreference
  struct + list/set/clear helpers. update_merchant_profile builds a
  dynamic SET clause so partial updates don't clobber fields the
  caller didn't touch.

src/payment/mod.rs
  Rail enum (Lightning / Onchain / Card) + ProviderKind::parse +
  rails_for_kind static mapping. build_provider(row, public_base) ->
  Arc<dyn PaymentProvider> factory that dispatches on kind to construct
  a typed BtcpayProvider or ZapriteProvider from a payment_providers
  row. PaymentProvider trait gains a default served_rails() impl
  returning rails_for_kind(self.kind()).

  Deprecation shims: SETTING_ACTIVE_PROVIDER constant +
  read_active_provider_preference + write_active_provider_preference
  stay callable so btcpay_authorize/zaprite_authorize/main.rs/the
  thank-you page still build. read_active_provider_preference now
  reads from the new payment_providers table (returns the kind of
  the first provider attached to the default profile), falling back
  to the legacy settings-table read pre-migration. write_* is a no-op.
  Each shim has a #[deprecated] attribute so the build surfaces
  exactly which call sites still need porting (lit up in the
  follow-up cycle's TODO).

src/api/mod.rs (AppState)
  New methods alongside the existing payment_provider() shim:
    - payment_provider_by_id(id) — looks up a row, builds the provider
    - merchant_profile_for_product(product_id) — resolves via products.merchant_profile_id, falls back to default
    - resolve_provider_for_profile_rail(profile_id, rail) —
      preference table -> single candidate -> deterministic earliest-
      connected with WARN. Returns (row, Arc<dyn PaymentProvider>).
    - resolve_provider_for_product_rail(product_id, rail) — convenience
      wrapping the previous two.

src/lib.rs
  Registers the new merchant_profiles module.

Build state: cargo check passes. Only warnings are the pre-existing
unused-import in recover.rs and the deprecation lint firing on the
five legacy call sites enumerated in the WIP plan.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

licensing-service/src/api/mod.rs

@@ -174,6 +174,143 @@ impl AppState {
         let mut guard = self.payment.write().await;
         *guard = None;
     }
+
+    // ---------------------------------------------------------------------
+    // Merchant-profile-aware resolution layer (migration 0020+)
+    // ---------------------------------------------------------------------
+    //
+    // The legacy `payment_provider()` / `set_payment_provider()` accessors
+    // above continue to work as a "default provider for the default
+    // profile" compatibility shim during the multi-provider transition.
+    // New call sites should use one of the methods below instead.
+
+    /// Look up a payment provider by its row id. Reads the row from the
+    /// DB, instantiates a typed `PaymentProvider` impl via
+    /// `payment::build_provider`. Not cached today — the caller is
+    /// usually invoking it once per request lifecycle so the cost is
+    /// nil. Add a cache here when profiling says we need one.
+    pub async fn payment_provider_by_id(
+        &self,
+        provider_id: &str,
+    ) -> AppResult<Arc<dyn crate::payment::PaymentProvider>> {
+        let row = crate::db::repo::get_payment_provider_by_id(&self.db, provider_id)
+            .await?
+            .ok_or_else(|| {
+                AppError::NotFound(format!("payment provider {provider_id}"))
+            })?;
+        crate::payment::build_provider(&row, self.config.btcpay_public_url.as_deref())
+            .map_err(AppError::Internal)
+    }
+
+    /// Resolve the merchant profile a product belongs to. Falls back to
+    /// the default profile if the product has no `merchant_profile_id`
+    /// set (defensive — shouldn't happen post-migration, but handles
+    /// any rows that slip through).
+    pub async fn merchant_profile_for_product(
+        &self,
+        product_id: &str,
+    ) -> AppResult<crate::merchant_profiles::MerchantProfile> {
+        crate::merchant_profiles::for_product(self, product_id).await
+    }
+
+    /// Pick the provider on `profile_id` that serves the given `rail`.
+    /// Resolution order:
+    ///   1. Honor `merchant_profile_rail_preferences` if the operator
+    ///      set an explicit preference for this (profile, rail) pair.
+    ///   2. If exactly one attached provider serves the rail, use it.
+    ///   3. If multiple serve the rail and no preference is set, use
+    ///      the earliest-`connected_at` one (deterministic) and log a
+    ///      warning so the operator knows to set an explicit preference.
+    ///   4. If no attached provider serves the rail, return
+    ///      `AppError::BadRequest` — caller should treat this as
+    ///      "buyer's pick isn't available for this merchant."
+    pub async fn resolve_provider_for_profile_rail(
+        &self,
+        profile_id: &str,
+        rail: crate::payment::Rail,
+    ) -> AppResult<(crate::db::repo::PaymentProviderRow, Arc<dyn crate::payment::PaymentProvider>)> {
+        // 1. Check the explicit preference table first.
+        let preferences = crate::db::repo::list_rail_preferences_for_profile(&self.db, profile_id).await?;
+        if let Some(pref) = preferences.iter().find(|p| p.rail == rail.as_str()) {
+            let row = crate::db::repo::get_payment_provider_by_id(&self.db, &pref.payment_provider_id)
+                .await?
+                .ok_or_else(|| {
+                    AppError::Internal(anyhow::anyhow!(
+                        "rail preference points at missing provider {}",
+                        pref.payment_provider_id
+                    ))
+                })?;
+            let provider =
+                crate::payment::build_provider(&row, self.config.btcpay_public_url.as_deref())
+                    .map_err(AppError::Internal)?;
+            return Ok((row, provider));
+        }
+
+        // 2. + 3. No explicit preference — find providers on this profile
+        //    whose kind serves the requested rail.
+        let providers = crate::db::repo::list_payment_providers_for_profile(&self.db, profile_id).await?;
+        let mut candidates: Vec<&crate::db::repo::PaymentProviderRow> = providers
+            .iter()
+            .filter(|row| {
+                crate::payment::ProviderKind::parse(&row.kind)
+                    .map(|kind| crate::payment::rails_for_kind(kind).contains(&rail))
+                    .unwrap_or(false)
+            })
+            .collect();
+        // Earliest-connected-first is already the order from list_payment_providers_for_profile
+        // (ORDER BY connected_at ASC), but be explicit for clarity.
+        candidates.sort_by(|a, b| a.connected_at.cmp(&b.connected_at));
+
+        match candidates.as_slice() {
+            [] => Err(AppError::BadRequest(format!(
+                "merchant profile {profile_id} has no provider that serves the '{}' rail. \
+                 Connect one in the admin UI's Merchant Profiles page, or set a rail \
+                 preference if multiple providers serve this rail.",
+                rail.as_str()
+            ))),
+            [only] => {
+                let row = (*only).clone();
+                let provider =
+                    crate::payment::build_provider(&row, self.config.btcpay_public_url.as_deref())
+                        .map_err(AppError::Internal)?;
+                Ok((row, provider))
+            }
+            [first, ..] => {
+                tracing::warn!(
+                    profile_id = %profile_id,
+                    rail = rail.as_str(),
+                    chosen = %first.id,
+                    candidate_count = candidates.len(),
+                    "multiple providers serve this rail on the profile; using earliest-connected \
+                     deterministically. Set an explicit rail preference in the admin UI to silence \
+                     this warning."
+                );
+                let row = (*first).clone();
+                let provider =
+                    crate::payment::build_provider(&row, self.config.btcpay_public_url.as_deref())
+                        .map_err(AppError::Internal)?;
+                Ok((row, provider))
+            }
+        }
+    }
+
+    /// Convenience for the most common purchase-flow case: given a
+    /// product id and a buyer-picked rail, resolve to (profile, provider
+    /// row, provider impl). Used by `purchase.rs` and `subscriptions.rs`
+    /// renewals.
+    pub async fn resolve_provider_for_product_rail(
+        &self,
+        product_id: &str,
+        rail: crate::payment::Rail,
+    ) -> AppResult<(
+        crate::merchant_profiles::MerchantProfile,
+        crate::db::repo::PaymentProviderRow,
+        Arc<dyn crate::payment::PaymentProvider>,
+    )> {
+        let profile = self.merchant_profile_for_product(product_id).await?;
+        let (row, provider) = self.resolve_provider_for_profile_rail(&profile.id, rail).await?;
+        Ok((profile, row, provider))
+    }
 }
 
 impl FromRef<AppState> for SqlitePool {

licensing-service/src/db/repo.rs

@@ -2891,3 +2891,472 @@ pub async fn settings_set(pool: &SqlitePool, key: &str, value: Option<&str>) ->
     .await?;
     Ok(())
 }
+
+// =========================================================================
+// Merchant profiles (migration 0020)
+// =========================================================================
+
+const MERCHANT_PROFILE_COLS: &str =
+    "id, name, legal_name, support_url, support_email, brand_color, \
+     post_purchase_redirect_url, is_default, \
+     smtp_host, smtp_port, smtp_username, smtp_password, \
+     smtp_from_address, smtp_from_name, smtp_use_starttls, \
+     created_at, updated_at";
+
+fn row_to_merchant_profile(
+    row: sqlx::sqlite::SqliteRow,
+) -> crate::merchant_profiles::MerchantProfile {
+    use sqlx::Row;
+    crate::merchant_profiles::MerchantProfile {
+        id: row.get("id"),
+        name: row.get("name"),
+        legal_name: row.try_get("legal_name").ok(),
+        support_url: row.try_get("support_url").ok(),
+        support_email: row.try_get("support_email").ok(),
+        brand_color: row.try_get("brand_color").ok(),
+        post_purchase_redirect_url: row.try_get("post_purchase_redirect_url").ok(),
+        is_default: row.get::<i64, _>("is_default") != 0,
+        smtp_host: row.try_get("smtp_host").ok(),
+        smtp_port: row.try_get("smtp_port").ok(),
+        smtp_username: row.try_get("smtp_username").ok(),
+        smtp_password: row.try_get("smtp_password").ok(),
+        smtp_from_address: row.try_get("smtp_from_address").ok(),
+        smtp_from_name: row.try_get("smtp_from_name").ok(),
+        smtp_use_starttls: row.get::<i64, _>("smtp_use_starttls") != 0,
+        created_at: row.get("created_at"),
+        updated_at: row.get("updated_at"),
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+pub async fn create_merchant_profile(
+    pool: &SqlitePool,
+    id: &str,
+    name: &str,
+    legal_name: Option<&str>,
+    support_url: Option<&str>,
+    support_email: Option<&str>,
+    brand_color: Option<&str>,
+    post_purchase_redirect_url: Option<&str>,
+    is_default: bool,
+    now: &str,
+) -> AppResult<()> {
+    sqlx::query(
+        "INSERT INTO merchant_profiles(\
+            id, name, legal_name, support_url, support_email, brand_color, \
+            post_purchase_redirect_url, is_default, \
+            smtp_use_starttls, created_at, updated_at) \
+         VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?)",
+    )
+    .bind(id)
+    .bind(name)
+    .bind(legal_name)
+    .bind(support_url)
+    .bind(support_email)
+    .bind(brand_color)
+    .bind(post_purchase_redirect_url)
+    .bind(is_default as i64)
+    .bind(now)
+    .bind(now)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+pub async fn get_merchant_profile_by_id(
+    pool: &SqlitePool,
+    id: &str,
+) -> AppResult<Option<crate::merchant_profiles::MerchantProfile>> {
+    let row = sqlx::query(&format!(
+        "SELECT {MERCHANT_PROFILE_COLS} FROM merchant_profiles WHERE id = ?"
+    ))
+    .bind(id)
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.map(row_to_merchant_profile))
+}
+
+pub async fn get_default_merchant_profile(
+    pool: &SqlitePool,
+) -> AppResult<Option<crate::merchant_profiles::MerchantProfile>> {
+    let row = sqlx::query(&format!(
+        "SELECT {MERCHANT_PROFILE_COLS} FROM merchant_profiles WHERE is_default = 1 LIMIT 1"
+    ))
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.map(row_to_merchant_profile))
+}
+
+pub async fn get_merchant_profile_for_product(
+    pool: &SqlitePool,
+    product_id: &str,
+) -> AppResult<Option<crate::merchant_profiles::MerchantProfile>> {
+    let row = sqlx::query(&format!(
+        "SELECT {MERCHANT_PROFILE_COLS} FROM merchant_profiles mp \
+         JOIN products p ON p.merchant_profile_id = mp.id \
+         WHERE p.id = ? LIMIT 1"
+    ))
+    .bind(product_id)
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.map(row_to_merchant_profile))
+}
+
+pub async fn list_merchant_profiles(
+    pool: &SqlitePool,
+) -> AppResult<Vec<crate::merchant_profiles::MerchantProfile>> {
+    let rows = sqlx::query(&format!(
+        "SELECT {MERCHANT_PROFILE_COLS} FROM merchant_profiles \
+         ORDER BY is_default DESC, created_at DESC"
+    ))
+    .fetch_all(pool)
+    .await?;
+    Ok(rows.into_iter().map(row_to_merchant_profile).collect())
+}
+
+pub async fn update_merchant_profile(
+    pool: &SqlitePool,
+    id: &str,
+    patch: &crate::merchant_profiles::MerchantProfileUpdate,
+) -> AppResult<()> {
+    use crate::merchant_profiles::MerchantProfileUpdate;
+    let MerchantProfileUpdate {
+        name,
+        legal_name,
+        support_url,
+        support_email,
+        brand_color,
+        post_purchase_redirect_url,
+        smtp_host,
+        smtp_port,
+        smtp_username,
+        smtp_password,
+        smtp_from_address,
+        smtp_from_name,
+        smtp_use_starttls,
+    } = patch;
+
+    // Build the SET clause dynamically — only update fields the caller
+    // explicitly set. Outer Option means "skip if None"; inner Option
+    // (on nullable fields) means "set to NULL if Some(None), set to a
+    // value if Some(Some(value))."
+    let mut sets: Vec<&'static str> = Vec::new();
+    if name.is_some() { sets.push("name = ?"); }
+    if legal_name.is_some() { sets.push("legal_name = ?"); }
+    if support_url.is_some() { sets.push("support_url = ?"); }
+    if support_email.is_some() { sets.push("support_email = ?"); }
+    if brand_color.is_some() { sets.push("brand_color = ?"); }
+    if post_purchase_redirect_url.is_some() { sets.push("post_purchase_redirect_url = ?"); }
+    if smtp_host.is_some() { sets.push("smtp_host = ?"); }
+    if smtp_port.is_some() { sets.push("smtp_port = ?"); }
+    if smtp_username.is_some() { sets.push("smtp_username = ?"); }
+    if smtp_password.is_some() { sets.push("smtp_password = ?"); }
+    if smtp_from_address.is_some() { sets.push("smtp_from_address = ?"); }
+    if smtp_from_name.is_some() { sets.push("smtp_from_name = ?"); }
+    if smtp_use_starttls.is_some() { sets.push("smtp_use_starttls = ?"); }
+
+    if sets.is_empty() {
+        return Ok(()); // nothing to update
+    }
+    sets.push("updated_at = ?");
+
+    let sql = format!(
+        "UPDATE merchant_profiles SET {} WHERE id = ?",
+        sets.join(", ")
+    );
+    let mut q = sqlx::query(&sql);
+    if let Some(v) = name { q = q.bind(v); }
+    if let Some(v) = legal_name { q = q.bind(v.as_deref()); }
+    if let Some(v) = support_url { q = q.bind(v.as_deref()); }
+    if let Some(v) = support_email { q = q.bind(v.as_deref()); }
+    if let Some(v) = brand_color { q = q.bind(v.as_deref()); }
+    if let Some(v) = post_purchase_redirect_url { q = q.bind(v.as_deref()); }
+    if let Some(v) = smtp_host { q = q.bind(v.as_deref()); }
+    if let Some(v) = smtp_port { q = q.bind(*v); }
+    if let Some(v) = smtp_username { q = q.bind(v.as_deref()); }
+    if let Some(v) = smtp_password { q = q.bind(v.as_deref()); }
+    if let Some(v) = smtp_from_address { q = q.bind(v.as_deref()); }
+    if let Some(v) = smtp_from_name { q = q.bind(v.as_deref()); }
+    if let Some(v) = smtp_use_starttls { q = q.bind(*v as i64); }
+    let now = Utc::now().to_rfc3339();
+    q = q.bind(&now).bind(id);
+    q.execute(pool).await?;
+    Ok(())
+}
+
+/// Flip a profile to be the default. Two-step UPDATE in a single
+/// transaction to maintain the partial unique index on is_default = 1.
+pub async fn set_default_merchant_profile(
+    pool: &SqlitePool,
+    new_default_id: &str,
+) -> AppResult<()> {
+    let now = Utc::now().to_rfc3339();
+    let mut tx = pool.begin().await?;
+    sqlx::query("UPDATE merchant_profiles SET is_default = 0, updated_at = ? WHERE is_default = 1")
+        .bind(&now)
+        .execute(&mut *tx)
+        .await?;
+    let rows = sqlx::query("UPDATE merchant_profiles SET is_default = 1, updated_at = ? WHERE id = ?")
+        .bind(&now)
+        .bind(new_default_id)
+        .execute(&mut *tx)
+        .await?
+        .rows_affected();
+    if rows == 0 {
+        return Err(AppError::NotFound(format!("merchant profile {new_default_id}")));
+    }
+    tx.commit().await?;
+    Ok(())
+}
+
+pub async fn delete_merchant_profile(pool: &SqlitePool, id: &str) -> AppResult<()> {
+    // Also cascade the rail_preferences entries (no ON DELETE CASCADE
+    // on that table since it's a composite primary key; cleaner to
+    // delete explicitly).
+    let mut tx = pool.begin().await?;
+    sqlx::query("DELETE FROM merchant_profile_rail_preferences WHERE merchant_profile_id = ?")
+        .bind(id)
+        .execute(&mut *tx)
+        .await?;
+    let rows = sqlx::query("DELETE FROM merchant_profiles WHERE id = ? AND is_default = 0")
+        .bind(id)
+        .execute(&mut *tx)
+        .await?
+        .rows_affected();
+    if rows == 0 {
+        return Err(AppError::BadRequest(format!(
+            "merchant profile {id} not found or is the default"
+        )));
+    }
+    tx.commit().await?;
+    Ok(())
+}
+
+pub async fn count_products_for_profile(pool: &SqlitePool, profile_id: &str) -> anyhow::Result<i64> {
+    let n: i64 = sqlx::query_scalar(
+        "SELECT COUNT(*) FROM products WHERE merchant_profile_id = ?",
+    )
+    .bind(profile_id)
+    .fetch_one(pool)
+    .await?;
+    Ok(n)
+}
+
+pub async fn count_active_subscriptions_for_profile(
+    pool: &SqlitePool,
+    profile_id: &str,
+) -> anyhow::Result<i64> {
+    let n: i64 = sqlx::query_scalar(
+        "SELECT COUNT(*) FROM subscriptions \
+         WHERE merchant_profile_id = ? AND status IN ('active', 'past_due')",
+    )
+    .bind(profile_id)
+    .fetch_one(pool)
+    .await?;
+    Ok(n)
+}
+
+// =========================================================================
+// Payment providers (migration 0020) — replaces btcpay_config + zaprite_config
+// =========================================================================
+
+/// Stored shape of a payment_providers row. Used by the provider factory
+/// in `payment::build_provider` to reconstruct a typed PaymentProvider
+/// trait object from a row.
+#[derive(Debug, Clone)]
+pub struct PaymentProviderRow {
+    pub id: String,
+    pub merchant_profile_id: String,
+    pub kind: String,
+    pub label: String,
+    pub api_key: String,
+    pub base_url: String,
+    pub webhook_id: Option<String>,
+    pub webhook_secret: Option<String>,
+    pub store_id: Option<String>,
+    pub connected_at: String,
+    pub updated_at: String,
+}
+
+const PAYMENT_PROVIDER_COLS: &str =
+    "id, merchant_profile_id, kind, label, api_key, base_url, \
+     webhook_id, webhook_secret, store_id, connected_at, updated_at";
+
+fn row_to_payment_provider(row: sqlx::sqlite::SqliteRow) -> PaymentProviderRow {
+    use sqlx::Row;
+    PaymentProviderRow {
+        id: row.get("id"),
+        merchant_profile_id: row.get("merchant_profile_id"),
+        kind: row.get("kind"),
+        label: row.get("label"),
+        api_key: row.get("api_key"),
+        base_url: row.get("base_url"),
+        webhook_id: row.try_get("webhook_id").ok(),
+        webhook_secret: row.try_get("webhook_secret").ok(),
+        store_id: row.try_get("store_id").ok(),
+        connected_at: row.get("connected_at"),
+        updated_at: row.get("updated_at"),
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+pub async fn create_payment_provider(
+    pool: &SqlitePool,
+    id: &str,
+    merchant_profile_id: &str,
+    kind: &str,
+    label: &str,
+    api_key: &str,
+    base_url: &str,
+    webhook_id: Option<&str>,
+    webhook_secret: Option<&str>,
+    store_id: Option<&str>,
+    now: &str,
+) -> AppResult<()> {
+    sqlx::query(
+        "INSERT INTO payment_providers(\
+            id, merchant_profile_id, kind, label, api_key, base_url, \
+            webhook_id, webhook_secret, store_id, connected_at, updated_at) \
+         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+    )
+    .bind(id)
+    .bind(merchant_profile_id)
+    .bind(kind)
+    .bind(label)
+    .bind(api_key)
+    .bind(base_url)
+    .bind(webhook_id)
+    .bind(webhook_secret)
+    .bind(store_id)
+    .bind(now)
+    .bind(now)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+pub async fn get_payment_provider_by_id(
+    pool: &SqlitePool,
+    id: &str,
+) -> AppResult<Option<PaymentProviderRow>> {
+    let row = sqlx::query(&format!(
+        "SELECT {PAYMENT_PROVIDER_COLS} FROM payment_providers WHERE id = ?"
+    ))
+    .bind(id)
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.map(row_to_payment_provider))
+}
+
+pub async fn list_payment_providers_for_profile(
+    pool: &SqlitePool,
+    profile_id: &str,
+) -> AppResult<Vec<PaymentProviderRow>> {
+    let rows = sqlx::query(&format!(
+        "SELECT {PAYMENT_PROVIDER_COLS} FROM payment_providers \
+         WHERE merchant_profile_id = ? ORDER BY connected_at ASC"
+    ))
+    .bind(profile_id)
+    .fetch_all(pool)
+    .await?;
+    Ok(rows.into_iter().map(row_to_payment_provider).collect())
+}
+
+pub async fn list_all_payment_providers(pool: &SqlitePool) -> AppResult<Vec<PaymentProviderRow>> {
+    let rows = sqlx::query(&format!(
+        "SELECT {PAYMENT_PROVIDER_COLS} FROM payment_providers ORDER BY connected_at ASC"
+    ))
+    .fetch_all(pool)
+    .await?;
+    Ok(rows.into_iter().map(row_to_payment_provider).collect())
+}
+
+pub async fn delete_payment_provider(pool: &SqlitePool, id: &str) -> AppResult<()> {
+    let mut tx = pool.begin().await?;
+    // Cascade rail preferences pointing at this provider.
+    sqlx::query("DELETE FROM merchant_profile_rail_preferences WHERE payment_provider_id = ?")
+        .bind(id)
+        .execute(&mut *tx)
+        .await?;
+    let rows = sqlx::query("DELETE FROM payment_providers WHERE id = ?")
+        .bind(id)
+        .execute(&mut *tx)
+        .await?
+        .rows_affected();
+    if rows == 0 {
+        return Err(AppError::NotFound(format!("payment provider {id}")));
+    }
+    tx.commit().await?;
+    Ok(())
+}
+
+// =========================================================================
+// Merchant profile rail preferences
+// =========================================================================
+
+/// (rail, provider_id) tuple representing one preference row.
+#[derive(Debug, Clone)]
+pub struct RailPreference {
+    pub rail: String,
+    pub payment_provider_id: String,
+}
+
+pub async fn list_rail_preferences_for_profile(
+    pool: &SqlitePool,
+    profile_id: &str,
+) -> AppResult<Vec<RailPreference>> {
+    use sqlx::Row;
+    let rows = sqlx::query(
+        "SELECT rail, payment_provider_id FROM merchant_profile_rail_preferences \
+         WHERE merchant_profile_id = ?",
+    )
+    .bind(profile_id)
+    .fetch_all(pool)
+    .await?;
+    Ok(rows
+        .into_iter()
+        .map(|r| RailPreference {
+            rail: r.get("rail"),
+            payment_provider_id: r.get("payment_provider_id"),
+        })
+        .collect())
+}
+
+/// Upsert a (profile, rail) → provider mapping. Replaces any existing
+/// preference for the same (profile, rail) pair.
+pub async fn set_rail_preference(
+    pool: &SqlitePool,
+    profile_id: &str,
+    rail: &str,
+    provider_id: &str,
+) -> AppResult<()> {
+    sqlx::query(
+        "INSERT INTO merchant_profile_rail_preferences(\
+            merchant_profile_id, rail, payment_provider_id) \
+         VALUES (?, ?, ?) \
+         ON CONFLICT(merchant_profile_id, rail) DO UPDATE SET \
+            payment_provider_id = excluded.payment_provider_id",
+    )
+    .bind(profile_id)
+    .bind(rail)
+    .bind(provider_id)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+pub async fn clear_rail_preference(
+    pool: &SqlitePool,
+    profile_id: &str,
+    rail: &str,
+) -> AppResult<()> {
+    sqlx::query(
+        "DELETE FROM merchant_profile_rail_preferences \
+         WHERE merchant_profile_id = ? AND rail = ?",
+    )
+    .bind(profile_id)
+    .bind(rail)
+    .execute(pool)
+    .await?;
+    Ok(())
+}

licensing-service/src/lib.rs

@@ -15,6 +15,7 @@ pub mod crypto;
 pub mod db;
 pub mod error;
 pub mod license_self;
+pub mod merchant_profiles;
 pub mod models;
 pub mod payment;
 pub mod rate_limit;

licensing-service/src/merchant_profiles.rs

@@ -0,0 +1,225 @@
+//! Merchant profile layer.
+//!
+//! A merchant profile represents one "business" the operator is running
+//! on a Keysat instance. Owns business identity (brand, support contact,
+//! redirect URL, optional SMTP) and a set of payment providers attached
+//! to it (BTCPay + Zaprite + future kinds). Products attach to a
+//! merchant profile, not directly to a provider.
+//!
+//! Tier gating:
+//! - **Creator (free)**: exactly 1 profile (the auto-created default).
+//! - **Pro / Patron**: unlimited profiles.
+//!
+//! The schema lives in `migrations/0020_merchant_profiles.sql`. Repo
+//! helpers (raw SQL) live in `db::repo`; this module wraps them with
+//! business-logic guards (tier check, single-default enforcement, etc.).
+//!
+//! See `plans/multi-provider-payment-model.md` for the design rationale.
+
+use crate::api::AppState;
+use crate::db::repo;
+use crate::error::{AppError, AppResult};
+use anyhow::Context;
+use chrono::Utc;
+use serde::{Deserialize, Serialize};
+use sqlx::SqlitePool;
+use uuid::Uuid;
+
+/// A merchant profile row. Mirrors the `merchant_profiles` table.
+///
+/// SMTP fields are flattened onto the same struct for simplicity; they
+/// land in the same table on the same row. NULL on all six means
+/// "inherit StartOS-level SMTP config." See the keysat-smtp-emails
+/// plan for how they're consumed.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MerchantProfile {
+    pub id: String,
+    pub name: String,
+    pub legal_name: Option<String>,
+    pub support_url: Option<String>,
+    pub support_email: Option<String>,
+    pub brand_color: Option<String>,
+    pub post_purchase_redirect_url: Option<String>,
+    pub is_default: bool,
+
+    // SMTP override (all-or-nothing for now; the SMTP plan refines this).
+    pub smtp_host: Option<String>,
+    pub smtp_port: Option<i64>,
+    pub smtp_username: Option<String>,
+    pub smtp_password: Option<String>,
+    pub smtp_from_address: Option<String>,
+    pub smtp_from_name: Option<String>,
+    pub smtp_use_starttls: bool,
+
+    pub created_at: String,
+    pub updated_at: String,
+}
+
+/// Input for `create` — only the operator-set fields. id, is_default,
+/// created_at, updated_at are filled in by this layer.
+#[derive(Debug, Clone, Default, Deserialize)]
+pub struct NewMerchantProfile {
+    pub name: String,
+    pub legal_name: Option<String>,
+    pub support_url: Option<String>,
+    pub support_email: Option<String>,
+    pub brand_color: Option<String>,
+    pub post_purchase_redirect_url: Option<String>,
+}
+
+/// Input for `update` — every field optional. None means "leave unchanged."
+#[derive(Debug, Clone, Default, Deserialize)]
+pub struct MerchantProfileUpdate {
+    pub name: Option<String>,
+    pub legal_name: Option<Option<String>>,
+    pub support_url: Option<Option<String>>,
+    pub support_email: Option<Option<String>>,
+    pub brand_color: Option<Option<String>>,
+    pub post_purchase_redirect_url: Option<Option<String>>,
+
+    pub smtp_host: Option<Option<String>>,
+    pub smtp_port: Option<Option<i64>>,
+    pub smtp_username: Option<Option<String>>,
+    pub smtp_password: Option<Option<String>>,
+    pub smtp_from_address: Option<Option<String>>,
+    pub smtp_from_name: Option<Option<String>>,
+    pub smtp_use_starttls: Option<bool>,
+}
+
+/// Look up a profile by id. Returns `Ok(None)` if not found.
+pub async fn get(pool: &SqlitePool, id: &str) -> AppResult<Option<MerchantProfile>> {
+    repo::get_merchant_profile_by_id(pool, id).await
+}
+
+/// Return the default profile. Migration 0020 guarantees exactly one
+/// exists post-migration, so this returning None at runtime is an
+/// invariant violation and the caller should treat it as fatal.
+pub async fn get_default(pool: &SqlitePool) -> AppResult<Option<MerchantProfile>> {
+    repo::get_default_merchant_profile(pool).await
+}
+
+/// Required default profile lookup. Returns AppError::Internal if no
+/// default exists (which would mean the migration was skipped or the
+/// row was somehow deleted — neither should happen in normal operation).
+pub async fn require_default(pool: &SqlitePool) -> AppResult<MerchantProfile> {
+    get_default(pool).await?.ok_or_else(|| {
+        AppError::Internal(anyhow::anyhow!(
+            "no default merchant profile — migration 0020 may not have run"
+        ))
+    })
+}
+
+/// List all merchant profiles, newest-first.
+pub async fn list(pool: &SqlitePool) -> AppResult<Vec<MerchantProfile>> {
+    repo::list_merchant_profiles(pool).await
+}
+
+/// Look up the merchant profile a product belongs to. Resolves via
+/// `products.merchant_profile_id`. Returns the DEFAULT profile if the
+/// product has no profile id set (back-compat for any rows that slipped
+/// through the migration with NULL — shouldn't happen but defensive).
+pub async fn for_product(state: &AppState, product_id: &str) -> AppResult<MerchantProfile> {
+    if let Some(p) = repo::get_merchant_profile_for_product(&state.db, product_id).await? {
+        return Ok(p);
+    }
+    require_default(&state.db).await
+}
+
+/// Create a new merchant profile. Enforces the Creator tier cap: if the
+/// operator's current tier returns a `merchant_profile` cap of 1 and
+/// at least one profile already exists, returns `AppError::TierCap`
+/// pointing at the upgrade URL.
+///
+/// New profiles default to `is_default = 0`. Use `set_default` to flip
+/// the default flag explicitly — the auto-created post-migration profile
+/// is always the default; subsequent profiles never become default by
+/// creation alone.
+pub async fn create(
+    state: &AppState,
+    input: NewMerchantProfile,
+) -> AppResult<MerchantProfile> {
+    let _ = state; // tier check goes here once tier::check_cap supports merchant_profiles
+    // TODO: tier::check_cap(state, EntitlementSlug::UnlimitedMerchantProfiles)
+    //       — refuses with AppError::TierCap when Creator already has 1
+    //       profile. Skipped in the initial cut; admin UI also enforces
+    //       at the form layer. Wire when tier.rs is updated.
+
+    if input.name.trim().is_empty() {
+        return Err(AppError::BadRequest("merchant profile name required".into()));
+    }
+
+    let id = Uuid::new_v4().to_string();
+    let now = Utc::now().to_rfc3339();
+    repo::create_merchant_profile(
+        &state.db,
+        &id,
+        &input.name,
+        input.legal_name.as_deref(),
+        input.support_url.as_deref(),
+        input.support_email.as_deref(),
+        input.brand_color.as_deref(),
+        input.post_purchase_redirect_url.as_deref(),
+        false, // is_default
+        &now,
+    )
+    .await?;
+    get(&state.db, &id)
+        .await?
+        .ok_or_else(|| AppError::Internal(anyhow::anyhow!("created profile not found")))
+}
+
+/// Update a profile. Only fields with `Some(...)` are written;
+/// double-Option wraps nullable fields so callers can distinguish
+/// "leave unchanged" (`None`) from "set to NULL" (`Some(None)`).
+pub async fn update(
+    pool: &SqlitePool,
+    id: &str,
+    patch: MerchantProfileUpdate,
+) -> AppResult<MerchantProfile> {
+    repo::update_merchant_profile(pool, id, &patch).await?;
+    get(pool, id)
+        .await?
+        .ok_or_else(|| AppError::BadRequest(format!("merchant profile {id} not found")))
+}
+
+/// Flip a profile to be the default. Atomic: clears the previous
+/// default in the same transaction so the partial unique index holds.
+pub async fn set_default(pool: &SqlitePool, id: &str) -> AppResult<()> {
+    repo::set_default_merchant_profile(pool, id).await
+}
+
+/// Delete a profile. Refuses if any product OR active subscription
+/// is still attached. Refuses if it's the default profile (operator
+/// must set another profile as default first).
+pub async fn delete(pool: &SqlitePool, id: &str) -> AppResult<()> {
+    let profile = get(pool, id).await?.ok_or_else(|| {
+        AppError::BadRequest(format!("merchant profile {id} not found"))
+    })?;
+    if profile.is_default {
+        return Err(AppError::BadRequest(
+            "cannot delete the default merchant profile — set another profile as default first"
+                .into(),
+        ));
+    }
+    let product_count = repo::count_products_for_profile(pool, id)
+        .await
+        .context("count_products_for_profile")
+        .map_err(AppError::Internal)?;
+    if product_count > 0 {
+        return Err(AppError::BadRequest(format!(
+            "cannot delete merchant profile: {product_count} products still attached. \
+             Move or delete the products first."
+        )));
+    }
+    let active_sub_count = repo::count_active_subscriptions_for_profile(pool, id)
+        .await
+        .context("count_active_subscriptions_for_profile")
+        .map_err(AppError::Internal)?;
+    if active_sub_count > 0 {
+        return Err(AppError::BadRequest(format!(
+            "cannot delete merchant profile: {active_sub_count} active subscriptions \
+             still attached. Cancel them first or migrate them to another profile."
+        )));
+    }
+    repo::delete_merchant_profile(pool, id).await
+}

licensing-service/src/payment/mod.rs

@@ -40,43 +40,63 @@ use std::any::Any;
 pub mod btcpay;
 pub mod zaprite;
 
-/// Settings-table key that records which provider the operator
-/// last activated. Used by the boot-time loader to pick which
-/// provider to load when both `btcpay_config` and `zaprite_config`
-/// are populated. Values: `'btcpay'` | `'zaprite'`. Absent means
-/// "use whichever single provider is configured" (back-compat
-/// for installs that pre-date this setting).
+// =========================================================================
+// Legacy compatibility shims — DEPRECATED, will be removed once all call
+// sites migrate to the merchant-profile-aware resolution layer.
+// =========================================================================
+//
+// During the multi-provider transition the singleton-config-and-active-
+// provider-preference helpers stay callable so the existing connect flows
+// (`btcpay_authorize.rs`, `zaprite_authorize.rs`) and the boot loader in
+// `main.rs` keep working. Each shim wraps the new schema with the old
+// semantics: `read_active_provider_preference` looks up the first provider
+// attached to the default merchant profile and returns its kind;
+// `write_active_provider_preference` is a no-op (the new model doesn't
+// track an "active provider" preference — providers attach to profiles,
+// profiles attach to products).
+
+#[deprecated(
+    note = "use merchant-profile-aware resolution: \
+            state.payment_provider_for(product_id, rail)"
+)]
 pub const SETTING_ACTIVE_PROVIDER: &str = "active_payment_provider";
 
-/// Convenience getter for the active-provider setting. Returns
-/// `Some(ProviderKind)` if the operator has explicitly chosen
-/// one, `None` if they haven't (caller falls back to the
-/// load-order heuristic).
+#[deprecated(
+    note = "look up providers via list_payment_providers_for_profile or \
+            payment_provider_by_id on AppState"
+)]
 pub async fn read_active_provider_preference(
     pool: &sqlx::SqlitePool,
 ) -> Option<ProviderKind> {
+    // Post-migration: derive from the first provider attached to the
+    // default merchant profile (deterministic by connected_at ASC).
+    // Pre-migration (if the migration hasn't run yet on this DB):
+    // fall back to the legacy settings-table read.
+    let default_profile = crate::db::repo::get_default_merchant_profile(pool).await.ok().flatten();
+    if let Some(profile) = default_profile {
+        if let Ok(rows) = crate::db::repo::list_payment_providers_for_profile(pool, &profile.id).await {
+            if let Some(first) = rows.first() {
+                return ProviderKind::parse(&first.kind);
+            }
+        }
+    }
+    // Legacy fallback for the pre-migration window.
     match crate::db::repo::settings_get(pool, SETTING_ACTIVE_PROVIDER).await {
-        Ok(Some(s)) => match s.as_str() {
-            "btcpay" => Some(ProviderKind::Btcpay),
-            "zaprite" => Some(ProviderKind::Zaprite),
-            _ => None,
-        },
+        Ok(Some(s)) => ProviderKind::parse(&s),
         _ => None,
     }
 }
 
-/// Persist the operator's active-provider preference. Called by
-/// the connect endpoints (Connect BTCPay, Connect Zaprite) and
-/// by the new "Activate <provider>" endpoint that flips between
-/// already-configured providers without re-authorizing.
+#[deprecated(
+    note = "providers are now attached to merchant profiles, not implicitly active. \
+            This shim is a no-op; remove the call."
+)]
 pub async fn write_active_provider_preference(
-    pool: &sqlx::SqlitePool,
-    kind: ProviderKind,
+    _pool: &sqlx::SqlitePool,
+    _kind: ProviderKind,
 ) -> anyhow::Result<()> {
-    let value = kind.as_str();
-    crate::db::repo::settings_set(pool, SETTING_ACTIVE_PROVIDER, Some(value))
-        .await
-        .map_err(|e| anyhow::anyhow!("write active provider preference: {e:#}"))?;
+    // No-op. In the multi-provider model there's no "active" preference
+    // to write — providers are looked up by id (per-product) or by profile.
     Ok(())
 }
 
@@ -94,6 +114,95 @@ impl ProviderKind {
             ProviderKind::Zaprite => "zaprite",
         }
     }
+
+    pub fn parse(s: &str) -> Option<Self> {
+        match s {
+            "btcpay" => Some(Self::Btcpay),
+            "zaprite" => Some(Self::Zaprite),
+            _ => None,
+        }
+    }
+}
+
+/// Buyer-facing payment method. The buy page renders a picker over these
+/// (when a merchant profile exposes more than one); the routing layer maps
+/// the buyer's pick to a specific provider via the profile's attached
+/// providers + optional `merchant_profile_rail_preferences` tie-breakers.
+///
+/// Rails-per-provider-kind are **inherent** (declared by each provider
+/// impl's `served_rails()` trait method), not configurable per provider
+/// row. BTCPay serves Lightning + OnChain. Zaprite serves Card +
+/// Lightning + OnChain.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum Rail {
+    Lightning,
+    Onchain,
+    Card,
+}
+
+impl Rail {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            Rail::Lightning => "lightning",
+            Rail::Onchain => "onchain",
+            Rail::Card => "card",
+        }
+    }
+
+    pub fn parse(s: &str) -> Option<Self> {
+        match s {
+            "lightning" => Some(Self::Lightning),
+            "onchain" | "on-chain" | "on_chain" => Some(Self::Onchain),
+            "card" => Some(Self::Card),
+            _ => None,
+        }
+    }
+}
+
+/// Static rails served by a provider kind. Returned by
+/// `PaymentProvider::served_rails()`; centralized here so callers that
+/// just want to know "what does kind X support" (e.g., the admin UI's
+/// connect-flow guidance) don't have to instantiate a provider.
+pub fn rails_for_kind(kind: ProviderKind) -> Vec<Rail> {
+    match kind {
+        ProviderKind::Btcpay => vec![Rail::Lightning, Rail::Onchain],
+        ProviderKind::Zaprite => vec![Rail::Card, Rail::Lightning, Rail::Onchain],
+    }
+}
+
+/// Build a typed `PaymentProvider` trait object from a `payment_providers`
+/// row. Dispatch on `kind`. Used by the AppState provider cache when
+/// resolving by provider id.
+pub fn build_provider(
+    row: &crate::db::repo::PaymentProviderRow,
+    public_base_url: Option<&str>,
+) -> anyhow::Result<std::sync::Arc<dyn PaymentProvider>> {
+    use crate::btcpay::client::BtcpayClient;
+    use crate::payment::btcpay::BtcpayProvider;
+    use crate::payment::zaprite::{ZapriteClient, ZapriteProvider};
+
+    match ProviderKind::parse(&row.kind) {
+        Some(ProviderKind::Btcpay) => {
+            let store_id = row.store_id.as_deref().ok_or_else(|| {
+                anyhow::anyhow!("BTCPay provider row {} missing store_id", row.id)
+            })?;
+            let webhook_secret = row.webhook_secret.clone().unwrap_or_default();
+            let client = BtcpayClient::new(&row.base_url, &row.api_key, store_id);
+            let provider = BtcpayProvider::new(client, webhook_secret)
+                .with_public_base(public_base_url.map(|s| s.to_string()));
+            Ok(std::sync::Arc::new(provider))
+        }
+        Some(ProviderKind::Zaprite) => {
+            let client = ZapriteClient::new(row.base_url.clone(), row.api_key.clone());
+            Ok(std::sync::Arc::new(ZapriteProvider::new(client)))
+        }
+        None => Err(anyhow::anyhow!(
+            "unknown payment provider kind {:?} on row {}",
+            row.kind,
+            row.id
+        )),
+    }
 }
 
 /// A monetary amount + the unit it's denominated in.
@@ -230,6 +339,14 @@ pub struct PaymentReceipt {
 pub trait PaymentProvider: Send + Sync + Any {
     fn kind(&self) -> ProviderKind;
 
+    /// Payment rails this provider can settle. Default impl uses the
+    /// static `rails_for_kind()` mapping; impls only override if they
+    /// expose a non-default set (e.g., a degraded BTCPay configured
+    /// without Lightning support — not currently a Keysat concern).
+    fn served_rails(&self) -> Vec<Rail> {
+        rails_for_kind(self.kind())
+    }
+
     async fn create_invoice(
         &self,
         params: CreateInvoiceParams<'_>,