//! # openfut-utas-host
//!
//! The first live FIFA 17 **UTAS migration host**. It fronts the client-visible
//! UTAS port and does route-level migration:
//!
//! ```text
//!   FIFA 17  ──HTTP──▶  openfut-utas-host
//!                         ├── GET …/club  ──▶ FIFA17 adapter ──▶ OpenFUT Core
//!                         └── everything else ──▶ Python UTAS oracle (verbatim)
//! ```
//!
//! ## Safety rules (see the mission brief)
//!
//! * **Classification happens once, before any execution** ([`classify`]). A
//!   request is either handled in Rust or proxied to Python — never both, and
//!   there is NO "try Rust then retry on Python", which could double-apply a
//!   mutation. `/club` is read-only, but the rule holds regardless.
//! * The Rust `/club` path NEVER contacts Python; the passthrough path NEVER
//!   runs Core logic.
//! * A Core failure on `/club` returns an empty (but valid) `{"itemData":[]}`
//!   and logs an error — it does NOT fall back to Python.
//!
//! ## Transport (worker D)
//!
//! UTAS is plaintext HTTP/1.1 keep-alive, no TLS. Body is read by `Content-Length`
//! before responding; responses carry `Content-Length` and `Content-Type:
//! application/json` only when a body is present.
//!
//! ## The asset-id boundary
//!
//! FIFA renders an owned card from a real FIFA asset id (`resourceId & 0xffffff`
//! resolved against the client's local DB). Core's synthetic catalogue has none,
//! so [`ItemIdentityResolver`] is injected and unresolved items are dropped, not
//! faked (see `club_response`). With today's empty mapping, `/club` returns
//! `{"itemData":[]}` — the honest state until Core inventory is asset-backed.

pub mod config;

use std::io::{BufRead, BufReader, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, Mutex};
use std::time::Instant;

use openfut_adapter_fifa17::fut::catalog::{Fifa17CardCatalog, Fifa17WireItemIdPolicy};
use openfut_adapter_fifa17::fut::club_response::{
    shape_club_response, CoreOwnedItem, Fifa17Identity, ItemIdentityResolver, ShapeStats,
};
use openfut_adapter_fifa17::fut::entities::Fifa17Entities;
use openfut_adapter_fifa17::fut::owned_query::{
    is_special_rareflag, map_to_core, parse_club_query, MapError,
};
use openfut_adapter_fifa17::fut::squad::{parse_squad_put, save_ack, SquadWireResolver};
use openfut_adapter_fifa17::fut::squad_ext::{
    build_squad_write, Fifa17SquadExtensionV1, SquadBuildError, EXT_NAMESPACE, EXT_SCHEMA_VERSION,
};
use openfut_adapter_fifa17::fut::squad_projection::{
    project_squad, squad_list, user_mass_info_squad, ProjectionSlot, SquadExtInput,
    SquadProjection, SquadProjectionInput,
};
use openfut_adapter_fifa17::fut::store_session::{
    validate_capability, SessionStore, StoreMode, SENTINEL_PACK_ID,
};
use openfut_identity::ExternalIdentityStore;
use serde_json::{json, Value};

use config::HostConfig;

// ───────────────────────────── Route classification ─────────────────────────

/// The route decision, taken once, before execution.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Route {
    /// `GET …/club` — the owned-player search, served from Core.
    Club,
    /// `PUT …/squad/<n>` — full squad replacement, committed to Core.
    SquadReplace,
    /// `GET …/squad/list` — the squad summary, projected from Core.
    SquadList,
    /// `GET …/squad/active` — the active squad object, projected from Core.
    SquadActive,
    /// `GET …/userMassInfo` — proxied to Python, with only `.squad` overlaid.
    UserMassInfo,
    /// `POST /ut/auth` — proxied to Python (persona adoption + SID mint); the
    /// returned `X-UT-SID` is observed to open a Rust session.
    Auth,
    /// `POST /openfut/fifa17/capability` — launcher capability registration,
    /// owned entirely in Rust (no economy, no proxy).
    Capability,
    /// `GET …/store/purchasegroup…` — proxied to Python for the authoritative
    /// economy body, with the empty-My-Packs topology overlaid from the Rust
    /// session mode (the 65534 sentinel is stripped for a verified clean-v1 SID).
    StorePurchaseGroup,
    /// Anything else — proxied verbatim to the Python oracle.
    Passthrough,
}

/// Classify a request ONCE, before execution. Rust owns exactly:
/// * `GET …/club`
/// * `PUT …/squad/<n>` (numeric id)
/// * `GET …/squad/list`
/// * `GET …/squad/active` (the active squad, projected from Core)
/// * `GET …/userMassInfo` (proxied, `.squad` overlaid)
///
/// Everything else — numeric `GET …/squad/<n>`, `/clubUser`, auth, packs,
/// market, other mutations — falls through to Python. There is no
/// "try Rust then Python", so a squad mutation can never be double-applied.
pub fn classify(method: &str, path: &str) -> Route {
    let get = method.eq_ignore_ascii_case("GET");
    let put = method.eq_ignore_ascii_case("PUT");
    let post = method.eq_ignore_ascii_case("POST");
    // Session/capability vertical (Rust session authority; economy stays Python).
    if post && path.starts_with("/ut/auth") {
        return Route::Auth;
    }
    if post && path == "/openfut/fifa17/capability" {
        return Route::Capability;
    }
    if get && is_exact_club_path(path) {
        return Route::Club;
    }
    match ut_tail(path) {
        Some("squad/list") if get => Route::SquadList,
        Some("squad/active") if get => Route::SquadActive,
        Some("userMassInfo") if get => Route::UserMassInfo,
        Some(tail) if get && tail.starts_with("store/purchasegroup") => Route::StorePurchaseGroup,
        Some(tail) if put && is_numeric_squad_tail(tail) => Route::SquadReplace,
        _ => Route::Passthrough,
    }
}

/// The tail after `/ut/game/<title>/` (non-empty title), or `None`.
fn ut_tail(path: &str) -> Option<&str> {
    let rest = path.strip_prefix("/ut/game/")?;
    let (title, tail) = rest.split_once('/')?;
    if title.is_empty() {
        None
    } else {
        Some(tail)
    }
}

fn is_exact_club_path(path: &str) -> bool {
    ut_tail(path) == Some("club")
}

/// `squad/<digits>` — the numeric full-squad target used by `PUT`. The active
/// squad READ (`GET …/squad/active`) is routed separately (Core-backed); a
/// numeric `GET …/squad/<n>` for a non-active squad stays on Python (there is no
/// Core model for multiple squads yet).
fn is_numeric_squad_tail(tail: &str) -> bool {
    match tail.strip_prefix("squad/") {
        Some(id) => !id.is_empty() && id.bytes().all(|b| b.is_ascii_digit()),
        None => false,
    }
}

// ───────────────────────────── Core access boundary ─────────────────────────

/// Failure reaching or reading OpenFUT Core.
#[derive(Debug)]
pub enum CoreError {
    Http(String),
    Status(u16),
    Parse(String),
}

impl std::fmt::Display for CoreError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            CoreError::Http(e) => write!(f, "core http error: {e}"),
            CoreError::Status(s) => write!(f, "core returned status {s}"),
            CoreError::Parse(e) => write!(f, "core response parse error: {e}"),
        }
    }
}

/// One page of owned items plus the filtered total, as returned by Core.
pub struct CorePage {
    pub items: Vec<CoreOwnedItem>,
    pub total: i64,
}

/// One canonical squad slot as read back from Core (game-independent).
#[derive(Debug, Clone)]
pub struct CoreSquadSlot {
    pub owned_card_id: String,
    pub index: i64,
    pub is_captain: bool,
    pub is_on_bench: bool,
}

/// Freshness of the stored opaque extension vs the current canonical squad, as
/// Core reports it. Carries the stored payload for Fresh/Stale (never applied
/// when Stale — the host decides policy).
#[derive(Debug, Clone)]
pub enum CoreExtState {
    Fresh {
        schema_version: i64,
        payload: String,
    },
    Stale {
        schema_version: i64,
        payload: String,
    },
    Missing,
}

/// The active squad, its canonical slots, and its opaque extension state — the
/// result of Core's `GET /squad/ext`.
#[derive(Debug, Clone)]
pub struct CoreSquadRead {
    pub name: String,
    /// FIFA formation token, verbatim (Core stores it opaquely).
    pub formation: String,
    pub slots: Vec<CoreSquadSlot>,
    pub ext: CoreExtState,
}

/// A canonical + extension squad replacement the host asks Core to commit
/// atomically (`PUT /squad/replace`).
pub struct CoreReplaceRequest {
    pub name: Option<String>,
    pub formation: Option<String>,
    pub slots: Vec<CoreSquadSlot>,
    pub client_reported: CoreClientEval,
    pub ext_namespace: String,
    pub ext_schema_version: i64,
    pub ext_payload: String,
}

/// Client-reported shadow evaluation carried through to Core (never Core's
/// authoritative evaluation).
#[derive(Debug, Clone, Default)]
pub struct CoreClientEval {
    pub chemistry: Option<i64>,
    pub rating: Option<i64>,
    pub star_rating: Option<i64>,
}

/// Outcome of a committed replacement.
#[derive(Debug, Clone)]
pub struct CoreReplaceResult {
    pub squad_id: String,
    pub canonical_fingerprint: String,
    pub slots_written: usize,
}

/// How the host reaches Core. The adapter never sees this — the host owns the
/// transport, mirroring the architecture rule. Tests inject a fake.
pub trait CoreAccess: Send + Sync {
    /// Query the owned inventory with semantic `/collection` query params.
    fn query_owned(&self, params: &[(&str, String)]) -> Result<CorePage, CoreError>;

    /// Every owned item for the active club, in one call (no pagination) — used
    /// to assemble a whole squad projection and to authorize squad writes. The
    /// default delegates to an unfiltered `query_owned`.
    fn all_owned(&self) -> Result<Vec<CoreOwnedItem>, CoreError> {
        Ok(self.query_owned(&[])?.items)
    }

    /// Read the active squad + its opaque extension for `namespace`.
    fn read_squad_ext(&self, namespace: &str) -> Result<CoreSquadRead, CoreError>;

    /// Replace the active squad's canonical slots + opaque extension atomically.
    fn replace_squad(&self, req: &CoreReplaceRequest) -> Result<CoreReplaceResult, CoreError>;
}

/// Default HTTP implementation: `GET {core_url}/collection?…` (plain HTTP JSON,
/// the same boundary Bridge uses to reach Core). Every request carries
/// `X-OpenFUT-Game: <game>` so Core resolves the game-scoped active profile — for
/// FIFA 17 that is the profile whose inventory is fully asset-mapped, so `/club`
/// filters/paginates a wholly renderable set (no post-pagination drops).
pub struct HttpCoreClient {
    base_url: String,
    game: String,
    client: reqwest::blocking::Client,
}

impl HttpCoreClient {
    pub fn new(base_url: impl Into<String>, game: impl Into<String>) -> Self {
        HttpCoreClient {
            base_url: base_url.into().trim_end_matches('/').to_string(),
            game: game.into(),
            client: reqwest::blocking::Client::new(),
        }
    }
}

impl CoreAccess for HttpCoreClient {
    fn query_owned(&self, params: &[(&str, String)]) -> Result<CorePage, CoreError> {
        let url = format!("{}/collection", self.base_url);
        let resp = self
            .client
            .get(&url)
            .header("X-OpenFUT-Game", &self.game)
            .query(params)
            .send()
            .map_err(|e| CoreError::Http(e.to_string()))?;
        let status = resp.status().as_u16();
        if !(200..300).contains(&status) {
            return Err(CoreError::Status(status));
        }
        let v: Value = resp.json().map_err(|e| CoreError::Parse(e.to_string()))?;
        parse_core_page(&v)
    }
    fn read_squad_ext(&self, namespace: &str) -> Result<CoreSquadRead, CoreError> {
        let url = format!("{}/squad/ext", self.base_url);
        let resp = self
            .client
            .get(&url)
            .header("X-OpenFUT-Game", &self.game)
            .query(&[("namespace", namespace)])
            .send()
            .map_err(|e| CoreError::Http(e.to_string()))?;
        let status = resp.status().as_u16();
        if !(200..300).contains(&status) {
            return Err(CoreError::Status(status));
        }
        let v: Value = resp.json().map_err(|e| CoreError::Parse(e.to_string()))?;
        parse_core_squad_read(&v)
    }

    fn replace_squad(&self, req: &CoreReplaceRequest) -> Result<CoreReplaceResult, CoreError> {
        let url = format!("{}/squad/replace", self.base_url);
        let resp = self
            .client
            .put(&url)
            .header("X-OpenFUT-Game", &self.game)
            .json(&replace_request_body(req))
            .send()
            .map_err(|e| CoreError::Http(e.to_string()))?;
        let status = resp.status().as_u16();
        if !(200..300).contains(&status) {
            return Err(CoreError::Status(status));
        }
        let v: Value = resp.json().map_err(|e| CoreError::Parse(e.to_string()))?;
        Ok(CoreReplaceResult {
            squad_id: v
                .get("squad_id")
                .and_then(|x| x.as_str())
                .unwrap_or("")
                .to_string(),
            canonical_fingerprint: v
                .get("canonical_fingerprint")
                .and_then(|x| x.as_str())
                .unwrap_or("")
                .to_string(),
            slots_written: v.get("slots_written").and_then(|x| x.as_u64()).unwrap_or(0) as usize,
        })
    }
}

/// Serialize a [`CoreReplaceRequest`] into the `PUT /squad/replace` JSON body.
pub fn replace_request_body(req: &CoreReplaceRequest) -> Value {
    let slots: Vec<Value> = req
        .slots
        .iter()
        .map(|s| {
            json!({
                "owned_card_id": s.owned_card_id,
                "slot": s.index,
                "is_captain": s.is_captain,
                "is_on_bench": s.is_on_bench,
            })
        })
        .collect();
    json!({
        "name": req.name,
        "formation": req.formation,
        "slots": slots,
        "client_reported": {
            "client_reported_chemistry": req.client_reported.chemistry,
            "client_reported_rating": req.client_reported.rating,
            "client_reported_star_rating": req.client_reported.star_rating,
        },
        "extension": {
            "namespace": req.ext_namespace,
            "schema_version": req.ext_schema_version,
            "payload": req.ext_payload,
        },
    })
}

/// Parse Core's `GET /squad/ext` response into a [`CoreSquadRead`].
pub fn parse_core_squad_read(v: &Value) -> Result<CoreSquadRead, CoreError> {
    let squad = v
        .get("squad")
        .ok_or_else(|| CoreError::Parse("missing `squad`".into()))?;
    let name = squad
        .get("name")
        .and_then(|x| x.as_str())
        .unwrap_or("")
        .to_string();
    let formation = squad
        .get("formation")
        .and_then(|x| x.as_str())
        .ok_or_else(|| CoreError::Parse("missing squad.formation".into()))?
        .to_string();
    let players = v
        .get("players")
        .and_then(|p| p.as_array())
        .ok_or_else(|| CoreError::Parse("missing `players`".into()))?;
    let slots = players
        .iter()
        .filter_map(|p| {
            Some(CoreSquadSlot {
                owned_card_id: p.get("owned_card_id")?.as_str()?.to_string(),
                index: p.get("position_index")?.as_i64()?,
                is_captain: p
                    .get("is_captain")
                    .and_then(|x| x.as_bool())
                    .unwrap_or(false),
                is_on_bench: p
                    .get("is_on_bench")
                    .and_then(|x| x.as_bool())
                    .unwrap_or(false),
            })
        })
        .collect();
    let ext_v = v
        .get("extension")
        .ok_or_else(|| CoreError::Parse("missing `extension`".into()))?;
    let ext = match ext_v.get("state").and_then(|x| x.as_str()) {
        Some("fresh") => CoreExtState::Fresh {
            schema_version: ext_v
                .get("schema_version")
                .and_then(|x| x.as_i64())
                .unwrap_or(0),
            payload: ext_v
                .get("payload")
                .and_then(|x| x.as_str())
                .unwrap_or("")
                .to_string(),
        },
        Some("stale") => CoreExtState::Stale {
            schema_version: ext_v
                .get("schema_version")
                .and_then(|x| x.as_i64())
                .unwrap_or(0),
            payload: ext_v
                .get("payload")
                .and_then(|x| x.as_str())
                .unwrap_or("")
                .to_string(),
        },
        Some("missing") => CoreExtState::Missing,
        other => {
            return Err(CoreError::Parse(format!(
                "unknown extension state {other:?}"
            )))
        }
    };
    Ok(CoreSquadRead {
        name,
        formation,
        slots,
        ext,
    })
}

/// Parse Core's `/collection` response `{ "collection": [...], "total": n }` into
/// semantic owned items.
pub fn parse_core_page(v: &Value) -> Result<CorePage, CoreError> {
    let arr = v
        .get("collection")
        .and_then(|c| c.as_array())
        .ok_or_else(|| CoreError::Parse("missing `collection` array".into()))?;
    let total = v
        .get("total")
        .and_then(|t| t.as_i64())
        .unwrap_or(arr.len() as i64);
    let items = arr.iter().filter_map(core_item_from_json).collect();
    Ok(CorePage { items, total })
}

fn core_item_from_json(e: &Value) -> Option<CoreOwnedItem> {
    let card = e.get("card")?;
    let attr = |k: &str| card.get(k).and_then(|v| v.as_i64()).unwrap_or(0) as u8;
    let position = e
        .get("effective_position")
        .and_then(|v| v.as_str())
        .or_else(|| card.get("position").and_then(|v| v.as_str()))?
        .to_string();
    let rating = e
        .get("effective_overall")
        .and_then(|v| v.as_i64())
        .or_else(|| card.get("overall").and_then(|v| v.as_i64()))
        .unwrap_or(0) as u8;
    Some(CoreOwnedItem {
        owned_card_id: e.get("owned_card_id")?.as_str()?.to_string(),
        card_id: card.get("id")?.as_str()?.to_string(),
        rating,
        position,
        nation: card.get("nation")?.as_str()?.to_string(),
        league: card.get("league")?.as_str()?.to_string(),
        club: card.get("club")?.as_str()?.to_string(),
        attributes: [
            attr("pace"),
            attr("shooting"),
            attr("passing"),
            attr("dribbling"),
            attr("defending"),
            attr("physical"),
        ],
    })
}

// ───────────────────────────── Item identity resolver ───────────────────────

/// The single production [`ItemIdentityResolver`]: it composes the two distinct
/// FIFA 17 identities from real, persistent sources — no placeholder, no hash,
/// no fabricated id.
///
/// * **Definition identity** (`resourceId`/`assetId`) comes from the
///   [`Fifa17CardCatalog`]: `card_id` → real FIFA asset id. An unmapped
///   definition resolves to `None` → the item is dropped and counted, never
///   faked.
/// * **Instance identity** (`item_id`) comes from the generic
///   [`ExternalIdentityStore`] under the FIFA 17 wire-id policy: the same owned
///   instance always resolves to the same monotonic wire id, it survives
///   restart, and it reverses exactly. Two copies of the same definition share a
///   `resourceId` but get distinct `item_id`s.
///
/// The wire-id namespace is **globally monotonic within `(game, "owned-item")`**,
/// not per-account. Python restarts numbering per save file; Core owned-instance
/// ids are globally-unique UUIDs, so a single monotonic sequence keeps every
/// wire id unique and its reverse lookup unambiguous across all accounts —
/// satisfying the client's only requirement (per-session unique/stable/
/// reversible ids). An account column is therefore unnecessary.
pub struct Fifa17IdentityResolver {
    catalog: Fifa17CardCatalog,
    store: Arc<dyn ExternalIdentityStore>,
}

impl Fifa17IdentityResolver {
    pub fn new(catalog: Fifa17CardCatalog, store: Arc<dyn ExternalIdentityStore>) -> Self {
        Fifa17IdentityResolver { catalog, store }
    }

    /// Reverse an owned-item wire id back to its Core owned-instance id (used by
    /// later item-operation slices). `None` = unknown wire id, never a guess.
    pub fn owned_id_for_wire(&self, wire: i64) -> Option<String> {
        self.store
            .core_for(
                Fifa17WireItemIdPolicy::GAME,
                Fifa17WireItemIdPolicy::OWNED_ITEM_KIND,
                wire,
            )
            .unwrap_or(None)
    }
}

impl ItemIdentityResolver for Fifa17IdentityResolver {
    fn resolve(&self, item: &CoreOwnedItem) -> Option<Fifa17Identity> {
        // Definition identity first: an unmapped card is dropped (never faked).
        let ident = self.catalog.lookup(&item.card_id)?;
        // Instance identity: stable, persistent, reversible wire id.
        let wire = match self.store.resolve_or_allocate(
            Fifa17WireItemIdPolicy::GAME,
            Fifa17WireItemIdPolicy::OWNED_ITEM_KIND,
            &item.owned_card_id,
            Fifa17WireItemIdPolicy::owned_item_base_floor(),
        ) {
            Ok(w) => w,
            Err(e) => {
                // Infrastructure failure allocating a wire id: drop this item
                // (freeze-safe) and log — never emit an unstable/fake id.
                eprintln!(
                    "utas-host ERROR identity store alloc failed for {}: {e}",
                    item.owned_card_id
                );
                return None;
            }
        };
        Some(Fifa17Identity {
            // Wire ids live in 1e8..9e8 (policy) — well within u32.
            item_id: wire as u32,
            asset_id: ident.asset_id,
            resource_id: ident.resource_id,
            rareflag: ident.rareflag,
        })
    }
}

/// The same production resolver reverses a wire id to a Core owned-instance id
/// for the squad PUT path — reusing the identity store, so `/club`, `/squad`,
/// and PUT all agree on wire↔owned. Identity ONLY; ownership is authorized
/// separately (a resolvable id is not proof of ownership).
impl SquadWireResolver for Fifa17IdentityResolver {
    fn owned_id_for_wire(&self, wire: i64) -> Option<String> {
        Fifa17IdentityResolver::owned_id_for_wire(self, wire)
    }
}

// ───────────────────────────── /club handler ────────────────────────────────

/// Safe, structured summary of a handled `/club` request (no auth/session/device
/// material — the club query carries none; auth is a header we never log).
#[derive(Debug, Clone)]
pub struct ClubLog {
    pub outcome: &'static str,
    pub filter: String,
    pub total: i64,
    pub emitted: usize,
    pub dropped_no_asset: usize,
    pub offset: Option<i64>,
    pub limit: Option<i64>,
}

/// Dependencies for the Rust `/club` path.
pub struct ClubDeps<'a> {
    pub core: &'a dyn CoreAccess,
    pub entities: &'a Fifa17Entities,
    pub assets: &'a (dyn ItemIdentityResolver + Send + Sync),
}

/// Filter already-shaped `/club` items to SPECIALS (`rareflag > 1`) and paginate
/// the filtered set locally. Returns `(page, total_specials)`. Pure — the whole
/// point is that "special" pagination is over the filtered set, never Core's
/// unfiltered page (which would drop specials or leak base cards).
fn special_filter_page(
    items: &[Value],
    offset: Option<i64>,
    limit: Option<i64>,
) -> (Vec<Value>, i64) {
    let specials: Vec<&Value> = items
        .iter()
        .filter(|it| {
            it.get("rareflag")
                .and_then(|v| v.as_i64())
                .map(is_special_rareflag)
                .unwrap_or(false)
        })
        .collect();
    let total = specials.len() as i64;
    let off = offset.unwrap_or(0).max(0) as usize;
    let paged: Vec<Value> = match limit {
        Some(l) => specials
            .into_iter()
            .skip(off)
            .take(l.max(0) as usize)
            .cloned()
            .collect(),
        None => specials.into_iter().skip(off).cloned().collect(),
    };
    (paged, total)
}

/// Handle `GET …/club?…` end to end: parse → map ids to names → Core query →
/// shape to the FIFA `{itemData:[…]}` envelope. Always returns HTTP 200 with a
/// JSON body (UTAS must never 401/403; an empty result is the safe degrade).
pub fn handle_club(query: &str, deps: &ClubDeps<'_>) -> (WireResponse, ClubLog) {
    let raw = parse_club_query(query);
    let core_q = match map_to_core(&raw, deps.entities) {
        Ok(c) => c,
        Err(e) => {
            // Unknown FIFA id — never a raw-id passthrough, never a guess.
            return (
                json_response(&json!({ "itemData": [] })),
                ClubLog {
                    outcome: "unknown_id",
                    filter: describe_map_error(&e),
                    total: 0,
                    emitted: 0,
                    dropped_no_asset: 0,
                    offset: raw.start.map(|s| s as i64),
                    limit: raw.count.map(|c| c as i64),
                },
            );
        }
    };
    let pairs = core_q.to_query_pairs();
    let filter = summarize(&pairs);
    let (offset, limit) = (core_q.offset, core_q.limit);

    // "Special" (rare=SP): rareflag lives in the FIFA catalog, not Core, so Core
    // cannot filter it. Fetch everything matching the OTHER filters (no
    // offset/limit), shape (which resolves each item's rareflag), keep only
    // specials (rareflag > 1), then paginate the filtered set locally.
    if core_q.special {
        let mut base = core_q.clone();
        base.offset = None;
        base.limit = None;
        return match deps.core.query_owned(&base.to_query_pairs()) {
            Ok(page) => {
                let (body, stats) = shape_club_response(&page.items, deps.entities, deps.assets);
                let all = body
                    .get("itemData")
                    .and_then(|v| v.as_array())
                    .cloned()
                    .unwrap_or_default();
                let (paged, total) = special_filter_page(&all, offset, limit);
                let emitted = paged.len();
                (
                    json_response(&json!({ "itemData": paged })),
                    ClubLog {
                        outcome: "ok",
                        filter: format!("{filter},rare=SP"),
                        total,
                        emitted,
                        dropped_no_asset: stats.dropped_no_asset,
                        offset,
                        limit,
                    },
                )
            }
            Err(e) => {
                eprintln!("utas-host ERROR /club (special) core query failed: {e}");
                (
                    json_response(&json!({ "itemData": [] })),
                    ClubLog {
                        outcome: "core_error",
                        filter: format!("{filter},rare=SP"),
                        total: 0,
                        emitted: 0,
                        dropped_no_asset: 0,
                        offset,
                        limit,
                    },
                )
            }
        };
    }
    match deps.core.query_owned(&pairs) {
        Ok(page) => {
            let (body, stats): (Value, ShapeStats) =
                shape_club_response(&page.items, deps.entities, deps.assets);
            (
                json_response(&body),
                ClubLog {
                    outcome: "ok",
                    filter,
                    total: page.total,
                    emitted: stats.emitted,
                    dropped_no_asset: stats.dropped_no_asset,
                    offset,
                    limit,
                },
            )
        }
        Err(e) => {
            // Degrade to a valid empty page; DO NOT fall back to Python.
            eprintln!("utas-host ERROR /club core query failed: {e}");
            (
                json_response(&json!({ "itemData": [] })),
                ClubLog {
                    outcome: "core_error",
                    filter,
                    total: 0,
                    emitted: 0,
                    dropped_no_asset: 0,
                    offset,
                    limit,
                },
            )
        }
    }
}

fn describe_map_error(e: &MapError) -> String {
    match e {
        MapError::UnknownLeague(id) => format!("unknown_league={id}"),
        MapError::UnknownNation(id) => format!("unknown_nation={id}"),
        MapError::UnknownTeam(id) => format!("unknown_team={id}"),
    }
}

fn summarize(pairs: &[(&str, String)]) -> String {
    pairs
        .iter()
        .map(|(k, v)| format!("{k}={v}"))
        .collect::<Vec<_>>()
        .join(",")
}

// ───────────────────────────── Squad handlers ───────────────────────────────

/// FIFA wire id of the single active squad (`…/squad/0`).
pub const ACTIVE_SQUAD_WIRE_ID: i64 = 0;

/// Dependencies for the Rust squad paths. `resolver` is the SAME production
/// identity resolver `/club` uses (forward shape + reverse wire→owned), so every
/// route agrees on wire↔owned identity.
pub struct SquadDeps<'a> {
    pub core: &'a dyn CoreAccess,
    pub resolver: &'a Fifa17IdentityResolver,
    pub entities: &'a Fifa17Entities,
}

/// Secret-free structured log line for a handled squad request.
#[derive(Debug, Clone)]
pub struct SquadLog {
    pub outcome: &'static str,
    pub detail: String,
}

/// The active squad projected from Core, with the freshness policy applied.
enum HostProjection {
    /// Fresh: the projected FIFA squad object (before any endpoint envelope).
    Squad(Value),
    /// Stored extension is stale vs the canonical squad — NEVER applied.
    Stale,
    /// No extension stored — nothing fabricated.
    Missing,
    /// Core unreachable / response unreadable / projection failed.
    Error(String),
}

/// Assemble the projection input from Core in a BOUNDED number of calls — one
/// `read_squad_ext` + one batch `all_owned`, never per slot — then project. The
/// Fresh/Stale/Missing policy is decided HERE, not buried in a default.
fn project_active_squad(deps: &SquadDeps<'_>) -> HostProjection {
    let read = match deps.core.read_squad_ext(EXT_NAMESPACE) {
        Ok(r) => r,
        Err(e) => return HostProjection::Error(e.to_string()),
    };
    let ext = match &read.ext {
        CoreExtState::Fresh {
            schema_version,
            payload,
        } => {
            match Fifa17SquadExtensionV1::from_payload(*schema_version, payload) {
                Ok(e) => e,
                // Fresh but the payload does not parse as our schema: corruption,
                // never coerced into a fabricated squad.
                Err(e) => return HostProjection::Error(format!("fresh extension unreadable: {e}")),
            }
        }
        CoreExtState::Stale { .. } => return HostProjection::Stale,
        CoreExtState::Missing => return HostProjection::Missing,
    };
    let owned = match deps.core.all_owned() {
        Ok(v) => v,
        Err(e) => return HostProjection::Error(e.to_string()),
    };
    let owned_by_id: std::collections::HashMap<String, CoreOwnedItem> = owned
        .into_iter()
        .map(|i| (i.owned_card_id.clone(), i))
        .collect();
    let slots: Vec<ProjectionSlot> = read
        .slots
        .iter()
        .map(|s| ProjectionSlot {
            owned_card_id: s.owned_card_id.clone(),
            index: s.index,
            is_captain: s.is_captain,
            is_on_bench: s.is_on_bench,
        })
        .collect();
    let input = SquadProjectionInput {
        fifa_squad_id: ACTIVE_SQUAD_WIRE_ID,
        name: read.name,
        formation: read.formation,
        slots,
        ext: SquadExtInput::Fresh(ext),
        owned: &owned_by_id,
    };
    match project_squad(&input, deps.resolver, deps.entities) {
        Ok(SquadProjection::Projected(v)) => HostProjection::Squad(v),
        Ok(SquadProjection::Stale) => HostProjection::Stale,
        Ok(SquadProjection::Missing) => HostProjection::Missing,
        Err(e) => HostProjection::Error(e.to_string()),
    }
}

/// A small JSON error body (UTAS mutations that cannot be honoured fail loudly —
/// they are NEVER retried against Python, which would risk a double mutation).
fn error_response(status: u16, code: &str) -> WireResponse {
    WireResponse {
        status,
        headers: vec![("Content-Type".to_string(), "application/json".to_string())],
        body: format!("{{\"error\":\"{code}\"}}").into_bytes(),
    }
}

/// `PUT …/squad/<n>` — parse the full replacement, reverse-resolve every wire id,
/// AUTHORIZE every resolved item against the active club, then commit the
/// canonical squad + FIFA extension to Core in one transaction. On any failure
/// it returns an error and NEVER falls back to Python (no double mutation).
pub fn handle_put_squad(body: &[u8], deps: &SquadDeps<'_>) -> (WireResponse, SquadLog) {
    let put = match parse_squad_put(body) {
        Ok(p) => p,
        Err(e) => {
            return (
                error_response(400, "parse_error"),
                SquadLog {
                    outcome: "parse_error",
                    detail: e.to_string(),
                },
            )
        }
    };
    // Reverse-resolve wire→owned and shape canonical + extension. Refuses on an
    // unresolved wire id or the same owned item placed twice.
    let build = match build_squad_write(&put, deps.resolver) {
        Ok(b) => b,
        Err(SquadBuildError::UnresolvedWireIds(ids)) => {
            return (
                error_response(400, "unresolved_wire_ids"),
                SquadLog {
                    outcome: "unresolved_wire_ids",
                    detail: format!("{ids:?}"),
                },
            )
        }
        Err(SquadBuildError::DuplicateOwnedItem(id)) => {
            return (
                error_response(400, "duplicate_owned_item"),
                SquadLog {
                    outcome: "duplicate_owned_item",
                    detail: id,
                },
            )
        }
    };
    // AUTHORIZATION — identity resolution is NOT authorization. Every resolved
    // owned item MUST belong to the active club; a globally-valid wire id that
    // belongs to another profile is rejected BEFORE any canonical mutation.
    let owned_set: std::collections::HashSet<String> = match deps.core.all_owned() {
        Ok(v) => v.into_iter().map(|i| i.owned_card_id).collect(),
        Err(e) => {
            return (
                error_response(502, "core_error"),
                SquadLog {
                    outcome: "core_error",
                    detail: e.to_string(),
                },
            )
        }
    };
    for slot in &build.canonical.slots {
        if !owned_set.contains(&slot.owned_card_id) {
            return (
                error_response(403, "not_owned"),
                SquadLog {
                    outcome: "unauthorized_item",
                    detail: slot.owned_card_id.clone(),
                },
            );
        }
    }
    // Commit canonical + extension atomically. No Python fallback on failure.
    let req = CoreReplaceRequest {
        name: build.canonical.name.clone(),
        formation: build.canonical.formation.clone(),
        slots: build
            .canonical
            .slots
            .iter()
            .map(|s| CoreSquadSlot {
                owned_card_id: s.owned_card_id.clone(),
                index: s.index,
                is_captain: s.is_captain,
                is_on_bench: s.is_on_bench,
            })
            .collect(),
        client_reported: CoreClientEval {
            chemistry: build.extension.client_reported.chemistry,
            rating: build.extension.client_reported.rating,
            star_rating: build.extension.client_reported.star_rating,
        },
        ext_namespace: EXT_NAMESPACE.to_string(),
        ext_schema_version: EXT_SCHEMA_VERSION,
        ext_payload: build.extension.to_payload(),
    };
    match deps.core.replace_squad(&req) {
        Ok(_) => (
            json_response(&save_ack(put.id)),
            SquadLog {
                outcome: "ok",
                detail: String::new(),
            },
        ),
        Err(e) => (
            error_response(502, "core_error"),
            SquadLog {
                outcome: "core_error",
                detail: e.to_string(),
            },
        ),
    }
}

/// `GET …/squad/list` — served from Core + the ONE projector. Stale/Missing are
/// integrity failures for the migrated dev profile: logged prominently, degraded
/// to an empty list, NEVER served from Python and NEVER projected from stale ext.
pub fn handle_squad_list(deps: &SquadDeps<'_>) -> (WireResponse, SquadLog) {
    match project_active_squad(deps) {
        HostProjection::Squad(v) => (
            json_response(&squad_list(&v)),
            SquadLog {
                outcome: "ok",
                detail: String::new(),
            },
        ),
        HostProjection::Stale => (
            json_response(&json!({ "squad": [] })),
            SquadLog {
                outcome: "stale_integrity",
                detail: "stale extension not applied".into(),
            },
        ),
        HostProjection::Missing => (
            json_response(&json!({ "squad": [] })),
            SquadLog {
                outcome: "missing_integrity",
                detail: "no extension stored".into(),
            },
        ),
        HostProjection::Error(e) => (
            json_response(&json!({ "squad": [] })),
            SquadLog {
                outcome: "core_error",
                detail: e,
            },
        ),
    }
}

/// `GET …/squad/active` — the active squad projected from Core, returned as the
/// top-level squad object (byte-identical to what `userMassInfo.squad` embeds).
/// Never served from Python and NEVER projected from a stale extension; on a
/// stale/missing extension or a Core error it degrades to an honest empty squad
/// (never 401/403, never a Python fallback that could mask split authority).
pub fn handle_squad_active(deps: &SquadDeps<'_>, persona_id: i64) -> (WireResponse, SquadLog) {
    match project_active_squad(deps) {
        HostProjection::Squad(v) => (
            json_response(&user_mass_info_squad(v, persona_id)),
            SquadLog {
                outcome: "ok",
                detail: String::new(),
            },
        ),
        HostProjection::Stale => (
            json_response(&empty_squad_overlay(persona_id)),
            SquadLog {
                outcome: "stale_integrity",
                detail: "stale extension not applied".into(),
            },
        ),
        HostProjection::Missing => (
            json_response(&empty_squad_overlay(persona_id)),
            SquadLog {
                outcome: "missing_integrity",
                detail: "no extension stored".into(),
            },
        ),
        HostProjection::Error(e) => (
            json_response(&empty_squad_overlay(persona_id)),
            SquadLog {
                outcome: "core_error",
                detail: e,
            },
        ),
    }
}

/// An explicit empty active squad used only when Rust squad authority cannot
/// produce a Fresh projection during a userMassInfo overlay. It is NOT Python's
/// squad (that would reintroduce split authority) and NOT fabricated extension
/// state — it is an honest "no squad available", surfaced with an ERROR log.
fn empty_squad_overlay(persona_id: i64) -> Value {
    json!({
        "id": ACTIVE_SQUAD_WIRE_ID,
        "personaId": persona_id,
        "changed": 0,
        "actives": [],
        "players": [],
    })
}

/// Replace `resp`'s body with `body`, fixing framing headers (Content-Length,
/// Content-Type; drops any stale length/transfer-encoding).
fn set_json_body(resp: &mut WireResponse, body: Vec<u8>) {
    resp.headers.retain(|(k, _)| {
        !k.eq_ignore_ascii_case("content-length")
            && !k.eq_ignore_ascii_case("content-type")
            && !k.eq_ignore_ascii_case("transfer-encoding")
    });
    resp.headers
        .push(("Content-Type".to_string(), "application/json".to_string()));
    resp.headers
        .push(("Content-Length".to_string(), body.len().to_string()));
    resp.body = body;
}

/// `GET …/userMassInfo` — proxy the request to Python verbatim, then overlay ONLY
/// `.squad` with the Rust/Core projection. Every unrelated field (`userInfo`,
/// `settings`, `userData`, `pileSizeClientData`, …) is preserved exactly.
pub fn handle_user_mass_info(
    method: &str,
    target: &str,
    headers: &[(String, String)],
    body: &[u8],
    deps: &SquadDeps<'_>,
    pass: &PassClient,
) -> (WireResponse, SquadLog) {
    let mut resp = match pass.forward(method, target, headers, body) {
        Ok(r) => r,
        Err(e) => {
            return (
                error_response(502, "upstream_unavailable"),
                SquadLog {
                    outcome: "python_unreachable",
                    detail: e.to_string(),
                },
            )
        }
    };
    // Only a successful JSON object carrying `.squad` is overlaid; anything else
    // is returned verbatim (we never invent a squad into an unrelated response).
    if !(200..300).contains(&resp.status) {
        return (
            resp,
            SquadLog {
                outcome: "python_non_2xx_passthrough",
                detail: String::new(),
            },
        );
    }
    let mut root: Value = match serde_json::from_slice::<Value>(&resp.body) {
        Ok(v) if v.is_object() => v,
        _ => {
            return (
                resp,
                SquadLog {
                    outcome: "python_body_unusable_passthrough",
                    detail: String::new(),
                },
            )
        }
    };
    if root.get("squad").is_none() {
        return (
            resp,
            SquadLog {
                outcome: "python_no_squad_passthrough",
                detail: String::new(),
            },
        );
    }
    // Preserve the client's persona from Python's own response.
    let persona = root["squad"]
        .get("personaId")
        .and_then(|x| x.as_i64())
        .or_else(|| {
            root.get("userInfo")
                .and_then(|u| u.get("personaId"))
                .and_then(|x| x.as_i64())
        })
        .unwrap_or(0);
    let (squad_val, log) = match project_active_squad(deps) {
        HostProjection::Squad(v) => (
            user_mass_info_squad(v, persona),
            SquadLog {
                outcome: "ok",
                detail: String::new(),
            },
        ),
        HostProjection::Stale => (
            empty_squad_overlay(persona),
            SquadLog {
                outcome: "stale_integrity",
                detail: "stale extension not applied".into(),
            },
        ),
        HostProjection::Missing => (
            empty_squad_overlay(persona),
            SquadLog {
                outcome: "missing_integrity",
                detail: "no extension stored".into(),
            },
        ),
        HostProjection::Error(e) => (
            empty_squad_overlay(persona),
            SquadLog {
                outcome: "core_error",
                detail: e,
            },
        ),
    };
    root["squad"] = squad_val;
    let new_body = serde_json::to_vec(&root).unwrap_or_else(|_| resp.body.clone());
    set_json_body(&mut resp, new_body);
    (resp, log)
}

// ───────────────────────────── HTTP wire types ──────────────────────────────

/// A response ready to write: status, headers, body.
#[derive(Debug, Clone)]
pub struct WireResponse {
    pub status: u16,
    pub headers: Vec<(String, String)>,
    pub body: Vec<u8>,
}

fn json_response(body: &Value) -> WireResponse {
    let bytes = serde_json::to_vec(body).unwrap_or_else(|_| b"{}".to_vec());
    WireResponse {
        status: 200,
        headers: vec![("Content-Type".to_string(), "application/json".to_string())],
        body: bytes,
    }
}

fn is_hop_by_hop(name: &str) -> bool {
    matches!(
        name.to_ascii_lowercase().as_str(),
        "connection"
            | "keep-alive"
            | "transfer-encoding"
            | "content-length"
            | "host"
            | "proxy-connection"
            | "te"
            | "trailer"
            | "upgrade"
    )
}

// ───────────────────────────── Python passthrough ───────────────────────────

/// Verbatim reverse proxy to the Python UTAS oracle. Preserves method, full
/// target (path + query), end-to-end headers, and body; returns the upstream's
/// status/headers/body faithfully.
pub struct PassClient {
    client: reqwest::blocking::Client,
    upstream: String,
}

impl PassClient {
    pub fn new(upstream: impl Into<String>) -> Self {
        PassClient {
            client: reqwest::blocking::Client::new(),
            upstream: upstream.into().trim_end_matches('/').to_string(),
        }
    }

    pub fn forward(
        &self,
        method: &str,
        target: &str,
        headers: &[(String, String)],
        body: &[u8],
    ) -> Result<WireResponse, CoreError> {
        let url = format!("{}{}", self.upstream, target);
        let m = reqwest::Method::from_bytes(method.as_bytes())
            .map_err(|e| CoreError::Http(format!("bad method: {e}")))?;
        let mut req = self.client.request(m, &url);
        for (k, v) in headers {
            if !is_hop_by_hop(k) {
                req = req.header(k, v);
            }
        }
        if !body.is_empty() {
            req = req.body(body.to_vec());
        }
        let resp = req.send().map_err(|e| CoreError::Http(e.to_string()))?;
        let status = resp.status().as_u16();
        let mut out = Vec::new();
        for (k, v) in resp.headers() {
            if !is_hop_by_hop(k.as_str()) {
                if let Ok(s) = v.to_str() {
                    out.push((k.to_string(), s.to_string()));
                }
            }
        }
        let bytes = resp
            .bytes()
            .map_err(|e| CoreError::Http(e.to_string()))?
            .to_vec();
        Ok(WireResponse {
            status,
            headers: out,
            body: bytes,
        })
    }
}

// ───────────────────────────── Server ───────────────────────────────────────

/// The migration host. Cheap to clone (all shared state is `Arc`).
#[derive(Clone)]
pub struct Server {
    core: Arc<dyn CoreAccess>,
    entities: Arc<Fifa17Entities>,
    /// The single production identity resolver, shared by `/club`, `/squad/*`
    /// and the userMassInfo overlay — one wire↔owned identity everywhere.
    resolver: Arc<Fifa17IdentityResolver>,
    pass: Arc<PassClient>,
    /// The launcher-selected FIFA persona id (injected via `OPENFUT_PERSONA_ID`),
    /// used to stamp `personaId` on the Core-backed `GET /squad/active` object.
    /// Never baked in — it must match the persona LSX/Blaze/POW/UTAS agree on.
    persona_id: i64,
    /// Per-login FIFA session/capability authority (empty-My-Packs topology).
    /// The economy (coins, unopened packs, BUY) stays Python's; this owns only
    /// session/capability state — Rust reads Python's live body, never writes it.
    sessions: Arc<Mutex<SessionStore>>,
    /// Monotonic clock origin for the session/pending TTLs.
    start: Instant,
}

impl Server {
    /// Assemble from injected parts (used by `from_config` and tests).
    pub fn new(
        core: Arc<dyn CoreAccess>,
        entities: Arc<Fifa17Entities>,
        resolver: Arc<Fifa17IdentityResolver>,
        pass: Arc<PassClient>,
        persona_id: i64,
    ) -> Self {
        Server {
            core,
            entities,
            resolver,
            pass,
            persona_id,
            sessions: Arc::new(Mutex::new(SessionStore::new())),
            start: Instant::now(),
        }
    }

    /// Build from config: load entity tables + the FIFA 17 identity catalog, open
    /// the persistent identity store, and wire the Core client + Python
    /// passthrough. Fails clearly if a required production identity source cannot
    /// be loaded — there is no placeholder fallback.
    pub fn from_config(cfg: &HostConfig) -> Result<Self, String> {
        let entities = Fifa17Entities::from_tables_dir(std::path::Path::new(&cfg.tables_dir))
            .map_err(|e| format!("loading entity tables from {}: {e}", cfg.tables_dir))?;
        let catalog = Fifa17CardCatalog::from_file(std::path::Path::new(&cfg.catalog_path))
            .map_err(|e| format!("loading card-definition catalog {}: {e}", cfg.catalog_path))?;
        let store = openfut_identity::JsonIdentityStore::open(&cfg.identity_store_path)
            .map_err(|e| format!("opening identity store {}: {e}", cfg.identity_store_path))?;
        let resolver = Arc::new(Fifa17IdentityResolver::new(catalog, Arc::new(store)));
        Ok(Server {
            core: Arc::new(HttpCoreClient::new(
                cfg.core_url.clone(),
                Fifa17WireItemIdPolicy::GAME,
            )),
            entities: Arc::new(entities),
            resolver,
            pass: Arc::new(PassClient::new(cfg.python_upstream.clone())),
            persona_id: cfg.persona_id,
            sessions: Arc::new(Mutex::new(SessionStore::new())),
            start: Instant::now(),
        })
    }

    /// Assemble the shared squad dependencies (Core access + the one production
    /// resolver + entity tables).
    fn squad_deps(&self) -> SquadDeps<'_> {
        SquadDeps {
            core: self.core.as_ref(),
            resolver: self.resolver.as_ref(),
            entities: self.entities.as_ref(),
        }
    }

    /// 4-arg entrypoint (tests + callers without a peer address). Session-bound
    /// routes fall back to a `None` client IP.
    pub fn handle(
        &self,
        method: &str,
        target: &str,
        headers: &[(String, String)],
        body: &[u8],
    ) -> WireResponse {
        self.handle_with_ip(method, target, headers, body, None)
    }

    /// Route one request to a response. Classification happens here, once, before
    /// either branch runs. `client_ip` is the peer address used to bind FIFA
    /// session capability (auxiliary to the authoritative X-UT-SID).
    pub fn handle_with_ip(
        &self,
        method: &str,
        target: &str,
        headers: &[(String, String)],
        body: &[u8],
        client_ip: Option<&str>,
    ) -> WireResponse {
        let path = target.split('?').next().unwrap_or(target);
        match classify(method, path) {
            Route::Club => {
                let query = target.split_once('?').map(|(_, q)| q).unwrap_or("");
                let deps = ClubDeps {
                    core: self.core.as_ref(),
                    entities: self.entities.as_ref(),
                    assets: self.resolver.as_ref(),
                };
                let (resp, log) = handle_club(query, &deps);
                eprintln!(
                    "utas-host owner=RUST route=club status={} outcome={} filter=[{}] total={} emitted={} dropped_no_asset={} offset={:?} limit={:?}",
                    resp.status, log.outcome, log.filter, log.total, log.emitted, log.dropped_no_asset, log.offset, log.limit
                );
                resp
            }
            Route::SquadReplace => {
                let deps = self.squad_deps();
                let (resp, log) = handle_put_squad(body, &deps);
                eprintln!(
                    "utas-host owner=RUST route=squad-replace status={} outcome={} detail=[{}]",
                    resp.status, log.outcome, log.detail
                );
                resp
            }
            Route::SquadList => {
                let deps = self.squad_deps();
                let (resp, log) = handle_squad_list(&deps);
                eprintln!(
                    "utas-host owner=RUST route=squad-list status={} outcome={} detail=[{}]",
                    resp.status, log.outcome, log.detail
                );
                resp
            }
            Route::SquadActive => {
                let deps = self.squad_deps();
                let (resp, log) = handle_squad_active(&deps, self.persona_id);
                eprintln!(
                    "utas-host owner=RUST route=squad-active status={} outcome={} detail=[{}]",
                    resp.status, log.outcome, log.detail
                );
                resp
            }
            Route::UserMassInfo => {
                let deps = self.squad_deps();
                let (resp, log) =
                    handle_user_mass_info(method, target, headers, body, &deps, self.pass.as_ref());
                eprintln!(
                    "utas-host owner=RUST_OVERLAY route=userMassInfo status={} squad_outcome={} detail=[{}]",
                    resp.status, log.outcome, log.detail
                );
                resp
            }
            Route::Auth => self.handle_auth(method, target, headers, body, client_ip),
            Route::Capability => self.handle_capability(body, client_ip),
            Route::StorePurchaseGroup => {
                self.handle_store_purchasegroup(method, target, headers, body, client_ip)
            }
            Route::Passthrough => {
                let resp = match self.pass.forward(method, target, headers, body) {
                    Ok(r) => r,
                    Err(e) => {
                        eprintln!("utas-host ERROR passthrough to Python failed: {e}");
                        WireResponse {
                            status: 502,
                            headers: vec![(
                                "Content-Type".to_string(),
                                "application/json".to_string(),
                            )],
                            body: br#"{"error":"upstream unavailable"}"#.to_vec(),
                        }
                    }
                };
                eprintln!(
                    "utas-host owner=PYTHON_FALLBACK method={} path={} status={}",
                    method, path, resp.status
                );
                resp
            }
        }
    }

    /// Monotonic seconds since server start — the clock for session/pending TTLs.
    fn now(&self) -> f64 {
        self.start.elapsed().as_secs_f64()
    }

    /// `POST /ut/auth` — proxy to Python (which mints the X-UT-SID, adopts the
    /// persona and refreshes its save), then OBSERVE the returned SID to open a
    /// Rust session bound to the peer IP + configured persona. Account/economy
    /// stays Python-authoritative; Rust only tracks the session. Python's response
    /// is returned byte-for-byte.
    fn handle_auth(
        &self,
        method: &str,
        target: &str,
        headers: &[(String, String)],
        body: &[u8],
        client_ip: Option<&str>,
    ) -> WireResponse {
        let resp = match self.pass.forward(method, target, headers, body) {
            Ok(r) => r,
            Err(e) => {
                eprintln!("utas-host ERROR auth proxy to Python failed: {e}");
                return error_response(502, "upstream_unavailable");
            }
        };
        let mut outcome = "no_sid";
        if (200..300).contains(&resp.status) {
            if let Some(sid) = observe_sid(&resp.body) {
                self.sessions.lock().unwrap().open_session(
                    &sid,
                    client_ip.map(|s| s.to_string()),
                    self.persona_id,
                    self.now(),
                );
                outcome = "session_opened";
            }
        }
        eprintln!(
            "utas-host owner=RUST_OBSERVE route=auth status={} ip={:?} outcome={}",
            resp.status, client_ip, outcome
        );
        resp
    }

    /// `POST /openfut/fifa17/capability` — Rust-owned launcher registration (no
    /// economy, no proxy). Fail-closed: an unsupported name/version is a 400 that
    /// records nothing, so the session stays on the sentinel fallback.
    fn handle_capability(&self, body: &[u8], client_ip: Option<&str>) -> WireResponse {
        let req = match parse_capability_request(body) {
            Ok(r) => r,
            Err(()) => {
                eprintln!("utas-host owner=RUST route=capability status=400 outcome=unsupported");
                return json_status(400, &json!({"error": "unsupported capability"}));
            }
        };
        let persona = req.persona_id.unwrap_or(self.persona_id);
        let outcome = self.sessions.lock().unwrap().register_capability(
            client_ip.map(|s| s.to_string()),
            persona,
            req.version,
            self.now(),
        );
        eprintln!(
            "utas-host owner=RUST route=capability status=200 ip={:?} persona={} -> {:?}",
            client_ip, persona, outcome
        );
        json_status(200, &json!({"status": "OK"}))
    }

    /// `GET …/store/purchasegroup…` — proxy to Python for the authoritative economy
    /// body (catalogue + owned packs + coins), then overlay ONLY the empty-My-Packs
    /// topology from the Rust session mode. Python (which does not know the Rust
    /// capability) always emits the 65534 sentinel when My Packs is empty; for a
    /// verified clean-v1 SID we strip it. Rust never writes economy state.
    fn handle_store_purchasegroup(
        &self,
        method: &str,
        target: &str,
        headers: &[(String, String)],
        body: &[u8],
        client_ip: Option<&str>,
    ) -> WireResponse {
        let mut resp = match self.pass.forward(method, target, headers, body) {
            Ok(r) => r,
            Err(e) => {
                eprintln!("utas-host ERROR purchasegroup proxy to Python failed: {e}");
                return error_response(502, "upstream_unavailable");
            }
        };
        let sid = header(headers, "x-ut-sid").unwrap_or("");
        // Freeze the session's empty-My-Packs mode at this first store request.
        let mode = self
            .sessions
            .lock()
            .unwrap()
            .empty_mypacks_mode(sid, client_ip, self.now());
        let mut stripped = 0usize;
        if (200..300).contains(&resp.status) {
            if let Ok(mut root) = serde_json::from_slice::<Value>(&resp.body) {
                stripped = overlay_empty_mypacks(&mut root, mode);
                if stripped > 0 {
                    if let Ok(new_body) = serde_json::to_vec(&root) {
                        set_json_body(&mut resp, new_body);
                    }
                }
            }
        }
        eprintln!(
            "utas-host owner=RUST_OVERLAY route=purchasegroup status={} sid={} mode={} sentinel_stripped={}",
            resp.status, fifa17_sidlog(sid), mode.as_str(), stripped
        );
        resp
    }

    /// Serve forever on `addr` (thread-per-connection, HTTP/1.1 keep-alive).
    pub fn serve(&self, addr: &str) -> std::io::Result<()> {
        let listener = TcpListener::bind(addr)?;
        eprintln!("utas-host listening on {addr}");
        self.serve_listener(listener);
        Ok(())
    }

    /// Accept loop on an already-bound listener (lets tests bind an ephemeral
    /// port and learn it before serving).
    pub fn serve_listener(&self, listener: TcpListener) {
        for stream in listener.incoming() {
            let stream = match stream {
                Ok(s) => s,
                Err(_) => continue,
            };
            let server = self.clone();
            std::thread::spawn(move || server.handle_conn(stream));
        }
    }

    fn handle_conn(&self, stream: TcpStream) {
        let mut reader = BufReader::new(match stream.try_clone() {
            Ok(s) => s,
            Err(_) => return,
        });
        let mut writer = stream;
        let peer_ip = writer.peer_addr().ok().map(|a| a.ip().to_string());
        loop {
            match read_request(&mut reader) {
                Ok(Some(req)) => {
                    let resp = self.handle_with_ip(
                        &req.method,
                        &req.target,
                        &req.headers,
                        &req.body,
                        peer_ip.as_deref(),
                    );
                    if write_response(&mut writer, &resp).is_err() {
                        return;
                    }
                    if req.close {
                        return;
                    }
                }
                Ok(None) => return, // clean EOF
                Err(_) => return,
            }
        }
    }
}

// ─────────────────────── Session/capability route helpers ───────────────────

/// Case-insensitive header lookup.
fn header<'a>(headers: &'a [(String, String)], name: &str) -> Option<&'a str> {
    headers
        .iter()
        .find(|(k, _)| k.eq_ignore_ascii_case(name))
        .map(|(_, v)| v.as_str())
}

/// A short, non-secret tag for correlating a session in logs (last 6 chars).
fn fifa17_sidlog(sid: &str) -> String {
    if sid.is_empty() {
        "-".to_string()
    } else {
        format!("\u{2026}{}", &sid[sid.len().saturating_sub(6)..])
    }
}

/// A JSON response with an explicit status.
fn json_status(status: u16, v: &Value) -> WireResponse {
    let body = serde_json::to_vec(v).unwrap_or_default();
    WireResponse {
        status,
        headers: vec![
            ("Content-Type".to_string(), "application/json".to_string()),
            ("Content-Length".to_string(), body.len().to_string()),
        ],
        body,
    }
}

/// Observe the minted `sid` from Python's `/ut/auth` JSON response body.
fn observe_sid(body: &[u8]) -> Option<String> {
    serde_json::from_slice::<Value>(body)
        .ok()?
        .get("sid")?
        .as_str()
        .map(str::to_string)
}

/// A validated capability registration request.
struct CapabilityRequest {
    persona_id: Option<i64>,
    version: u32,
}

/// Parse + validate a capability POST body. Mirrors the Python route: anything but
/// `capability == "empty_mypacks_resolver"` at the supported version is rejected.
fn parse_capability_request(body: &[u8]) -> Result<CapabilityRequest, ()> {
    let v: Value = serde_json::from_slice(body).map_err(|_| ())?;
    let obj = v.as_object().ok_or(())?;
    let name = obj.get("capability").and_then(|x| x.as_str()).ok_or(())?;
    let version = obj
        .get("version")
        .and_then(|x| x.as_u64())
        .and_then(|n| u32::try_from(n).ok())
        .ok_or(())?;
    validate_capability(name, version).map_err(|_| ())?;
    let persona_id = obj.get("personaId").and_then(|x| x.as_i64());
    Ok(CapabilityRequest {
        persona_id,
        version,
    })
}

/// Overlay the empty-My-Packs topology on Python's `purchasegroup` body. Python
/// always emits the synthetic 65534 sentinel when My Packs is empty (it does not
/// know the Rust capability); for a verified clean-v1 session we remove it so the
/// client's resolver guard routes to Browse. Sentinel mode leaves it in place, and
/// real owned packs (no 65534) are untouched in either mode. Returns the count
/// removed. Pure — the topology decision is unit-testable.
fn overlay_empty_mypacks(root: &mut Value, mode: StoreMode) -> usize {
    if mode != StoreMode::CleanV1 {
        return 0;
    }
    let Some(arr) = root.get_mut("purchase").and_then(|p| p.as_array_mut()) else {
        return 0;
    };
    let before = arr.len();
    arr.retain(|entry| {
        entry
            .get("id")
            .and_then(|x| x.as_u64())
            .map(|id| id != SENTINEL_PACK_ID)
            .unwrap_or(true)
    });
    before - arr.len()
}

// ───────────────────────────── HTTP/1.1 request reader ──────────────────────

/// A parsed request. `target` is the raw request target (path + optional query).
pub struct ParsedRequest {
    pub method: String,
    pub target: String,
    pub headers: Vec<(String, String)>,
    pub body: Vec<u8>,
    pub close: bool,
}

/// Read one HTTP/1.1 request. `Ok(None)` = clean connection close before a
/// request line. Body is read exactly per `Content-Length` (chunked is not used
/// by this client population — worker D).
pub fn read_request<R: BufRead>(reader: &mut R) -> std::io::Result<Option<ParsedRequest>> {
    let mut line = String::new();
    let n = reader.read_line(&mut line)?;
    if n == 0 {
        return Ok(None);
    }
    let request_line = line.trim_end();
    if request_line.is_empty() {
        // Tolerate a stray blank line before the request line.
        return read_request(reader);
    }
    let mut parts = request_line.split_whitespace();
    let method = parts.next().unwrap_or("").to_string();
    let target = parts.next().unwrap_or("").to_string();

    let mut headers = Vec::new();
    let mut content_length = 0usize;
    let mut close = false;
    loop {
        let mut h = String::new();
        if reader.read_line(&mut h)? == 0 {
            break;
        }
        let h = h.trim_end();
        if h.is_empty() {
            break;
        }
        if let Some((k, v)) = h.split_once(':') {
            let k = k.trim().to_string();
            let v = v.trim().to_string();
            if k.eq_ignore_ascii_case("content-length") {
                content_length = v.parse().unwrap_or(0);
            } else if k.eq_ignore_ascii_case("connection") && v.eq_ignore_ascii_case("close") {
                close = true;
            }
            headers.push((k, v));
        }
    }

    let mut body = vec![0u8; content_length];
    if content_length > 0 {
        reader.read_exact(&mut body)?;
    }

    Ok(Some(ParsedRequest {
        method,
        target,
        headers,
        body,
        close,
    }))
}

fn reason(status: u16) -> &'static str {
    match status {
        200 => "OK",
        204 => "No Content",
        400 => "Bad Request",
        404 => "Not Found",
        500 => "Internal Server Error",
        502 => "Bad Gateway",
        _ => "OK",
    }
}

fn write_response<W: Write>(w: &mut W, resp: &WireResponse) -> std::io::Result<()> {
    let mut head = format!("HTTP/1.1 {} {}\r\n", resp.status, reason(resp.status));
    for (k, v) in &resp.headers {
        if is_hop_by_hop(k) {
            continue;
        }
        head.push_str(&format!("{k}: {v}\r\n"));
    }
    head.push_str(&format!("Content-Length: {}\r\n", resp.body.len()));
    head.push_str("\r\n");
    w.write_all(head.as_bytes())?;
    w.write_all(&resp.body)?;
    w.flush()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn special_filter_keeps_only_specials_and_paginates_filtered_set() {
        let mk = |id: i64, rf: i64| serde_json::json!({ "id": id, "rareflag": rf });
        // base rare(1)/common(0) interleaved with specials(3,11,24).
        let items = vec![mk(1, 1), mk(2, 3), mk(3, 1), mk(4, 24), mk(5, 0), mk(6, 11)];
        let (all, total) = special_filter_page(&items, None, None);
        assert_eq!(total, 3, "only rareflag>1 counted");
        assert!(all.iter().all(|it| it["rareflag"].as_i64().unwrap() > 1));
        assert_eq!(
            all.iter()
                .map(|it| it["id"].as_i64().unwrap())
                .collect::<Vec<_>>(),
            vec![2, 4, 6],
            "base rare/common excluded, order preserved"
        );
        // pagination is over the FILTERED set, no overlap, no base leakage.
        let (p0, t0) = special_filter_page(&items, Some(0), Some(2));
        let (p1, t1) = special_filter_page(&items, Some(2), Some(2));
        assert_eq!((t0, t1), (3, 3));
        assert_eq!(
            p0.iter()
                .map(|it| it["id"].as_i64().unwrap())
                .collect::<Vec<_>>(),
            vec![2, 4]
        );
        assert_eq!(
            p1.iter()
                .map(|it| it["id"].as_i64().unwrap())
                .collect::<Vec<_>>(),
            vec![6]
        );
    }

    #[test]
    fn classify_club_only_on_exact_get() {
        assert_eq!(classify("GET", "/ut/game/fifa17/club"), Route::Club);
        assert_eq!(classify("get", "/ut/game/fifa18/club"), Route::Club);
        // method must be GET
        assert_eq!(classify("PUT", "/ut/game/fifa17/club"), Route::Passthrough);
        // near-misses stay on Python
        assert_eq!(
            classify("GET", "/ut/game/fifa17/club/stats/staff"),
            Route::Passthrough
        );
        assert_eq!(
            classify("GET", "/ut/game/fifa17/clubUser"),
            Route::Passthrough
        );
        assert_eq!(
            classify("GET", "/ut/game/fifa17/tradePile"),
            Route::Passthrough
        );
        assert_eq!(
            classify("POST", "/ut/game/fifa17/purchased/items"),
            Route::Passthrough
        );
        assert_eq!(classify("GET", "/ut/game//club"), Route::Passthrough);
        assert_eq!(classify("GET", "/club"), Route::Passthrough);
    }

    #[test]
    fn parse_core_page_reads_collection_and_total() {
        let v = json!({
            "collection": [{
                "owned_card_id": "oc1",
                "effective_overall": 86,
                "effective_position": "CDM",
                "card": {"id":"card_ch_1","overall":85,"position":"CDM","nation":"Argentina","league":"Premier League","club":"Chelsea","pace":80,"shooting":70,"passing":75,"dribbling":78,"defending":84,"physical":82}
            }],
            "total": 42
        });
        let page = parse_core_page(&v).unwrap();
        assert_eq!(page.total, 42);
        assert_eq!(page.items.len(), 1);
        let it = &page.items[0];
        assert_eq!(it.owned_card_id, "oc1");
        assert_eq!(it.card_id, "card_ch_1");
        assert_eq!(it.rating, 86, "effective_overall wins over base");
        assert_eq!(it.position, "CDM");
        assert_eq!(it.attributes, [80, 70, 75, 78, 84, 82]);
    }

    // ── Fifa17IdentityResolver: the single production identity path ──────────

    fn test_catalog(cards: &[(&str, u32)]) -> Fifa17CardCatalog {
        let entries: Vec<String> = cards
            .iter()
            .map(|(id, asset)| format!("\"{id}\":{{\"asset_id\":{asset}}}"))
            .collect();
        let doc = format!(
            "{{\"schema_version\":1,\"game\":\"fifa17\",\"cards\":{{{}}}}}",
            entries.join(",")
        );
        Fifa17CardCatalog::from_json_str(&doc).unwrap()
    }

    fn temp_store_path(tag: &str) -> std::path::PathBuf {
        use std::sync::atomic::{AtomicU64, Ordering};
        static N: AtomicU64 = AtomicU64::new(0);
        let n = N.fetch_add(1, Ordering::SeqCst);
        std::env::temp_dir().join(format!(
            "ofut-resolver-{tag}-{}-{n}.json",
            std::process::id()
        ))
    }

    fn owned(owned_id: &str, card: &str) -> CoreOwnedItem {
        CoreOwnedItem {
            owned_card_id: owned_id.into(),
            card_id: card.into(),
            rating: 90,
            position: "ST".into(),
            nation: "Argentina".into(),
            league: "Premier League".into(),
            club: "Chelsea".into(),
            attributes: [90, 90, 80, 91, 33, 80],
        }
    }

    fn resolver(path: &std::path::Path, cards: &[(&str, u32)]) -> Fifa17IdentityResolver {
        let store = openfut_identity::JsonIdentityStore::open(path).unwrap();
        Fifa17IdentityResolver::new(test_catalog(cards), Arc::new(store))
    }

    #[test]
    fn resolver_maps_definition_and_allocates_wire_id() {
        let p = temp_store_path("map");
        let r = resolver(&p, &[("card_gold_001", 20801)]);
        let id = r.resolve(&owned("oc1", "card_gold_001")).unwrap();
        assert_eq!(id.asset_id, 20801, "real asset from the catalog");
        assert_eq!(
            id.item_id, 100_000_001,
            "first wire id from the policy floor"
        );
        let _ = std::fs::remove_file(&p);
    }

    #[test]
    fn resolver_drops_unmapped_definition_never_faking() {
        let p = temp_store_path("drop");
        let r = resolver(&p, &[("card_gold_001", 20801)]);
        assert!(
            r.resolve(&owned("oc1", "card_unknown")).is_none(),
            "no catalog entry => dropped, never a fabricated id"
        );
        let _ = std::fs::remove_file(&p);
    }

    #[test]
    fn two_copies_of_a_definition_share_resource_but_get_distinct_wire_ids() {
        let p = temp_store_path("copies");
        let r = resolver(&p, &[("card_gold_001", 20801)]);
        let a = r.resolve(&owned("oc1", "card_gold_001")).unwrap();
        let b = r.resolve(&owned("oc2", "card_gold_001")).unwrap();
        assert_eq!(a.asset_id, b.asset_id, "same definition => same resourceId");
        assert_ne!(a.item_id, b.item_id, "distinct copies => distinct wire ids");
        // Idempotent: the same owned instance re-resolves to the same wire id.
        assert_eq!(
            r.resolve(&owned("oc1", "card_gold_001")).unwrap().item_id,
            a.item_id
        );
        let _ = std::fs::remove_file(&p);
    }

    #[test]
    fn wire_id_survives_restart_and_reverses_exactly() {
        let p = temp_store_path("restart");
        let first = {
            let r = resolver(&p, &[("card_gold_001", 20801)]);
            r.resolve(&owned("oc-stable", "card_gold_001"))
                .unwrap()
                .item_id
        };
        // Reopen the same store file (simulating a host restart).
        let r2 = resolver(&p, &[("card_gold_001", 20801)]);
        let again = r2
            .resolve(&owned("oc-stable", "card_gold_001"))
            .unwrap()
            .item_id;
        assert_eq!(
            first, again,
            "same owned instance keeps its wire id across restart"
        );
        assert_eq!(
            r2.owned_id_for_wire(again as i64).as_deref(),
            Some("oc-stable"),
            "reverse lookup returns the exact owned instance"
        );
        assert_eq!(
            r2.owned_id_for_wire(999_999_999),
            None,
            "unknown wire id => None"
        );
        let _ = std::fs::remove_file(&p);
    }
    // ── Session/capability vertical (this slice) ────────────────────────────

    #[test]
    fn classify_routes_session_vertical() {
        assert_eq!(classify("POST", "/ut/auth"), Route::Auth);
        assert_eq!(classify("POST", "/ut/auth/"), Route::Auth);
        // Only POST is auth; a GET falls through to Python.
        assert_eq!(classify("GET", "/ut/auth"), Route::Passthrough);
        assert_eq!(
            classify("POST", "/openfut/fifa17/capability"),
            Route::Capability
        );
        assert_eq!(
            classify("GET", "/ut/game/fifa17/store/purchasegroup/all"),
            Route::StorePurchaseGroup
        );
        assert_eq!(
            classify("GET", "/ut/game/fifa17/store/purchasegroup/all?ppInfo=true"),
            Route::StorePurchaseGroup
        );
        // A store MUTATION stays on Python (never Rust): economy is not ours.
        assert_eq!(
            classify("PUT", "/ut/game/fifa17/store/transaction/0"),
            Route::Passthrough
        );
        assert_eq!(
            classify("POST", "/openfut/account/sync"),
            Route::Passthrough
        );
    }

    #[test]
    fn observe_sid_extracts_minted_sid() {
        let body = br#"{"protocol":1,"sid":"OPENFUT-SID-42C15A6F78DC6E74","serverTime":"x"}"#;
        assert_eq!(
            observe_sid(body).as_deref(),
            Some("OPENFUT-SID-42C15A6F78DC6E74")
        );
        assert_eq!(observe_sid(b"{}"), None);
        assert_eq!(observe_sid(b"not json"), None);
    }

    #[test]
    fn parse_capability_request_validates() {
        let ok = br#"{"capability":"empty_mypacks_resolver","version":1,"personaId":33068179,"fifaPid":42}"#;
        let r = parse_capability_request(ok).expect("valid");
        assert_eq!(r.version, 1);
        assert_eq!(r.persona_id, Some(33068179));
        assert!(parse_capability_request(
            br#"{"capability":"empty_mypacks_resolver","version":2}"#
        )
        .is_err());
        assert!(parse_capability_request(br#"{"capability":"other","version":1}"#).is_err());
        assert!(parse_capability_request(b"[]").is_err());
        assert!(parse_capability_request(b"nope").is_err());
    }

    #[test]
    fn overlay_strips_sentinel_only_for_clean_v1() {
        let base = serde_json::json!({
            "purchase": [
                {"id": 1, "packType": "BRONZE"},
                {"id": 65534, "packType": "GOLD"},
                {"id": 5, "packType": "GOLD"}
            ]
        });
        // clean-v1: the 65534 sentinel is stripped; real packs remain.
        let mut clean = base.clone();
        assert_eq!(overlay_empty_mypacks(&mut clean, StoreMode::CleanV1), 1);
        let ids: Vec<u64> = clean["purchase"]
            .as_array()
            .unwrap()
            .iter()
            .map(|e| e["id"].as_u64().unwrap())
            .collect();
        assert_eq!(ids, vec![1, 5]);
        // sentinel mode: unchanged (the compatibility shim is kept).
        let mut sent = base.clone();
        assert_eq!(overlay_empty_mypacks(&mut sent, StoreMode::Sentinel), 0);
        assert_eq!(sent["purchase"].as_array().unwrap().len(), 3);
        // no purchase array -> no-op.
        let mut other = serde_json::json!({"other": 1});
        assert_eq!(overlay_empty_mypacks(&mut other, StoreMode::CleanV1), 0);
    }
}