From 97c1dea849ef9757ca8dcb193320d86e0a898847 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 19:54:54 +0000 Subject: [PATCH] Serveur Rust : module grid, phase 2 (calcul de la grille, routes /api/grid et /api/v1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - product_rows.rs : getProductRows complet (7 requêtes FF en parallèle, métriques réseau Qlik, report du dernier snapshot, filtre gamme Y, rattrapage d'un magasin par l'API FF), branché sur rows_cache (calculs en cours partagés, compteur d'enregistrements, compléments par magasin), instantané grid_rows resservi s'il date du mois (rafraîchi au-delà de 20 h), persistance non bloquante ; getStorePatch ; get_product_rows_for_sync pour la synchro nocturne et le préchauffage. - store_patch.rs : buildStorePatchEntries. - enrich.rs (api-enrich) et api_gammes.rs (affecterGammes, lireCorpsJson). - Routes /api/grid/rows (NDJSON), /api/grid/rows/store-patch, /api/v1/grid (diffusion complète sans limit), /products/search, /products/{codein}, /products/{codein}/gamme, /gammes, /fournisseurs, /network/{codeCentrale}. - store.rs : upsert_grid_rows_shared, écriture par lots depuis les lignes partagées du cache (sans copie intégrale). Écart volontaire : après un calcul à la demande, /api/v1 attend la fin de l'écriture de l'instantané avant de le relire (le TS ne l'attendait pas et pouvait répondre à tort « aucun article »). Test d'intégration sur la base ff_fixture (GRID_FF_TEST_DATABASE_URL) : calcul, cache, instantané, complément magasin via une fausse API FF, enrichissement et toutes les routes (session et clé d'API). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LG2LLJuswrrTTPKJmunEVf --- server/src/grid/api_gammes.rs | 143 ++++++ server/src/grid/api_v1_grid.rs | 551 ++++++++++++++++++++ server/src/grid/enrich.rs | 148 ++++++ server/src/grid/grid_rows.rs | 135 +++++ server/src/grid/mod.rs | 32 +- server/src/grid/product_rows.rs | 879 ++++++++++++++++++++++++++++++++ server/src/grid/store.rs | 38 +- server/src/grid/store_patch.rs | 95 +++- server/src/grid/tests_db.rs | 2 + server/src/grid/tests_ff.rs | 302 +++++++++++ 10 files changed, 2306 insertions(+), 19 deletions(-) create mode 100644 server/src/grid/api_gammes.rs create mode 100644 server/src/grid/api_v1_grid.rs create mode 100644 server/src/grid/enrich.rs create mode 100644 server/src/grid/grid_rows.rs create mode 100644 server/src/grid/product_rows.rs create mode 100644 server/src/grid/tests_ff.rs diff --git a/server/src/grid/api_gammes.rs b/server/src/grid/api_gammes.rs new file mode 100644 index 0000000..da7bba7 --- /dev/null +++ b/server/src/grid/api_gammes.rs @@ -0,0 +1,143 @@ +//! CollectFlow — Affectation de gammes via l'API `/api/v1` +//! (port de `src/lib/api-gammes.ts`). +//! +//! L'API n'enregistre pas la gamme : elle la **propose**. La proposition est +//! rangée dans `gammes_a_valider` et apparaît dans la Grille comme une +//! modification non enregistrée — on y voit donc quels produits changent de gamme +//! avant de valider (« Enregistrer ») ou de rejeter (« Annuler les modifications »). +//! Tant qu'elle n'est pas validée, `codeGamme` reste inchangée et la proposition +//! est exposée dans `gammeAValider`. +//! +//! Tout ou rien : si un seul article est inconnu chez le fournisseur, rien n'est +//! proposé et la réponse liste les articles en cause. + +use super::gammes_a_valider::{proposer_gammes, PropositionGamme}; +use super::product_rows::{attendre_persistance, get_product_rows, GetProductRowsInput}; +use super::store::{get_grid_freshness, get_grid_rows_gammes}; +use crate::api_v1_response::{fail, ApiErrorCode}; +use axum::response::Response; +use serde::Serialize; +use serde_json::{json, Value}; + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ResultatGamme { + pub codein: String, + /// Gamme proposée. + pub gamme: String, + /// Gamme enregistrée au moment de l'appel (modifications validées comprises). + pub gamme_precedente: Option, + /// Gamme présente en base FF, non modifiée par l'API. + pub code_gamme_serveur: Option, + /// `true` = la gamme change : proposition déposée, à valider dans la Grille. + /// `false` = l'article a déjà cette gamme (une proposition antérieure en + /// attente pour lui est retirée). + pub modifie: bool, + /// `a_valider` quand `modifie`, sinon `inchangee`. + pub statut: &'static str, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AffectationGammes { + pub code_fournisseur: String, + pub nom_fournisseur: Option, + pub resultats: Vec, +} + +/// Lit un corps JSON, ou renvoie une réponse 400 prête à être retournée. +pub fn lire_corps_json(body: &[u8]) -> Result { + serde_json::from_slice(body) + .map_err(|_| fail(ApiErrorCode::BadRequest, "Corps de requête JSON attendu (Content-Type: application/json).", None)) +} + +/// Propose des gammes pour des articles d'un fournisseur (à valider dans la Grille). +/// +/// Si le fournisseur n'a encore jamais été calculé, son instantané est calculé à la +/// demande (`compute`), comme sur `/grid` : sans lui, impossible de vérifier que les +/// articles lui appartiennent. +/// +/// `auteur` : nom de la clé d'API ou de l'utilisateur — affiché dans la Grille. +pub async fn affecter_gammes( + code_fournisseur: &str, + changes: &[(String, String)], + compute: bool, + auteur: &str, +) -> Result { + // Un même article cité deux fois : la dernière valeur l'emporte (à la place + // de sa première occurrence, comme une Map JS). + let mut demandes: Vec<(String, String)> = Vec::new(); + for (codein, gamme) in changes { + match demandes.iter_mut().find(|(c, _)| c == codein) { + Some(d) => d.1 = gamme.clone(), + None => demandes.push((codein.clone(), gamme.clone())), + } + } + let codeins: Vec = demandes.iter().map(|(c, _)| c.clone()).collect(); + let interne = |e: crate::error::AppError| fail(ApiErrorCode::InternalError, e.message, None); + + if compute && get_grid_freshness(code_fournisseur).await.map_err(interne)?.is_none() { + tracing::info!("[api/v1/gammes] instantané absent pour {code_fournisseur} — calcul à la demande"); + let calcul = get_product_rows(GetProductRowsInput::total(code_fournisseur, false)).await; + attendre_persistance(code_fournisseur).await; + if let Err(e) = calcul { + tracing::error!("[api/v1/gammes] calcul KO pour {code_fournisseur}: {e:#}"); + return Err(fail( + ApiErrorCode::InternalError, + format!("Le calcul de la grille a échoué pour le fournisseur « {code_fournisseur} »."), + None, + )); + } + } + + let actuelles = get_grid_rows_gammes(code_fournisseur, &codeins).await.map_err(interne)?; + let inconnus: Vec<&String> = codeins.iter().filter(|c| !actuelles.contains_key(*c)).collect(); + if !inconnus.is_empty() { + return Err(fail( + ApiErrorCode::NotFound, + format!( + "{} article(s) introuvable(s) chez le fournisseur « {code_fournisseur} » : aucune gamme n'a été enregistrée.", + inconnus.len() + ), + Some(json!({ "fournisseur": code_fournisseur, "inconnus": inconnus })), + )); + } + + let mut resultats = Vec::with_capacity(demandes.len()); + let mut propositions = Vec::new(); + let mut annules = Vec::new(); + let mut nom_fournisseur: Option = None; + + for (codein, gamme) in demandes { + let actuelle = &actuelles[&codein]; + if nom_fournisseur.is_none() { + nom_fournisseur = actuelle.nom_fournisseur.clone(); + } + let modifie = actuelle.code_gamme.as_deref() != Some(gamme.as_str()); + if modifie { + propositions.push(PropositionGamme { + codein: codein.clone(), + gamme: gamme.clone(), + gamme_precedente: actuelle.code_gamme.clone(), + }); + } else { + annules.push(codein.clone()); + } + resultats.push(ResultatGamme { + codein, + gamme, + gamme_precedente: actuelle.code_gamme.clone(), + code_gamme_serveur: actuelle.code_gamme_init.clone(), + modifie, + statut: if modifie { "a_valider" } else { "inchangee" }, + }); + } + + if let Err(e) = proposer_gammes(code_fournisseur, &propositions, &annules, auteur).await { + tracing::error!("[api/v1/gammes] proposition KO pour {code_fournisseur}: {}", e.message); + return Err(fail(ApiErrorCode::InternalError, "L'enregistrement des gammes à valider a échoué.", None)); + } + + tracing::info!("[api/v1/gammes] {code_fournisseur} — {} gamme(s) à valider proposée(s) par {auteur}", propositions.len()); + Ok(AffectationGammes { code_fournisseur: code_fournisseur.to_string(), nom_fournisseur, resultats }) +} diff --git a/server/src/grid/api_v1_grid.rs b/server/src/grid/api_v1_grid.rs new file mode 100644 index 0000000..e5c4590 --- /dev/null +++ b/server/src/grid/api_v1_grid.rs @@ -0,0 +1,551 @@ +//! Routes `/api/v1` adossées au calcul de la grille (phase 2) : `/grid`, +//! `/products/search`, `/products/{codein}`, `/products/{codein}/gamme`, +//! `/gammes`, `/fournisseurs`, `/network/{codeCentrale}`. + +use super::api_gammes::{affecter_gammes, lire_corps_json}; +use super::api_schemas::{ + parse_fournisseurs_query, parse_gammes_batch_body, parse_grid_query, parse_product_detail, parse_product_gamme_body, + parse_product_search, pick_fields, query_map, to_sort_key, GridQueryParams, +}; +use super::enrich::enrich_rows; +use super::product_rows::{attendre_persistance, get_product_rows, GetProductRowsInput}; +use super::store::{ + get_grid_freshness, get_grid_row_by_codein, list_grid_suppliers, list_grid_suppliers_for_codein, query_grid_rows, GridFreshness, + GridQuery, SortOrder, +}; +use super::types::ProductRow; +use crate::api_v1_response::{build_pagination, fail, ok, ApiErrorCode}; +use crate::auth::ApiAuth; +use crate::error::AppError; +use axum::body::{Body, Bytes}; +use axum::extract::{Path, Query}; +use axum::http::header; +use axum::response::{IntoResponse, Response}; +use futures::channel::mpsc; +use futures::SinkExt; +use serde_json::{json, Map, Value}; +use std::collections::HashMap; + +fn interne(e: AppError) -> Response { + fail(ApiErrorCode::InternalError, e.message, None) +} + +fn meta(entrees: Vec<(&str, Value)>) -> Option> { + Some(entrees.into_iter().map(|(k, v)| (k.to_string(), v)).collect()) +} + +/// Lignes prêtes à sérialiser : enrichies (`enrich=1`) puis projetées (`fields`). +async fn lignes_json(rows: Vec, enrichir: bool, fields: Option<&str>) -> Vec { + let valeurs: Vec = if enrichir { + enrich_rows(rows).await.into_iter().map(|r| serde_json::to_value(r).unwrap_or(Value::Null)).collect() + } else { + rows.into_iter().map(|r| serde_json::to_value(r).unwrap_or(Value::Null)).collect() + }; + valeurs.into_iter().map(|v| pick_fields(v, fields)).collect() +} + +fn ordre(order: &str) -> Option { + Some(if order == "asc" { SortOrder::Asc } else { SortOrder::Desc }) +} + +/// `GET /api/v1/grid?fournisseur=…&gamme=&code1..3=&search=&sort=&order=&page=&limit=&fields=&compute=` +/// +/// Lignes de grille d'un fournisseur, filtrées / triées / paginées **en SQL** depuis +/// l'instantané persisté (`grid_rows`). +/// +/// Si le fournisseur n'a pas encore d'instantané, l'endpoint le **calcule à la demande** +/// (`compute=1`, défaut) ; `compute=0` restaure le comportement strict : échec +/// immédiat en `202 not_ready`. +pub async fn get_grid(ApiAuth(auth): ApiAuth, Query(params): Query>) -> Response { + let q = match parse_grid_query(&query_map(params)) { + Ok(q) => q, + Err(issues) => return fail(ApiErrorCode::BadRequest, "Paramètres invalides.", Some(issues.to_json())), + }; + + // Distingue « fournisseur jamais calculé » de « filtres sans résultat » (200 vide). + let mut freshness = match get_grid_freshness(&q.fournisseur).await { + Ok(f) => f, + Err(e) => return interne(e), + }; + let mut computed_on_demand = false; + + if freshness.is_none() { + if q.compute == "0" { + return fail( + ApiErrorCode::NotReady, + format!( + "Aucun instantané pour le fournisseur « {} ». Relancez sans « compute=0 » pour le calculer à la demande.", + q.fournisseur + ), + Some(json!({ "fournisseur": q.fournisseur })), + ); + } + + tracing::info!("[api/v1/grid] instantané absent pour {} — calcul à la demande (via {})", q.fournisseur, auth.via); + // magasin TOTAL : c'est la seule variante persistée par getProductRows, + // et ProductRow embarque déjà les ventilations par magasin. + match get_product_rows(GetProductRowsInput::total(&q.fournisseur, false)).await { + // La persistance est lancée en arrière-plan par le calcul : on attend + // qu'elle soit écrite avant de relire l'instantané. + Ok(_) => attendre_persistance(&q.fournisseur).await, + Err(e) => { + tracing::error!("[api/v1/grid] calcul à la demande KO pour {}: {e:#}", q.fournisseur); + return fail( + ApiErrorCode::InternalError, + format!("Le calcul de la grille a échoué pour le fournisseur « {} ».", q.fournisseur), + Some(json!({ "fournisseur": q.fournisseur })), + ); + } + } + + freshness = match get_grid_freshness(&q.fournisseur).await { + Ok(f) => f, + Err(e) => return interne(e), + }; + if freshness.is_none() { + // Calcul réussi mais aucune ligne : le code fournisseur n'existe pas, ou + // il n'a aucun article. À distinguer d'une panne. + return fail( + ApiErrorCode::NotFound, + format!("Aucun article pour le fournisseur « {} ». Vérifiez le code auprès de /api/v1/fournisseurs.", q.fournisseur), + Some(json!({ "fournisseur": q.fournisseur })), + ); + } + computed_on_demand = true; + } + let freshness = freshness.unwrap_or(GridFreshness { row_count: 0, computed_at: None }); + + let filtres = GridQuery { + code_fournisseur: Some(q.fournisseur.clone()), + gamme: q.gamme.clone(), + code1: q.code1.clone(), + code2: q.code2.clone(), + code3: q.code3.clone(), + nomenclature: q.nomenclature.clone(), + search: q.search.clone(), + sort: to_sort_key(q.sort.sort.as_deref()), + order: ordre(&q.sort.order), + page: None, + limit: None, + }; + + // Sans `limit`, la réponse peut porter tout un fournisseur — et certains en + // comptent plus de 130 000. On diffuse par lots, à mémoire constante. + let Some(limit) = q.limit else { + return stream_all_rows(filtres, q, freshness, computed_on_demand); + }; + + let result = match query_grid_rows(&GridQuery { page: Some(q.page), limit: Some(limit), ..filtres }).await { + Ok(r) => r, + Err(e) => return interne(e), + }; + let enrichi = q.sort.enrich == "1"; + let nb = result.rows.len(); + let rows = lignes_json(result.rows, enrichi, q.sort.fields.as_deref()).await; + let pagination = build_pagination(q.page, limit, result.total); + let has_more = pagination.has_more; + + let mut m = meta(vec![ + ("fournisseur", json!(q.fournisseur)), + ("computedAt", json!(result.computed_at)), + ("snapshotComputedAt", json!(freshness.computed_at)), + ("snapshotRowCount", json!(freshness.row_count)), + ("enrichi", json!(enrichi)), + // true = l'instantané n'existait pas et vient d'être calculé par cet appel. + ("computedOnDemand", json!(computed_on_demand)), + // true = cette réponse contient **toutes** les lignes du fournisseur. + ("complet", json!(!has_more)), + ]) + .unwrap_or_default(); + if has_more { + m.insert( + "avertissement".into(), + json!(format!( + "Réponse partielle : {nb} lignes sur {}, parce que « limit » a été précisé. Retirez « limit » pour obtenir tout le fournisseur en un appel, ou passez à page={}.", + result.total, + q.page + 1 + )), + ); + } + ok(rows, Some(pagination), Some(m)) +} + +/// Lots internes : compromis entre nombre d'allers-retours SQL et mémoire retenue. +const STREAM_BATCH: i64 = 1000; + +/// Diffuse **toutes** les lignes du fournisseur en une seule réponse JSON, sans +/// jamais la construire entièrement en mémoire. +/// +/// La forme reste celle des autres réponses (`{ data, pagination, meta }`) ; +/// `pagination` et `meta` sont écrits en dernier, une fois le total connu. Une +/// erreur survenant après les premiers octets ne peut plus changer le code HTTP : +/// on ferme alors le document avec `meta.erreur`. +fn stream_all_rows(filtres: GridQuery, q: GridQueryParams, freshness: GridFreshness, computed_on_demand: bool) -> Response { + let (mut tx, rx) = mpsc::channel::>(4); + tokio::spawn(async move { + let mut envoyees: i64 = 0; + let mut total: i64 = 0; + let mut computed_at: Option = None; + let mut erreur: Option = None; + let enrichi = q.sort.enrich == "1"; + + if tx.send(Ok(Bytes::from_static(b"{\"data\":["))).await.is_err() { + return; + } + let mut page = 1; + loop { + let lot = match query_grid_rows(&GridQuery { page: Some(page), limit: Some(STREAM_BATCH), ..filtres.clone() }).await { + Ok(l) => l, + Err(e) => { + erreur = Some(e.message); + break; + } + }; + if page == 1 { + total = lot.total; + computed_at = lot.computed_at.clone(); + } + let n = lot.rows.len() as i64; + if n == 0 { + break; + } + let mut morceau = Vec::new(); + for r in lignes_json(lot.rows, enrichi, q.sort.fields.as_deref()).await { + if envoyees > 0 { + morceau.push(b','); + } + morceau.extend(serde_json::to_vec(&r).unwrap_or_default()); + envoyees += 1; + } + if tx.send(Ok(Bytes::from(morceau))).await.is_err() { + return; + } + if n < STREAM_BATCH { + break; + } + page += 1; + } + if let Some(e) = &erreur { + tracing::error!("[api/v1/grid] diffusion interrompue pour {} après {envoyees} lignes: {e}", q.fournisseur); + } + + let complet = erreur.is_none() && envoyees == total; + let mut meta = json!({ + "fournisseur": q.fournisseur, + "computedAt": computed_at, + "snapshotComputedAt": freshness.computed_at, + "snapshotRowCount": freshness.row_count, + "enrichi": enrichi, + "computedOnDemand": computed_on_demand, + "diffuse": true, + "complet": complet, + }); + if let Some(e) = erreur { + meta["erreur"] = json!(format!("Diffusion interrompue après {envoyees} lignes sur {total} : {e}")); + } + let fin = json!({ + "pagination": { "page": 1, "limit": envoyees, "total": total, "totalPages": 1, "hasMore": !complet }, + "meta": meta, + }); + let texte = serde_json::to_string(&fin).unwrap_or_default(); + let _ = tx.send(Ok(Bytes::from(format!("],{}", &texte[1..])))).await; + }); + + ( + [ + (header::CONTENT_TYPE, "application/json; charset=utf-8"), + (header::CACHE_CONTROL, "no-store"), + // Empêche un proxy de tamponner la réponse entière avant de la relayer. + (header::HeaderName::from_static("x-accel-buffering"), "no"), + ], + Body::from_stream(rx), + ) + .into_response() +} + +/// `GET /api/v1/products/search?q=…&fournisseur=&gamme=&sort=&order=&page=&limit=&fields=` +/// +/// **Recherche transversale** sur libellé, codein, GTIN, référence et code centrale — +/// tous fournisseurs confondus. Lit uniquement l'instantané `grid_rows`. +pub async fn get_products_search(ApiAuth(_auth): ApiAuth, Query(params): Query>) -> Response { + let q = match parse_product_search(&query_map(params)) { + Ok(q) => q, + Err(issues) => return fail(ApiErrorCode::BadRequest, "Paramètres invalides.", Some(issues.to_json())), + }; + let result = match query_grid_rows(&GridQuery { + code_fournisseur: q.fournisseur.clone(), + gamme: q.gamme.clone(), + search: Some(q.q.clone()), + sort: to_sort_key(q.sort.sort.as_deref()), + order: ordre(&q.sort.order), + page: Some(q.page), + limit: Some(q.limit), + ..Default::default() + }) + .await + { + Ok(r) => r, + Err(e) => return interne(e), + }; + let enrichi = q.sort.enrich == "1"; + let total = result.total; + let computed_at = result.computed_at.clone(); + let rows = lignes_json(result.rows, enrichi, q.sort.fields.as_deref()).await; + ok( + rows, + Some(build_pagination(q.page, q.limit, total)), + meta(vec![ + ("query", json!(q.q)), + ("computedAt", json!(computed_at)), + ("scope", json!(q.fournisseur.as_ref().map(|f| format!("fournisseur:{f}")).unwrap_or_else(|| "tous fournisseurs".into()))), + ("enrichi", json!(enrichi)), + ]), + ) +} + +/// `GET /api/v1/products/{codein}?fournisseur=…` +/// +/// Fiche complète d'un produit, enrichie des métriques réseau Qlik lues **en cache**. +/// `fournisseur` lève l'ambiguïté d'un article référencé chez plusieurs fournisseurs ; +/// sans lui, l'instantané le plus récent est renvoyé. +pub async fn get_product(ApiAuth(_auth): ApiAuth, Path(codein): Path, Query(params): Query>) -> Response { + if codein.is_empty() { + return fail(ApiErrorCode::BadRequest, "Paramètre 'codein' requis.", None); + } + let parsed = match parse_product_detail(&query_map(params)) { + Ok(p) => p, + Err(issues) => return fail(ApiErrorCode::BadRequest, "Paramètres invalides.", Some(issues.to_json())), + }; + + let mut found = match get_grid_row_by_codein(&codein, parsed.fournisseur.as_deref()).await { + Ok(f) => f, + Err(e) => return interne(e), + }; + + // Absent de l'instantané : on peut le calculer, mais seulement si l'appelant a + // précisé le fournisseur — le calcul se fait par lot fournisseur, pas par article. + if found.is_none() && parsed.compute == "1" { + if let Some(f) = parsed.fournisseur.as_deref() { + tracing::info!("[api/v1/products] {codein} absent — calcul du lot {f} à la demande"); + let calcul = async { + get_product_rows(GetProductRowsInput::total(f, false)).await?; + attendre_persistance(f).await; + get_grid_row_by_codein(&codein, Some(f)).await.map_err(|e| anyhow::anyhow!(e.message)) + }; + match calcul.await { + Ok(r) => found = r, + Err(e) => tracing::error!("[api/v1/products] calcul KO pour {f}: {e:#}"), + } + } + } + + let Some(found) = found else { + return fail( + ApiErrorCode::NotFound, + match &parsed.fournisseur { + Some(f) => format!("Produit « {codein} » introuvable chez le fournisseur « {f} »."), + None => format!( + "Aucun instantané pour le produit « {codein} ». Ajoutez « fournisseur=… » pour que l'API calcule le lot à la demande." + ), + }, + None, + ); + }; + + // Métriques réseau + gamme serveur relues maintenant : le payload date du dernier + // calcul de la grille, alors que ces deux données évoluent indépendamment. + let Some(enriched) = enrich_rows(vec![found.row]).await.into_iter().next() else { + return fail(ApiErrorCode::InternalError, "Enrichissement impossible.", None); + }; + let network_fetched_at = enriched.network.as_ref().and_then(|n| n.fetched_at.clone()); + let has_network = enriched.network.is_some(); + ok( + enriched, + None, + meta(vec![ + ("computedAt", json!(found.computed_at)), + ("networkFetchedAt", json!(network_fetched_at)), + ("hasNetwork", json!(has_network)), + ]), + ) +} + +fn maintenant_iso() -> String { + crate::db::iso_utc(chrono::Utc::now()) +} + +/// `PUT /api/v1/products/{codein}/gamme` `{ "gamme": "A", "fournisseur"?: "…" }` +/// +/// Propose une gamme pour un article. La proposition reste **à valider** dans la +/// Grille. `fournisseur` n'est requis que si l'article est référencé chez plusieurs +/// fournisseurs, ou s'il n'apparaît pas encore dans l'instantané. +pub async fn put_product_gamme(ApiAuth(auth): ApiAuth, Path(codein): Path, body: Bytes) -> Response { + if codein.is_empty() { + return fail(ApiErrorCode::BadRequest, "Paramètre 'codein' requis.", None); + } + let corps = match lire_corps_json(&body) { + Ok(c) => c, + Err(r) => return r, + }; + let parsed = match parse_product_gamme_body(&corps) { + Ok(p) => p, + Err(issues) => return fail(ApiErrorCode::BadRequest, "Corps de requête invalide.", Some(issues.to_json())), + }; + + let fournisseur = match parsed.fournisseur { + Some(f) => f, + None => { + let fournisseurs = match list_grid_suppliers_for_codein(&codein).await { + Ok(f) => f, + Err(e) => return interne(e), + }; + if fournisseurs.is_empty() { + return fail( + ApiErrorCode::NotFound, + format!( + "Aucun instantané pour le produit « {codein} ». Ajoutez « fournisseur » au corps de la requête pour que l'API calcule le lot à la demande." + ), + None, + ); + } + if fournisseurs.len() > 1 { + return fail( + ApiErrorCode::BadRequest, + format!("Le produit « {codein} » est référencé chez plusieurs fournisseurs : précisez « fournisseur »."), + Some(json!({ "fournisseurs": fournisseurs })), + ); + } + fournisseurs[0].code_fournisseur.clone() + } + }; + + let res = match affecter_gammes(&fournisseur, &[(codein, parsed.gamme)], parsed.compute, &auth.subject).await { + Ok(r) => r, + Err(r) => return r, + }; + let mut data = match res.resultats.first().map(serde_json::to_value) { + Some(Ok(Value::Object(m))) => m, + _ => Map::new(), + }; + data.insert("codeFournisseur".into(), json!(res.code_fournisseur)); + data.insert("nomFournisseur".into(), json!(res.nom_fournisseur)); + ok(Value::Object(data), None, meta(vec![("enregistreLe", json!(maintenant_iso()))])) +} + +/// `POST /api/v1/gammes` `{ "fournisseur": "…", "changes": [{ "codein": "…", "gamme": "A" }, …] }` +/// +/// Propose la gamme de plusieurs articles d'un fournisseur en un appel, à valider +/// dans la Grille. Tout ou rien : un article inconnu chez le fournisseur fait +/// échouer l'appel (404). +pub async fn post_gammes(ApiAuth(auth): ApiAuth, body: Bytes) -> Response { + let corps = match lire_corps_json(&body) { + Ok(c) => c, + Err(r) => return r, + }; + let parsed = match parse_gammes_batch_body(&corps) { + Ok(p) => p, + Err(issues) => return fail(ApiErrorCode::BadRequest, "Corps de requête invalide.", Some(issues.to_json())), + }; + let changes: Vec<(String, String)> = parsed.changes.into_iter().map(|c| (c.codein, c.gamme)).collect(); + let res = match affecter_gammes(&parsed.fournisseur, &changes, parsed.compute, &auth.subject).await { + Ok(r) => r, + Err(r) => return r, + }; + let demandes = res.resultats.len(); + let modifies = res.resultats.iter().filter(|r| r.modifie).count(); + let a_valider = res.resultats.iter().filter(|r| r.statut == "a_valider").count(); + ok( + &res.resultats, + None, + meta(vec![ + ("codeFournisseur", json!(res.code_fournisseur)), + ("nomFournisseur", json!(res.nom_fournisseur)), + ("demandes", json!(demandes)), + ("modifies", json!(modifies)), + ("aValider", json!(a_valider)), + ("enregistreLe", json!(maintenant_iso())), + ]), + ) +} + +/// `GET /api/v1/fournisseurs?search=…&withData=1` +/// +/// Liste des fournisseurs : tout le référentiel (`fouident`), enrichi de la +/// fraîcheur de l'instantané de grille quand il existe. `withData=1` restreint aux +/// seuls fournisseurs déjà présents dans l'instantané. +pub async fn get_fournisseurs(ApiAuth(_auth): ApiAuth, Query(params): Query>) -> Response { + let parsed = match parse_fournisseurs_query(&query_map(params)) { + Ok(p) => p, + Err(issues) => return fail(ApiErrorCode::BadRequest, "Paramètres invalides.", Some(issues.to_json())), + }; + let snapshot = match list_grid_suppliers().await { + Ok(s) => s, + Err(e) => return interne(e), + }; + + if parsed.with_data.as_deref() == Some("1") { + let term = parsed.search.as_ref().map(|s| s.to_lowercase()); + let data: Vec = snapshot + .iter() + .filter(|s| match &term { + None => true, + Some(t) => { + s.code_fournisseur.to_lowercase().contains(t.as_str()) + || s.nom_fournisseur.as_deref().unwrap_or("").to_lowercase().contains(t.as_str()) + } + }) + .map(|s| { + json!({ "code": s.code_fournisseur, "nom": s.nom_fournisseur, "hasSnapshot": true, "rowCount": s.row_count, "computedAt": s.computed_at }) + }) + .collect(); + let total = data.len(); + return ok(data, None, meta(vec![("total", json!(total)), ("scope", json!("instantané uniquement"))])); + } + + let by_code: HashMap<&str, &super::store::GridSupplier> = snapshot.iter().map(|s| (s.code_fournisseur.as_str(), s)).collect(); + let fournisseurs = crate::ff::pg::pg_get_fournisseurs(parsed.search.as_deref()).await; + let data: Vec = fournisseurs + .iter() + .map(|f| { + let snap = by_code.get(f.code.as_str()); + json!({ + "code": f.code, + "nom": f.nom, + "hasSnapshot": snap.is_some(), + "rowCount": snap.map(|s| s.row_count).unwrap_or(0), + "computedAt": snap.and_then(|s| s.computed_at.clone()), + }) + }) + .collect(); + let total = data.len(); + ok(data, None, meta(vec![("total", json!(total)), ("withSnapshot", json!(by_code.len()))])) +} + +/// `GET /api/v1/network/{codeCentrale}` +/// +/// Métriques réseau Qlik d'un produit et sa courbe de quantités mensuelles. +/// Lecture **exclusive** de la table `qlik_network_metrics` : aucun appel à Qlik. +pub async fn get_network(ApiAuth(_auth): ApiAuth, Path(code_centrale): Path) -> Response { + if code_centrale.is_empty() { + return fail(ApiErrorCode::BadRequest, "Paramètre 'codeCentrale' requis.", None); + } + let metrics = match crate::qlik::network_cache::get_network_metrics_by_code_centrale(std::slice::from_ref(&code_centrale)).await { + Ok(m) => m, + Err(e) => return fail(ApiErrorCode::InternalError, format!("{e:#}"), None), + }; + let Some(found) = metrics.get(&code_centrale) else { + return fail( + ApiErrorCode::NotFound, + format!( + "Aucune donnée réseau en cache pour le code centrale « {code_centrale} ». Lancez une synchronisation Qlik ou une recherche réseau." + ), + None, + ); + }; + let mut data = Map::new(); + data.insert("codeCentrale".into(), json!(code_centrale)); + if let Ok(Value::Object(m)) = serde_json::to_value(found) { + data.extend(m); + } + ok(Value::Object(data), None, meta(vec![("fetchedAt", json!(found.fetched_at)), ("source", json!("cache"))])) +} diff --git a/server/src/grid/enrich.rs b/server/src/grid/enrich.rs new file mode 100644 index 0000000..335e4c7 --- /dev/null +++ b/server/src/grid/enrich.rs @@ -0,0 +1,148 @@ +//! CollectFlow — Enrichissement des lignes servies par `/api/v1` +//! (port de `src/lib/api-enrich.ts`). +//! +//! L'instantané `grid_rows` fige la ligne au moment du calcul. Deux informations +//! doivent pourtant refléter l'état **courant** au moment de l'appel : +//! +//! 1. **Métriques réseau Qlik** — la table `qlik_network_metrics` est mise à jour par +//! les synchronisations et les recherches réseau, indépendamment du calcul de la +//! grille. On les relit donc à la lecture, et on les expose telles quelles quand +//! elles existent (`network: null` sinon). +//! +//! 2. **Gamme serveur** — `codeGamme` peut être surchargée par un snapshot de session. +//! L'API expose la gamme **non modifiée**, celle réellement présente en base, +//! relue à chaque appel. +//! +//! Ce n'est pas un recalcul : des lectures indexées bornées au nombre de lignes. + +use super::gammes_a_valider::gammes_a_valider_par_article; +use super::months; +use super::network_trend::{compute_network_trend, trend_label, Direction}; +use super::types::ProductRow; +use crate::ff::pg::pg_get_gammes_by_codeins; +use crate::qlik::network_cache::{get_network_metrics_by_code_centrale, NetworkMetricCached}; +use serde::Serialize; +use std::collections::{BTreeMap, BTreeSet, HashMap}; + +/// Tendance réseau 12 mois, telle qu'affichée dans la colonne « Tendance » de la Grille. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ProductTrend { + pub direction: Direction, + /// Variation des 4 derniers mois vs les 4 premiers (0.23 = +23 %). `null` si aucune base. + #[serde(serialize_with = "super::js::opt_num::serialize")] + pub pct: Option, + /// Libellé prêt à afficher : « Forte hausse », « Nouveau », « Stable »… + pub label: &'static str, + /// `true` = tendance calculée sur la quantité **par magasin vendeur**. + pub sur_qte_par_magasin: bool, + /// Aucune vente sur les 4 premiers mois, des ventes sur les 4 derniers. + pub nouveau: bool, + /// Nombre de mois de la série (12 quand elle est complète). + pub months_used: usize, +} + +/// Ligne enrichie : la ligne, puis la gamme serveur, la proposition en attente, +/// les métriques réseau et la tendance (ordre des clés du TS). +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct EnrichedProductRow { + #[serde(flatten)] + pub row: ProductRow, + /// Gamme telle qu'elle existe sur le serveur PostgreSQL — non modifiée. + pub code_gamme_serveur: Option, + /// Gamme proposée par l'API et pas encore validée dans la Grille. + pub gamme_a_valider: Option, + /// Métriques réseau Qlik en cache, ou `null` si le produit n'en a pas. + pub network: Option, + /// Tendance réseau sur les 12 derniers mois, `null` sans série mensuelle. + pub trend: Option, +} + +/// Enrichit un lot de lignes. Les lectures sont indépendantes et lancées en +/// parallèle ; un échec de l'une n'empêche pas les autres (l'API reste servie, avec +/// la valeur de l'instantané en repli). +pub async fn enrich_rows(rows: Vec) -> Vec { + if rows.is_empty() { + return vec![]; + } + let mut vus = BTreeSet::new(); + let codes_centraux: Vec = + rows.iter().filter_map(|r| r.code_centrale.clone()).filter(|c| !c.is_empty() && vus.insert(c.clone())).collect(); + let mut vus = BTreeSet::new(); + let codeins: Vec = rows.iter().map(|r| r.codein.clone()).filter(|c| !c.is_empty() && vus.insert(c.clone())).collect(); + + let (network, gammes, a_valider) = tokio::join!( + async { + if codes_centraux.is_empty() { + return HashMap::new(); + } + get_network_metrics_by_code_centrale(&codes_centraux).await.unwrap_or_else(|e| { + tracing::error!("[api-enrich] métriques réseau KO: {}", crate::error::short(&e, 160)); + HashMap::new() + }) + }, + async { + if codeins.is_empty() { + return HashMap::new(); + } + pg_get_gammes_by_codeins(&codeins).await.unwrap_or_else(|e| { + tracing::error!("[api-enrich] gammes serveur KO: {}", crate::error::short(&e, 160)); + HashMap::new() + }) + }, + async { + gammes_a_valider_par_article(&codeins).await.unwrap_or_else(|e| { + tracing::error!("[api-enrich] gammes à valider KO: {}", crate::error::short(&e.message, 160)); + HashMap::new() + }) + }, + ); + + let aujourd_hui = months::today(); + rows.into_iter() + .map(|mut row| { + let metrics = row.code_centrale.as_deref().filter(|c| !c.is_empty()).and_then(|c| network.get(c)).cloned(); + // Gamme absente de la base = pas de gamme, et non « inconnue ». + let gamme_serveur = gammes.get(&row.codein).cloned(); + let gamme_a_valider = a_valider.get(&format!("{}|{}", row.code_fournisseur, row.codein)).cloned(); + + // `codeGammeInit` porte déjà la sémantique « état serveur » dans la Grille : + // on la garde alignée pour que les deux champs ne divergent jamais. + if !gammes.is_empty() { + row.code_gamme_init = gamme_serveur.clone(); + } + + // Les colonnes réseau de la ligne datent du calcul : on les réaligne sur le + // cache quand il a quelque chose à dire. + if let Some(m) = &metrics { + row.ca_reseau = Some(m.ca_reseau); + row.qte_reseau = Some(m.qte_reseau); + row.nb_magasins_reseau = Some(m.nb_magasins_reseau); + row.ca_par_magasin_reseau = Some(m.ca_par_magasin_reseau); + row.marge_pct_reseau = Some(m.marge_pct_reseau); + row.qte_reseau_by_month = Some(m.qte_by_month.clone()); + row.network_fetched_at = m.fetched_at.clone(); + } + + // Tendance : mêmes entrées que la colonne « Tendance » de la Grille — la + // série des quantités **et** celle du nombre de magasins vendeurs. + let serie = row.qte_reseau_by_month.clone().flatten(); + let nb_mag_by_month: Option> = metrics + .as_ref() + .and_then(|m| m.metrics_by_month.as_ref()) + .map(|d| d.iter().map(|(mois, v)| (mois.clone(), v.nb_mag.filter(|n| !n.is_nan()).unwrap_or(0.0))).collect()); + let t = compute_network_trend(serie.as_ref(), nb_mag_by_month.as_ref(), aujourd_hui); + let trend = t.has_data.then(|| ProductTrend { + direction: t.direction, + pct: t.pct, + label: trend_label(t.pct, t.nouveau), + sur_qte_par_magasin: t.sur_qte_par_magasin, + nouveau: t.nouveau, + months_used: t.values.len(), + }); + + EnrichedProductRow { row, code_gamme_serveur: gamme_serveur, gamme_a_valider, network: metrics, trend } + }) + .collect() +} diff --git a/server/src/grid/grid_rows.rs b/server/src/grid/grid_rows.rs new file mode 100644 index 0000000..1042d1a --- /dev/null +++ b/server/src/grid/grid_rows.rs @@ -0,0 +1,135 @@ +//! Routes internes de la Grille (ports de `src/app/api/grid/rows/route.ts` et +//! `src/app/api/grid/rows/store-patch/route.ts`). + +use super::js::JsNum; +use super::product_rows::{get_product_rows, get_store_patch, GetProductRowsInput}; +use super::types::ProductRow; +use crate::auth::SessionUser; +use axum::body::{Body, Bytes}; +use axum::extract::Query; +use axum::http::{header, StatusCode}; +use axum::response::{IntoResponse, Response}; +use axum::Json; +use futures::channel::mpsc; +use futures::SinkExt; +use serde::Serialize; +use serde_json::json; + +/// `searchParams.get(nom)` : première occurrence. +pub fn premier<'a>(params: &'a [(String, String)], nom: &str) -> Option<&'a str> { + params.iter().find(|(k, _)| k == nom).map(|(_, v)| v.as_str()) +} + +#[derive(Serialize)] +struct Lot<'a> { + #[serde(rename = "type")] + type_: &'static str, + rows: &'a [ProductRow], + loaded: JsNum, + total: usize, +} + +/// Indice de `Array.prototype.slice` (troncature, bornage à la longueur). +fn indice_slice(x: f64, len: usize) -> usize { + if x.is_nan() { + 0 + } else { + x.trunc().clamp(0.0, len as f64) as usize + } +} + +fn ndjson(v: &impl Serialize) -> Bytes { + let mut out = serde_json::to_vec(v).unwrap_or_default(); + out.push(b'\n'); + Bytes::from(out) +} + +/// `GET /api/grid/rows?fournisseur=…&magasin=…&chunkSize=…&refresh=1` +/// +/// Flux NDJSON : `{ type: "start" }`, puis des lots `{ type: "chunk", rows, loaded, +/// total }`, puis `{ type: "done" }` — ou `{ type: "error", error }`. +pub async fn get_grid_rows(SessionUser(_utilisateur): SessionUser, Query(params): Query>) -> Response { + let Some(code_fournisseur) = premier(¶ms, "fournisseur").filter(|f| !f.is_empty()).map(str::to_string) else { + return (StatusCode::BAD_REQUEST, Json(json!({ "error": "Missing fournisseur" }))).into_response(); + }; + let magasin = premier(¶ms, "magasin").unwrap_or("TOTAL").to_string(); + let chunk_size = { + let brut = premier(¶ms, "chunkSize").map(super::js::js_number_from_str).unwrap_or(150.0); + // Math.max(50, Math.min(500, x)) : NaN se propage. + if brut.is_nan() { + f64::NAN + } else { + brut.clamp(50.0, 500.0) + } + }; + // `code1`/`code2`/`code3` ne sont pas lus : la Grille filtre en local + // (getProductRows ne s'en sert pas). + let force_refresh = premier(¶ms, "refresh") == Some("1"); + + // Client parti (changement de fournisseur, page quittée) : on arrête de + // sérialiser et d'envoyer. Le calcul lui-même continue — il est partagé et + // mis en cache, la prochaine ouverture en profitera. + let (mut tx, rx) = mpsc::channel::>(4); + tokio::spawn(async move { + if tx.send(Ok(ndjson(&json!({ "type": "start", "loaded": 0, "total": null })))).await.is_err() { + return; + } + let input = GetProductRowsInput { code_fournisseur, magasin: Some(magasin), force_refresh }; + match get_product_rows(input).await { + Ok(shared) => { + let total = shared.len(); + let mut start = 0.0f64; + while start < total as f64 { + let ligne = { + let rows = shared.rows.read(); + let len = rows.len(); + let (a, b) = (indice_slice(start, len), indice_slice(start + chunk_size, len)); + let lot = &rows[a..b.max(a)]; + ndjson(&Lot { type_: "chunk", rows: lot, loaded: JsNum((start + lot.len() as f64).min(total as f64)), total }) + }; + if tx.send(Ok(ligne)).await.is_err() { + return; + } + start += chunk_size; + tokio::task::yield_now().await; + } + let _ = tx.send(Ok(ndjson(&json!({ "type": "done", "loaded": total, "total": total })))).await; + } + Err(e) => { + let _ = tx.send(Ok(ndjson(&json!({ "type": "error", "error": e.to_string() })))).await; + } + } + }); + + ( + [ + (header::CONTENT_TYPE, "application/x-ndjson; charset=utf-8"), + (header::CACHE_CONTROL, "no-store"), + (header::HeaderName::from_static("x-accel-buffering"), "no"), + ], + Body::from_stream(rx), + ) + .into_response() +} + +/// `GET /api/grid/rows/store-patch?fournisseur=…&magasin=…` +/// +/// Complément de l'API FF pour un magasin (cf. `store_patch`). La Grille charge +/// toujours les lignes « tous magasins » et applique ce complément à son arrivée : +/// le changement de magasin n'attend plus rien. +pub async fn get_grid_store_patch(SessionUser(_utilisateur): SessionUser, Query(params): Query>) -> Response { + let code_fournisseur = premier(¶ms, "fournisseur").unwrap_or(""); + let magasin = premier(¶ms, "magasin").unwrap_or(""); + // Un code inconnu ferait de chaque article un candidat : autant d'appels + // inutiles à l'API FF. « Tous magasins » n'a pas de complément. + if code_fournisseur.is_empty() || !crate::domain::MAGASINS.iter().any(|m| m.code == magasin) { + return (StatusCode::BAD_REQUEST, Json(json!({ "error": "Paramètres fournisseur et magasin requis." }))).into_response(); + } + match get_store_patch(code_fournisseur, magasin).await { + Ok(patch) => ([(header::CACHE_CONTROL, "no-store")], Json(&*patch)).into_response(), + Err(e) => { + tracing::error!("[store-patch] {code_fournisseur}/{magasin} KO: {}", crate::error::short(&e, 200)); + (StatusCode::BAD_GATEWAY, Json(json!({ "error": "Complément indisponible." }))).into_response() + } + } +} diff --git a/server/src/grid/mod.rs b/server/src/grid/mod.rs index 75a9cef..9966e94 100644 --- a/server/src/grid/mod.rs +++ b/server/src/grid/mod.rs @@ -1,27 +1,30 @@ //! Module `grid` — grille d'assortiment. //! -//! Phase 1 (ce qui ne dépend ni de la base FF ni de Qlik) : //! - [`types`] : `ProductRow` et types associés (JSON identique au TS) ; -//! - [`store`] : instantané persisté `grid_rows` (lu par `/api/v1`) ; -//! - [`rows_cache`] : cache mémoire des lignes calculées (10 min) ; +//! - [`product_rows`] : calcul des lignes (`getProductRows`), branché sur +//! [`rows_cache`] (cache mémoire 10 min) et [`store`] (instantané `grid_rows`) ; //! - [`network_trend`], [`months`], [`store_patch`] : logique pure ; //! - [`gammes_a_valider`], [`enregistrer`], [`snapshots`] : gammes et snapshots ; -//! - [`exports`] : exports Excel ; [`api_schemas`], [`api_v1`], [`openapi`] : API publique. -//! -//! Phase 2 : calcul de la grille (`getProductRows`), enrichissement `/api/v1` -//! (`api-enrich`), routes `/api/v1/grid`, `/products/*`, `/gammes`, -//! `/fournisseurs`, `/network/*` et `/api/grid/rows*`. +//! - [`grid_rows`] : `/api/grid/rows` (NDJSON) et `/api/grid/rows/store-patch` ; +//! - [`exports`] : exports Excel ; +//! - API publique : [`api_schemas`], [`enrich`], [`api_gammes`], [`api_v1`], +//! [`api_v1_grid`], [`openapi`]. +pub mod api_gammes; pub mod api_schemas; pub mod api_v1; +pub mod api_v1_grid; pub mod enregistrer; +pub mod enrich; pub mod exports; pub mod gammes; pub mod gammes_a_valider; +pub mod grid_rows; pub mod js; pub mod months; pub mod network_trend; pub mod openapi; +pub mod product_rows; pub mod rows_cache; pub mod session; pub mod snapshots; @@ -32,9 +35,11 @@ pub mod zod; #[cfg(test)] mod tests_db; +#[cfg(test)] +mod tests_ff; use crate::rpc::Handler; -use axum::routing::{get, post}; +use axum::routing::{get, post, put}; use axum::Router; /// Routes HTTP du module. @@ -44,6 +49,15 @@ pub fn router() -> Router { .route("/api/export/modified-gammes", post(exports::export_modified_gammes)) .route("/api/v1/nomenclatures", get(api_v1::get_nomenclatures)) .route("/api/v1/openapi.json", get(openapi::get_openapi)) + .route("/api/grid/rows", get(grid_rows::get_grid_rows)) + .route("/api/grid/rows/store-patch", get(grid_rows::get_grid_store_patch)) + .route("/api/v1/grid", get(api_v1_grid::get_grid)) + .route("/api/v1/products/search", get(api_v1_grid::get_products_search)) + .route("/api/v1/products/{codein}", get(api_v1_grid::get_product)) + .route("/api/v1/products/{codein}/gamme", put(api_v1_grid::put_product_gamme)) + .route("/api/v1/gammes", post(api_v1_grid::post_gammes)) + .route("/api/v1/fournisseurs", get(api_v1_grid::get_fournisseurs)) + .route("/api/v1/network/{codeCentrale}", get(api_v1_grid::get_network)) } /// Actions (`POST /api/actions/{nom}`) du module. diff --git a/server/src/grid/product_rows.rs b/server/src/grid/product_rows.rs new file mode 100644 index 0000000..bb76d8c --- /dev/null +++ b/server/src/grid/product_rows.rs @@ -0,0 +1,879 @@ +//! Calcul des lignes de la Grille (port de `src/features/grid/api/get-product-rows.ts`). +//! +//! Sept requêtes SQL en parallèle sur la base miroir FF, jointure avec le cache +//! réseau Qlik, report des gammes du dernier snapshot. Le résultat est gardé +//! 10 minutes en mémoire ([`super::rows_cache`]) et persisté dans `grid_rows` +//! (magasin « TOTAL » seulement), d'où il est resservi tant qu'il date du mois +//! en cours (`serveFromSnapshot`). + +use super::months; +use super::network_trend::NB_MAGASINS_RESEAU; +use super::rows_cache::{self, GridRows, RowsFuture, SharedRows, StorePatchFuture}; +use super::store::{read_grid_snapshot, upsert_grid_rows_shared}; +use super::store_patch::{apply_store_patch_in_place, build_store_patch_entries, store_patch_candidates}; +use super::types::{ProductRow, StorePatch, StorePatchEntry}; +use crate::ff::api::{build_last12_months_range, get_mensuel_by_articles, FfArticle}; +use crate::ff::pg::{self, PgStockRow}; +use crate::qlik::network_cache::{get_network_metrics_by_code_centrale, NetworkMetricCached}; +use futures::future::{BoxFuture, FutureExt}; +use serde_json::{Map, Value}; +use sqlx::Row; +use std::collections::{BTreeMap, BTreeSet, HashMap}; +use std::sync::Arc; +use std::time::Duration; + +/// Paramètres de `getProductRows`. (`filters` n'est pas repris : le TS ne s'en +/// servait pas, la Grille filtre en local.) +#[derive(Debug, Clone, Default)] +pub struct GetProductRowsInput { + pub code_fournisseur: String, + /// `None` = « TOTAL ». + pub magasin: Option, + pub force_refresh: bool, +} + +impl GetProductRowsInput { + pub fn total(code_fournisseur: &str, force_refresh: bool) -> Self { + Self { code_fournisseur: code_fournisseur.to_string(), magasin: Some("TOTAL".into()), force_refresh } + } + fn magasin(&self) -> &str { + self.magasin.as_deref().unwrap_or("TOTAL") + } +} + +/// Au-delà, l'instantané est servi mais un recalcul est relancé en arrière-plan. +const SNAPSHOT_REFRESH_AFTER: Duration = Duration::from_secs(20 * 60 * 60); + +fn court(e: &dyn std::fmt::Display) -> String { + crate::error::short(e, 200) +} + +/// `Number(x) || 0` sur une valeur déjà numérique. +fn or0(v: f64) -> f64 { + if v.is_nan() { + 0.0 + } else { + v + } +} + +/// Rafraîchit UNIQUEMENT la colonne INIT (codeGammeInit = gamme serveur courante) +/// sur des lignes servies depuis le cache. La colonne Gamme (codeGamme) n'est pas +/// touchée : elle conserve la valeur issue du snapshot. Ainsi l'état serveur (INIT) +/// est toujours à jour à chaque chargement, même sur un hit du cache 10 min — et le +/// marqueur "modifié" (Gamme ≠ INIT) se résorbe dès que le serveur rattrape la modif. +async fn refresh_gamme_init(rows: &GridRows, code_fournisseur: &str) { + match pg::pg_get_gammes_by_fournisseur(code_fournisseur).await { + Ok(gamme_map) => appliquer_gammes_init(rows, &gamme_map), + Err(e) => tracing::error!("[getProductRows] refreshGammeInit error: {}", court(&e)), + } +} + +fn appliquer_gammes_init(rows: &GridRows, gamme_map: &HashMap) { + if gamme_map.is_empty() { + return; + } + for row in rows.rows.write().iter_mut() { + if let Some(live) = gamme_map.get(&row.codein) { + row.code_gamme_init = Some(live.clone()); + } + } +} + +/// Écrit l'instantané de grille lu par /api/v1 (table `grid_rows`). +/// +/// Seul `magasin = TOTAL` est persisté : `ProductRow` embarque déjà les ventilations +/// par magasin (sales12mByStore, caByStore…), inutile de stocker trois variantes. +/// Toute erreur est avalée : la persistance ne doit jamais casser l'affichage. +async fn persist_grid_snapshot(input: &GetProductRowsInput, rows: &GridRows) { + if input.magasin() != "TOTAL" { + return; + } + match upsert_grid_rows_shared(&input.code_fournisseur, rows).await { + Ok(n) => tracing::info!("[getProductRows] instantané persisté: {n} lignes pour {}", input.code_fournisseur), + Err(e) => tracing::error!("[getProductRows] persistance grid_rows KO: {}", court(&e.message)), + } +} + +/// Lignes de la Grille d'un fournisseur (cache 10 min, instantané, sinon calcul). +pub fn get_product_rows(input: GetProductRowsInput) -> BoxFuture<'static, anyhow::Result> { + async move { + let cache_key = rows_cache::grid_rows_cache_key(&input.code_fournisseur, input.magasin.as_deref()); + if !input.force_refresh { + if let Some(cached) = rows_cache::get_fresh(&cache_key) { + // La colonne INIT doit refléter l'état serveur à CHAQUE chargement, + // y compris sur un hit de cache (les données lourdes restent, elles, cachées). + refresh_gamme_init(&cached, &input.code_fournisseur).await; + return Ok(cached); + } + } + + // Un calcul déjà en cours est réutilisé même en rechargement forcé : il part + // de données fraîches, et en lancer un second en parallèle doublait la charge + // (double clic, onglet rouvert…) sans rien apporter. + if let Some(pending) = rows_cache::pending_get(&cache_key) { + return pending.await.map_err(|e| anyhow::anyhow!(e)); + } + + // Instantané persisté (calcul de la nuit ou de la dernière ouverture) : servi + // en une lecture au lieu d'un recalcul de 10 à 40 s. Jamais en rechargement + // forcé — c'est précisément ce que l'utilisateur demande d'éviter. + if !input.force_refresh { + if let Some(servi) = serve_from_snapshot(&input, &cache_key).await { + return Ok(servi); + } + } + + let version_au_depart = rows_cache::gamme_save_version(&input.code_fournisseur); + // Le calcul est lancé à part, comme une promesse JS : il continue même si + // l'appelant s'en va (client déconnecté), et il est partagé entre appelants. + let (tx, rx) = tokio::sync::oneshot::channel::>(); + let partage: RowsFuture = + async move { rx.await.unwrap_or_else(|_| Err("calcul de la grille interrompu".to_string())) }.boxed().shared(); + rows_cache::pending_insert(&cache_key, partage.clone()); + let entree = input.clone(); + let cle = cache_key.clone(); + tokio::spawn(async move { + let resultat = async { + let mut rows = build_product_rows(&entree).await?; + if rows_cache::gamme_save_version(&entree.code_fournisseur) != version_au_depart { + // Gammes enregistrées pendant le calcul : on reprend le dernier snapshot. + apply_snapshot_changes(&mut rows, load_latest_snapshot_changes(&entree.code_fournisseur).await.as_ref()); + } + let shared = GridRows::new(rows); + rows_cache::cache_grid_rows(&cle, shared.clone()); + rows_cache::pending_remove(&cle); + // Persiste l'instantané pour /api/v1 : le calcul vient d'avoir lieu, on + // arrête simplement de jeter le résultat. Volontairement NON bloquant — + // l'affichage de la Grille ne doit pas attendre l'écriture. + let a_persister = shared.clone(); + let code = entree.code_fournisseur.clone(); + let ecriture: Persistance = + tokio::spawn(async move { persist_grid_snapshot(&entree, &a_persister).await }).map(|_| ()).boxed().shared(); + PERSISTANCES.lock().insert(code.clone(), ecriture.clone()); + tokio::spawn(async move { + ecriture.clone().await; + let mut p = PERSISTANCES.lock(); + if p.get(&code).map(|e| e.ptr_eq(&ecriture)).unwrap_or(false) { + p.remove(&code); + } + }); + anyhow::Ok(shared) + } + .await; + let resultat = resultat.map_err(|e| { + rows_cache::pending_remove(&cle); + format!("{e:#}") + }); + let _ = tx.send(resultat); + }); + partage.await.map_err(|e| anyhow::anyhow!(e)) + } + .boxed() +} + +/// Écriture de l'instantané en cours, par fournisseur. +type Persistance = futures::future::Shared>; +static PERSISTANCES: std::sync::LazyLock>> = + std::sync::LazyLock::new(|| parking_lot::Mutex::new(HashMap::new())); + +/// Attend la fin de l'écriture de l'instantané lancée par le dernier calcul du +/// fournisseur (s'il y en a une en cours). +/// +/// Écart volontaire avec le TS : `/api/v1` relisait `grid_rows` juste après un +/// calcul à la demande alors que la persistance, non attendue, pouvait ne pas +/// être finie — d'où un faux « aucun article ». Les routes attendent ici l'écriture. +pub async fn attendre_persistance(code_fournisseur: &str) { + let en_cours = PERSISTANCES.lock().get(code_fournisseur).cloned(); + if let Some(e) = en_cours { + e.await; + } +} + +/// `getProductRows({ codeFournisseur, magasin: "TOTAL", forceRefresh: true })` : +/// recalcul complet (synchro nocturne, préchauffage). Renvoie le nombre de lignes. +pub async fn get_product_rows_for_sync(code_fournisseur: &str) -> anyhow::Result { + let rows = get_product_rows(GetProductRowsInput::total(code_fournisseur, true)).await?; + Ok(rows.len()) +} + +/// Sert la Grille depuis `grid_rows` quand c'est possible : mêmes lignes que le +/// calcul en direct (elles en sont issues), avec la gamme FF du moment (INIT) et +/// les dernières gammes enregistrées reportées dessus. Hors « tous magasins », +/// le rattrapage du magasin choisi est appliqué comme après un calcul. +/// +/// `None` = pas d'instantané utilisable (absent, ancien format, autre mois) : +/// l'appelant recalcule. +async fn serve_from_snapshot(input: &GetProductRowsInput, cache_key: &str) -> Option { + let code = input.code_fournisseur.as_str(); + let magasin = input.magasin(); + let snapshot = match read_grid_snapshot(code).await { + Ok(s) => s, + Err(e) => { + tracing::error!("[getProductRows] lecture grid_rows KO: {}", court(&e.message)); + return None; + } + }; + let snapshot = snapshot.filter(|s| !s.rows.is_empty())?; + let computed_at = snapshot.computed_at; + let mut rows = snapshot.rows; + + // Les colonnes réseau de l'instantané datent de son calcul : on les relit dans + // le cache Qlik, comme le calcul en direct. Sans cela, une synchro Qlik faite + // depuis (au changement de mois notamment) restait invisible jusqu'au + // prochain recalcul, et la tendance réseau disparaissait. + let codes: Vec = rows.iter().map(|r| r.code_centrale.clone().unwrap_or_default()).collect(); + let (changes, network, gammes) = + tokio::join!(load_latest_snapshot_changes(code), fetch_network_metrics(&codes), pg::pg_get_gammes_by_fournisseur(code)); + match gammes { + Ok(g) if !g.is_empty() => { + for row in rows.iter_mut() { + if let Some(live) = g.get(&row.codein) { + row.code_gamme_init = Some(live.clone()); + } + } + } + Ok(_) => {} + Err(e) => tracing::error!("[getProductRows] refreshGammeInit error: {}", court(&e)), + } + apply_snapshot_changes(&mut rows, changes.as_ref()); + enrich_with_network_metrics(&mut rows, network.as_ref()); + + if magasin != "TOTAL" { + let range = build_last12_months_range(); + reconcile_selected_store_from_mensuel_api(&mut rows, magasin, &range.date_debut, &range.date_fin, &last12_periods()).await; + } + + let shared = GridRows::new(rows); + rows_cache::cache_grid_rows(cache_key, shared.clone()); + + // Instantané qui n'a pas été recalculé par la nuit : on le rafraîchit sans + // faire attendre l'utilisateur (la prochaine ouverture en profitera). + // Toujours le calcul « tous magasins » : c'est lui qui réécrit l'instantané. + let age = chrono::Utc::now().naive_utc() - computed_at; + if age.to_std().map(|d| d > SNAPSHOT_REFRESH_AFTER).unwrap_or(false) + && !rows_cache::pending_contains(&rows_cache::grid_rows_cache_key(code, Some("TOTAL"))) + { + let code = code.to_string(); + tokio::spawn(async move { + if let Err(e) = get_product_rows(GetProductRowsInput::total(&code, true)).await { + tracing::error!("[getProductRows] rafraîchissement en arrière-plan KO: {}", court(&e)); + } + }); + } + Some(shared) +} + +/// Les 12 derniers mois complets, clés « YYYYMM » triées (même fenêtre que le calcul). +fn last12_periods() -> Vec { + months::get_last12_months(months::today()) +} + +#[derive(Default, Clone, Copy)] +struct PeriodData { + qty: f64, + ca: f64, + marge: f64, + stock: f64, + receptions: f64, +} + +/// `Math.max(...valeurs)` : `NaN` si l'une l'est. +fn js_max(valeurs: impl Iterator) -> f64 { + valeurs.fold(f64::NEG_INFINITY, |m, v| if m.is_nan() || v.is_nan() { f64::NAN } else { m.max(v) }) +} + +/// Le calcul complet (`buildProductRows`). +async fn build_product_rows(input: &GetProductRowsInput) -> anyhow::Result> { + let code_fournisseur = input.code_fournisseur.as_str(); + let magasin = input.magasin(); + tracing::info!("\n>>> [getProductRows] supplier: {code_fournisseur}, magasin: {magasin}"); + + let range = build_last12_months_range(); + let (date_debut, date_fin) = (range.date_debut.as_str(), range.date_fin.as_str()); + + // ─── Phase 1 : 7 requêtes SQL en parallèle ──────────────────────────── + // Phases 8 et 9 lancées dès maintenant : elles n'attendent que la liste des + // articles (codes centraux) ou le seul code fournisseur. + let articles_et_reseau = async { + let articles = pg::pg_get_articles_by_fournisseur(code_fournisseur).await.unwrap_or_else(|e| { + tracing::error!("[getProductRows] pgGetArticlesByFournisseur ERROR: {e:#}"); + vec![] + }); + let codes: Vec = articles + .iter() + .map(|a| a.code_centrale.as_deref().filter(|c| !c.is_empty()).map(|c| c.trim().to_string()).unwrap_or_default()) + .collect(); + let network = fetch_network_metrics(&codes).await; + (articles, network) + }; + let ou_vide = |nom: &'static str| move |e: &anyhow::Error| tracing::error!("[getProductRows] {nom} ERROR: {e:#}"); + let ((articles, network), mensuel_rows, gamme_map, nom_map, stock_map, commandes_map, prix_vente_map, snapshot_changes) = tokio::join!( + articles_et_reseau, + async { + pg::pg_get_mensuel_by_fournisseur(code_fournisseur, date_debut, date_fin) + .await + .inspect_err(ou_vide("pgGetMensuelByFournisseur")) + .unwrap_or_default() + }, + async { + pg::pg_get_gammes_by_fournisseur(code_fournisseur).await.inspect_err(ou_vide("pgGetGammesByFournisseur")).unwrap_or_default() + }, + async { + pg::pg_get_nomenclature_by_fournisseur(code_fournisseur) + .await + .inspect_err(ou_vide("pgGetNomenclatureByFournisseur")) + .unwrap_or_default() + }, + async { + pg::pg_get_stock_by_fournisseur(code_fournisseur).await.inspect_err(ou_vide("pgGetStockByFournisseur")).unwrap_or_default() + }, + async { + pg::pg_get_commandes_by_fournisseur(code_fournisseur) + .await + .inspect_err(ou_vide("pgGetCommandesByFournisseur")) + .unwrap_or_default() + }, + async { + pg::pg_get_prix_vente_by_fournisseur(code_fournisseur) + .await + .inspect_err(ou_vide("pgGetPrixVenteByFournisseur")) + .unwrap_or_default() + }, + load_latest_snapshot_changes(code_fournisseur), + ); + + tracing::info!("[getProductRows] {} articles, {} mensuel rows, {} gammes", articles.len(), mensuel_rows.len(), gamme_map.len()); + + // ─── Phase 2 : Fenêtre temporelle 12 mois complets ─────────────────── + let sorted_periods = last12_periods(); + + // ─── Phase 3 : Seed productMap depuis articles ──────────────────────── + // Ordre d'insertion conservé (Map JS) : index par codein. + let mut products: Vec = Vec::with_capacity(articles.len()); + let mut index: HashMap = HashMap::with_capacity(articles.len()); + for art in &articles { + if art.codein.is_empty() || index.contains_key(&art.codein) { + continue; + } + index.insert(art.codein.clone(), products.len()); + let non_nul = |v: Option| v.filter(|x| *x != 0.0 && !x.is_nan()); + products.push(ProductRow { + codein: art.codein.clone(), + code_fournisseur: art.codefou.clone(), + nom_fournisseur: art.nomfou.clone().unwrap_or_default(), + fournisseur_principal_code: art.codefou_principal.clone(), + fournisseur_principal_nom: art.nomfou_principal.clone(), + libelle1: art.libelle1.clone().unwrap_or_default(), + gtin: art.gtin.clone().unwrap_or_default(), + reference: art.reference.clone().unwrap_or_default(), + code_centrale: art.code_centrale.as_deref().filter(|c| !c.is_empty()).map(|c| c.trim().to_string()), + code_gamme: None, + code_gamme_init: None, + code_gamme_draft: None, + total_quantite: 0.0, + total_ca: 0.0, + total_marge: 0.0, + taux_marge: 0.0, + working_stores: vec![], + noid: (art.no_id != 0).then_some(art.no_id as f64), + pcb: non_nul(art.pcb), + prix_vente: non_nul(art.pv_central), + ..Default::default() + }); + } + + // ─── Phase 4 : Agréger mensuelRows → byPeriod par codein ───────────── + // SQL a déjà fait l'agrégation par (codein, site, mois). + // On somme les 2 sites (TOTAL) ET on accumule par site pour le switch client-side. + let allowed: BTreeSet<&str> = sorted_periods.iter().map(String::as_str).collect(); + let mut mensuel_by_codein: Vec<(String, HashMap)> = Vec::new(); + let mut mensuel_index: HashMap = HashMap::new(); + let mut mensuel_by_site: HashMap>> = HashMap::new(); + let mut store_months: HashMap> = HashMap::new(); + + for row in &mensuel_rows { + let periode = row.mois.replacen('-', "", 1); // "2026-02" → "202602" + if !allowed.contains(periode.as_str()) { + continue; + } + let stock = or0(row.stock_fin_mois.unwrap_or(f64::NAN)); + // Accumulation TOTAL (toujours tous sites) + let i = *mensuel_index.entry(row.codein.clone()).or_insert_with(|| { + mensuel_by_codein.push((row.codein.clone(), HashMap::new())); + mensuel_by_codein.len() - 1 + }); + let p = mensuel_by_codein[i].1.entry(periode.clone()).or_default(); + p.stock += stock; + p.qty += or0(row.qte_vendue); + p.ca += or0(row.ca_ht); + p.marge += or0(row.marge); + p.receptions += or0(row.qte_recue); + + // Accumulation par site (pour switch client-side) + let sp = mensuel_by_site.entry(row.codein.clone()).or_default().entry(row.site.clone()).or_default().entry(periode).or_default(); + sp.qty += or0(row.qte_vendue); + sp.ca += or0(row.ca_ht); + sp.marge += or0(row.marge); + sp.stock = stock; // SQL donne déjà le dernier mouvement + sp.receptions += or0(row.qte_recue); + + // Suivi magasins actifs (pour workingStores) : 1 vente suffit + if row.qte_vendue > 0.0 { + store_months.entry(row.codein.clone()).or_default().insert(row.site.clone()); + } + } + + // ─── Phase 5 : Remplir ProductRow depuis mensuelByCodein ───────────── + for (codein, by_period) in &mensuel_by_codein { + let Some(&pi) = index.get(codein) else { continue }; + let product = &mut products[pi]; + + let mut last_stock = 0.0; + for periode in &sorted_periods { + match by_period.get(periode) { + Some(p) => { + product.sales12m.insert(periode.clone(), p.qty); + product.stock12m.insert(periode.clone(), p.stock); + product.receptions12m.insert(periode.clone(), p.receptions); + product.total_quantite += p.qty; + product.total_ca += p.ca; + product.total_marge += p.marge; + last_stock = p.stock; + } + None => { + product.sales12m.insert(periode.clone(), 0.0); + product.stock12m.insert(periode.clone(), last_stock); // carry-forward + product.receptions12m.insert(periode.clone(), 0.0); + } + } + } + + product.taux_marge = if product.total_ca > 0.0 { (product.total_marge / product.total_ca) * 100.0 } else { 0.0 }; + + // Breakdowns par site pour le switch client-side + if let Some(site_map) = mensuel_by_site.get(codein) { + let mut sales_by_store = BTreeMap::new(); + let mut stock_by_store = BTreeMap::new(); + let mut receptions_by_store = BTreeMap::new(); + let mut ca_by_store = BTreeMap::new(); + let mut quantite_by_store = BTreeMap::new(); + let mut marge_by_store = BTreeMap::new(); + for (site, site_by_period) in site_map { + let (mut sales, mut stock, mut receptions) = (BTreeMap::new(), BTreeMap::new(), BTreeMap::new()); + let (mut s_qty, mut s_ca, mut s_marge) = (0.0, 0.0, 0.0); + let mut last_site_stock = 0.0; + for periode in &sorted_periods { + let sp = site_by_period.get(periode); + sales.insert(periode.clone(), sp.map(|p| p.qty).unwrap_or(0.0)); + receptions.insert(periode.clone(), sp.map(|p| p.receptions).unwrap_or(0.0)); + match sp { + Some(p) => { + stock.insert(periode.clone(), p.stock); + last_site_stock = p.stock; + } + None => { + stock.insert(periode.clone(), last_site_stock); // carry-forward + } + } + s_qty += sp.map(|p| p.qty).unwrap_or(0.0); + s_ca += sp.map(|p| p.ca).unwrap_or(0.0); + s_marge += sp.map(|p| p.marge).unwrap_or(0.0); + } + sales_by_store.insert(site.clone(), sales); + stock_by_store.insert(site.clone(), stock); + receptions_by_store.insert(site.clone(), receptions); + quantite_by_store.insert(site.clone(), s_qty); + ca_by_store.insert(site.clone(), s_ca); + marge_by_store.insert(site.clone(), s_marge); + } + + // Le stock TOTAL se recalcule ici, à partir des séries par site. + // + // Il était sommé plus haut depuis les lignes mensuelles brutes, ce + // qui perdait tout magasin SANS MOUVEMENT dans le mois : le stock + // est un niveau, pas un flux — un magasin qui n'a rien vendu ni reçu + // détient toujours sa marchandise. Les séries par site, elles, sont + // reportées d'un mois sur l'autre : leur somme est la bonne. + if !stock_by_store.is_empty() { + for periode in &sorted_periods { + let total: f64 = stock_by_store.values().map(|s: &BTreeMap| s.get(periode).copied().unwrap_or(0.0)).sum(); + product.stock12m.insert(periode.clone(), total); + } + } + product.sales12m_by_store = Some(sales_by_store); + product.stock12m_by_store = Some(stock_by_store); + product.receptions12m_by_store = Some(receptions_by_store); + product.ca_by_store = Some(ca_by_store); + product.quantite_by_store = Some(quantite_by_store); + product.marge_by_store = Some(marge_by_store); + } + + // workingStores : sites avec au moins 1 vente sur les 12 derniers mois + if let Some(sites) = store_months.get(codein) { + product.working_stores = sites.iter().cloned().collect(); + } + + // Commandes en cours + if let Some(&cmd) = commandes_map.get(codein) { + if cmd != 0.0 && !cmd.is_nan() { + product.commandes_en_cours = Some(cmd); + } + } + } + + // ─── Phase 6 : Gammes (saison active) ──────────────────────────────── + for (codein, gamme) in &gamme_map { + let Some(&pi) = index.get(codein) else { continue }; + products[pi].code_gamme_init = Some(gamme.clone()); + products[pi].code_gamme = Some(gamme.clone()); + } + + // ─── Phase 6b : Nomenclature (famille / sous-famille) ──────────────── + for (codein, nom) in &nom_map { + let Some(&pi) = index.get(codein) else { continue }; + let p = &mut products[pi]; + p.code3 = nom.code3.clone(); + p.libelle3 = nom.libelle3.clone(); + p.code2 = nom.code2.clone().unwrap_or_default(); + p.libelle_niveau2 = nom.libelle2.clone().unwrap_or_default(); + p.code1 = nom.code1.clone().unwrap_or_default(); + p.libelle_niveau1 = nom.libelle1nom.clone().unwrap_or_default(); + } + + // ─── Phase 7 : Stock temps réel (cube_stock) ───────────────────────── + for (codein, stocks) in &stock_map { + let Some(&pi) = index.get(codein) else { continue }; + let product = &mut products[pi]; + let somme = |f: fn(&PgStockRow) -> Option| stocks.iter().map(|r| or0(f(r).unwrap_or(f64::NAN))).sum::(); + product.stock_actuel = Some(somme(|r| r.stockdispo)); + product.stock_total = Some(somme(|r| r.qte)); + product.stock_valeur = Some(somme(|r| r.valstock)); + + if let Some(prmp) = stocks.iter().filter_map(|r| r.prmp).find(|p| *p > 0.0) { + product.pa = Some(prmp); + product.prix_achat = Some(prmp); + } + + let plus_recente = |dates: Vec<&String>| -> String { + dates.into_iter().fold(String::new(), |best, d| if !d.is_empty() && *d > best { d.clone() } else { best }) + }; + let dern_vente = plus_recente(stocks.iter().filter_map(|r| r.dernierevente.as_ref()).collect()); + if !dern_vente.is_empty() { + product.derniere_vente = Some(dern_vente); + } + let dern_liv = plus_recente(stocks.iter().filter_map(|r| r.dernierereception.as_ref()).collect()); + if !dern_liv.is_empty() { + product.derniere_livraison = Some(dern_liv); + } + + // Map store-specific delivery dates + let mut par_site = BTreeMap::new(); + for r in stocks { + if let Some(d) = r.dernierereception.as_ref().filter(|d| !d.is_empty() && !r.site.is_empty()) { + par_site.insert(r.site.clone(), d.clone()); + } + } + product.derniere_livraison_by_store = Some(par_site); + } + + // ─── Phase 7b : Prix de vente (cube_pv, sinon le PV du cube de stock) ─ + // + // Le vrai prix vit dans `cube_pv`, par article ET par site. On garde le + // détail par magasin — deux magasins peuvent ne pas pratiquer le même prix, + // et l'afficher vaut mieux que le moyenner. + for product in products.iter_mut() { + match prix_vente_map.get(&product.codein).filter(|m| !m.is_empty()) { + Some(par_site) => product.prix_vente_by_store = Some(par_site.clone()), + None => { + // Filet de secours : le cube de stock porte le même PV, mais + // seulement pour les articles qui ont une ligne de stock. + let mut depuis_stock = BTreeMap::new(); + for s in stock_map.get(&product.codein).map(|v| v.as_slice()).unwrap_or_default() { + let pv = s.pv.unwrap_or(f64::NAN); + let site = s.site.trim(); + if !site.is_empty() && pv.is_finite() && pv > 0.0 { + depuis_stock.insert(site.to_string(), pv); + } + } + if !depuis_stock.is_empty() { + product.prix_vente_by_store = Some(depuis_stock); + } + } + } + // Le plus élevé quand les magasins divergent : une moyenne inventerait + // un prix qu'aucune caisse ne pratique. L'écart est signalé à l'écran. + if let Some(prix) = product.prix_vente_by_store.as_ref().filter(|m| !m.is_empty()) { + product.prix_vente = Some(js_max(prix.values().copied())); + } + } + + // ─── Phase 8 : Données réseau Qlik (CA / Qté / nb magasins par code centrale) ── + enrich_with_network_metrics(&mut products, network.as_ref()); + + // ─── Phase 9 : Restaurer gammes depuis dernier snapshot ────────────── + // La colonne Gamme (codeGamme) conserve la valeur du snapshot telle quelle. + // La colonne INIT (codeGammeInit) reste, elle, l'état LIVE du serveur + // rechargé en Phase 6 — jamais écrasée ici. + apply_snapshot_changes(&mut products, snapshot_changes.as_ref()); + + // ─── Phase 10 : Filtrer gamme Y sans ventes ────────────────────────── + let tous = products.len(); + let mut rows: Vec = + products.into_iter().filter(|p| p.code_gamme.as_deref() != Some("Y") || p.total_quantite > 0.0).collect(); + reconcile_selected_store_from_mensuel_api(&mut rows, magasin, date_debut, date_fin, &sorted_periods).await; + let exclus_y = tous - rows.len(); + tracing::info!( + "[getProductRows] {} produits ({exclus_y} gamme Y sans ventes exclus), {} avec ventes", + rows.len(), + mensuel_by_codein.len() + ); + + Ok(rows) +} + +/// Articles complétés par l'API FF, au plus, lors d'un changement de magasin +/// (`GRID_STORE_RECONCILE_MAX`, 300 par défaut). +fn plafond_complement() -> f64 { + match std::env::var("GRID_STORE_RECONCILE_MAX") { + Ok(v) => super::js::js_number_from_str(&v), + Err(_) => 300.0, + } +} + +/// `candidates.slice(0, plafond)`. +fn js_slice_end(len: usize, fin: f64) -> usize { + if fin.is_nan() { + return 0; + } + let fin = fin.trunc(); + if fin < 0.0 { + (len as f64 + fin).max(0.0) as usize + } else { + fin.min(len as f64) as usize + } +} + +/// Article candidat au complément, copié hors verrou. +struct Candidat { + codein: String, + libelle1: String, + codefou: String, + noid: Option, +} + +fn candidats(rows: &[ProductRow], magasin: &str) -> Vec { + store_patch_candidates(rows, magasin) + .into_iter() + .map(|r| Candidat { codein: r.codein.clone(), libelle1: r.libelle1.clone(), codefou: r.code_fournisseur.clone(), noid: r.noid }) + .collect() +} + +/// Interroge l'API FF pour les articles du magasin à compléter (cf. store-patch). +async fn fetch_store_patch_entries( + candidates: Vec, + magasin: &str, + date_debut: &str, + date_fin: &str, + periods: &[String], +) -> Vec { + if candidates.is_empty() { + return vec![]; + } + + // Ce rattrapage fait UNE requête HTTP par article : sur un gros fournisseur, + // les candidats se comptent par milliers et le changement de magasin se fige + // plusieurs minutes. On le borne, et on dit ce qui a été laissé de côté + // plutôt que de tronquer en silence. + let nb = candidates.len(); + let retenus = &candidates[..js_slice_end(nb, plafond_complement())]; + if nb > retenus.len() { + tracing::warn!( + "[getProductRows] rattrapage magasin {magasin} borné à {} articles sur {nb} : les autres gardent la valeur SQL.", + retenus.len() + ); + } + + let articles: Vec = retenus + .iter() + .map(|c| FfArticle { + codein: c.codein.clone(), + libelle1: c.libelle1.clone(), + codefou: c.codefou.clone(), + noid: c.noid.filter(|n| n.is_finite()).map(|n| n as i64), + ..Default::default() + }) + .collect(); + let mensuel_map = get_mensuel_by_articles(&articles, date_debut, date_fin, 25).await; + let codes: Vec = retenus.iter().map(|c| c.codein.clone()).collect(); + build_store_patch_entries(&codes, magasin, periods, &mensuel_map) +} + +async fn reconcile_selected_store_from_mensuel_api( + rows: &mut [ProductRow], + magasin: &str, + date_debut: &str, + date_fin: &str, + sorted_periods: &[String], +) { + if magasin == "TOTAL" { + return; + } + let entries = fetch_store_patch_entries(candidats(rows, magasin), magasin, date_debut, date_fin, sorted_periods).await; + let fixed = apply_store_patch_in_place(rows, magasin, sorted_periods, &entries); + if fixed > 0 { + tracing::info!("[getProductRows] {fixed} produits corrigés via API mensuelle pour magasin {magasin}"); + } +} + +/// Complément de l'API FF pour un magasin, calculé sur les lignes « tous +/// magasins » (jamais modifiées ici). La Grille l'applique à son arrivée, sans +/// recharger les lignes. Les demandes simultanées partagent le même calcul, et +/// un échec n'est jamais gardé. +pub async fn get_store_patch(code_fournisseur: &str, magasin: &str) -> anyhow::Result> { + let rows = match rows_cache::peek_fresh(&rows_cache::grid_rows_cache_key(code_fournisseur, Some("TOTAL"))) { + Some(r) => r, + None => get_product_rows(GetProductRowsInput::total(code_fournisseur, false)).await?, + }; + + let patch: StorePatchFuture = { + let mut table = rows.store_patches.lock(); + match table.get(magasin) { + Some(p) => p.clone(), + None => { + let periods = last12_periods(); + let range = build_last12_months_range(); + let candidates = candidats(&rows.rows.read(), magasin); + let magasin_s = magasin.to_string(); + let (tx, rx) = tokio::sync::oneshot::channel(); + tokio::spawn(async move { + let entries = fetch_store_patch_entries(candidates, &magasin_s, &range.date_debut, &range.date_fin, &periods).await; + let _ = tx.send(Ok(Arc::new(StorePatch { magasin: magasin_s, periods, entries }))); + }); + let en_cours: StorePatchFuture = + async move { rx.await.unwrap_or_else(|_| Err("complément interrompu".to_string())) }.boxed().shared(); + table.insert(magasin.to_string(), en_cours.clone()); + en_cours + } + } + }; + match patch.clone().await { + Ok(p) => Ok(p), + Err(e) => { + let mut table = rows.store_patches.lock(); + if table.get(magasin).map(|p| p.ptr_eq(&patch)).unwrap_or(false) { + table.remove(magasin); + } + Err(anyhow::anyhow!(e)) + } + } +} + +/// Extrait la série « nombre de magasins vendeurs » du détail mensuel complet. +/// +/// Un mois PRÉSENT dans le détail est un mois extrait : s'il n'a pas de `nbMag`, +/// c'est qu'aucun magasin n'a vendu, donc zéro. Renvoie `None` plutôt qu'un objet +/// vide pour que l'UI distingue « pas de donnée » de « zéro magasin ». +fn nb_mag_par_mois(m: &NetworkMetricCached) -> Option> { + let detail = m.metrics_by_month.as_ref()?; + let par_mois: BTreeMap = + detail.iter().map(|(mois, v)| (mois.clone(), v.nb_mag.filter(|n| n.is_finite()).unwrap_or(0.0))).collect(); + (!par_mois.is_empty()).then_some(par_mois) +} + +/// Phase 9 : la colonne Gamme (codeGamme) reprend la valeur du snapshot ; +/// codeGammeInit reste figé (état serveur). +pub fn apply_snapshot_changes(rows: &mut [ProductRow], changes: Option<&Map>) { + let Some(changes) = changes else { return }; + for row in rows.iter_mut() { + let after = changes.get(&row.codein).and_then(|c| c.get("after")); + if super::js::js_truthy(after) { + row.code_gamme = Some(super::js::js_string(after)); + } + } +} + +/// Dernières gammes enregistrées pour ce fournisseur (Phase 9), ou `None`. +pub async fn load_latest_snapshot_changes(code_fournisseur: &str) -> Option> { + let res = async { + let pool = crate::db::pool().map_err(|e| anyhow::anyhow!(e.message))?; + let row = sqlx::query("select changes from session_snapshots where code_fournisseur = $1 order by created_at desc limit 1") + .bind(code_fournisseur) + .fetch_optional(&pool) + .await?; + anyhow::Ok(row.map(|r| r.try_get::("changes")).transpose()?) + } + .await; + match res { + Ok(Some(Value::Object(m))) => Some(m), + Ok(_) => None, + Err(e) => { + tracing::error!("[getProductRows] Snapshot restore error: {e:#}"); + None + } + } +} + +/// Lit les métriques réseau Qlik (cache qlik_network_metrics) des codes centraux +/// donnés. `None` en cas d'erreur : la grille s'affiche alors sans données réseau. +async fn fetch_network_metrics(codes_centrale: &[String]) -> Option> { + let mut vus = BTreeSet::new(); + let codes: Vec = codes_centrale.iter().filter(|c| !c.is_empty() && vus.insert(c.as_str())).cloned().collect(); + if codes.is_empty() { + return None; + } + match get_network_metrics_by_code_centrale(&codes).await { + Ok(m) => Some(m), + Err(e) => { + tracing::error!("[getProductRows] enrichWithNetworkMetrics error: {e:#}"); + None + } + } +} + +/// Enrichit les produits avec les métriques réseau Qlik, jointes par code centrale. +/// Dégradation propre si la sync Qlik n'a jamais tourné ou si le code centrale +/// n'est pas encore disponible. +pub fn enrich_with_network_metrics(products: &mut [ProductRow], metrics: Option<&HashMap>) { + let Some(metrics) = metrics.filter(|m| !m.is_empty()) else { return }; + for product in products.iter_mut() { + let Some(m) = product.code_centrale.as_deref().filter(|c| !c.is_empty()).and_then(|c| metrics.get(c)) else { continue }; + product.ca_reseau = Some(m.ca_reseau); + product.qte_reseau = Some(m.qte_reseau); + product.nb_magasins_reseau = Some(m.nb_magasins_reseau); + product.ca_par_magasin_reseau = Some(m.ca_par_magasin_reseau); + product.marge_pct_reseau = Some(m.marge_pct_reseau); + product.taux_presence_reseau = Some(m.nb_magasins_reseau / NB_MAGASINS_RESEAU); + product.qte_reseau_by_month = Some(m.qte_by_month.clone()); + product.nb_mag_reseau_by_month = Some(nb_mag_par_mois(m)); + product.network_fetched_at = m.fetched_at.clone(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn slice_et_max_comme_js() { + assert_eq!(js_slice_end(10, 300.0), 10); + assert_eq!(js_slice_end(10, 3.7), 3); + assert_eq!(js_slice_end(10, -2.0), 8); + assert_eq!(js_slice_end(10, f64::NAN), 0); + assert_eq!(js_max([1.0, 3.0, 2.0].into_iter()), 3.0); + assert!(js_max([1.0, f64::NAN].into_iter()).is_nan()); + } + + #[test] + fn report_des_gammes_du_snapshot() { + let mut rows = vec![ + ProductRow { codein: "1".into(), code_gamme: Some("B".into()), ..Default::default() }, + ProductRow { codein: "2".into(), code_gamme: Some("B".into()), ..Default::default() }, + ProductRow { codein: "3".into(), code_gamme: Some("B".into()), ..Default::default() }, + ]; + let changes = serde_json::json!({ "1": { "before": "B", "after": "Z" }, "2": { "before": "B", "after": "" } }); + apply_snapshot_changes(&mut rows, changes.as_object()); + let g: Vec<_> = rows.iter().map(|r| r.code_gamme.as_deref().unwrap()).collect(); + assert_eq!(g, vec!["Z", "B", "B"]); + } +} diff --git a/server/src/grid/store.rs b/server/src/grid/store.rs index 7c37fe6..909e7b7 100644 --- a/server/src/grid/store.rs +++ b/server/src/grid/store.rs @@ -159,16 +159,40 @@ pub async fn upsert_grid_rows(code_fournisseur: &str, rows: &[ProductRow]) -> Ap if code_fournisseur.is_empty() || rows.is_empty() { return Ok(0); } + let values: Vec<&ProductRow> = rows.iter().filter(|r| !r.codein.is_empty()).collect(); + upsert_lots(code_fournisseur, values.len(), |plage| values[plage].iter().map(|r| (*r).clone()).collect()).await +} + +/// Même écriture depuis les lignes partagées du cache mémoire : chaque lot est +/// copié sous verrou, puis écrit sans le tenir (pas de copie intégrale d'un +/// fournisseur de 130 000 articles). +pub async fn upsert_grid_rows_shared(code_fournisseur: &str, rows: &super::rows_cache::GridRows) -> AppResult { + let indices: Vec = { + let lignes = rows.rows.read(); + if code_fournisseur.is_empty() || lignes.is_empty() { + return Ok(0); + } + lignes.iter().enumerate().filter(|(_, r)| !r.codein.is_empty()).map(|(i, _)| i).collect() + }; + upsert_lots(code_fournisseur, indices.len(), |plage| { + let lignes = rows.rows.read(); + indices[plage].iter().filter_map(|i| lignes.get(*i).cloned()).collect() + }) + .await +} + +/// Écrit `nb` lignes par lots de [`CHUNK_UPSERT`] (`lot(plage)` fournit les lignes +/// d'une plage), puis purge les lignes du fournisseur restées sur un calcul antérieur. +async fn upsert_lots(code_fournisseur: &str, nb: usize, lot: impl Fn(std::ops::Range) -> Vec) -> AppResult { + if nb == 0 { + return Ok(0); + } let pool = db::pool()?; let computed_at = maintenant_utc_ms(); - let values: Vec<&ProductRow> = rows.iter().filter(|r| !r.codein.is_empty()).collect(); - if values.is_empty() { - return Ok(0); - } - let mut written = 0usize; - for batch in values.chunks(CHUNK_UPSERT) { + for debut in (0..nb).step_by(CHUNK_UPSERT) { + let batch = lot(debut..(debut + CHUNK_UPSERT).min(nb)); let mut qb: QueryBuilder = QueryBuilder::new( "insert into grid_rows (code_fournisseur, codein, nom_fournisseur, libelle1, gtin, reference, \ code_centrale, code1, code2, code3, code_gamme, code_gamme_init, total_ca, total_quantite, \ @@ -177,7 +201,7 @@ pub async fn upsert_grid_rows(code_fournisseur: &str, rows: &[ProductRow]) -> Ap ); let mut erreur_payload = None; qb.push_values(batch.iter(), |mut b, r| { - let mut payload = (*r).clone(); + let mut payload = r.clone(); payload.payload_version = Some(GRID_PAYLOAD_VERSION); let payload = serde_json::to_value(&payload).unwrap_or_else(|e| { erreur_payload = Some(e.to_string()); diff --git a/server/src/grid/store_patch.rs b/server/src/grid/store_patch.rs index bb94270..9cb3ecf 100644 --- a/server/src/grid/store_patch.rs +++ b/server/src/grid/store_patch.rs @@ -4,13 +4,77 @@ //! Le calcul SQL de la Grille peut manquer les ventes d'un magasin pour certains //! articles (quantité à 0 pour ce magasin). L'API FF les connaît : on en tire, //! pour ces articles seulement, les séries et totaux du magasin. -//! -//! `buildStorePatchEntries` (qui lit les entrées mensuelles de l'API FF) viendra -//! avec le calcul de la grille (phase 2). use super::types::{ProductRow, StorePatchEntry}; +use crate::ff::api::FfMensuelEntry; +use crate::ff::js::js_number_or0; use std::collections::HashMap; +/// Chiffres du magasin tirés de l'API, pour les articles où elle montre des ventes. +/// +/// `retenus` : codes article des candidats retenus (cf. [`store_patch_candidates`]). +pub fn build_store_patch_entries( + retenus: &[String], + magasin: &str, + periods: &[String], + mensuel: &HashMap>, +) -> Vec { + struct Valeurs { + qty: f64, + ca: f64, + marge: f64, + stock: f64, + receptions: f64, + } + let mut resultat = Vec::new(); + for codein in retenus { + let entries: Vec<&FfMensuelEntry> = + mensuel.get(codein).map(|v| v.iter().filter(|e| e.site == magasin).collect()).unwrap_or_default(); + if entries.is_empty() { + continue; + } + + let mut by_period: HashMap = HashMap::new(); + for entry in entries { + let period = entry.mois.replacen('-', "", 1); + if !periods.contains(&period) { + continue; + } + let ventes = entry.ventes.as_ref(); + // qte_vendue / ca_ht sont NÉGATIFS côté API (ventes nettes) : on + // les nie au lieu de Math.abs pour que les retours restent déduits. + let qty = -ventes.map(|v| js_number_or0(&v.qte_vendue)).unwrap_or(0.0); + let ca = -ventes.map(|v| js_number_or0(&v.ca_ht)).unwrap_or(0.0); + let marge = ventes.map(|v| js_number_or0(&v.marge)).unwrap_or(0.0); + let stock = js_number_or0(&entry.stock_fin_mois); + let receptions = entry.receptions.as_ref().map(|r| js_number_or0(&r.qte_recue)).unwrap_or(0.0); + by_period.insert(period, Valeurs { qty, ca, marge, stock, receptions }); + } + + let api_qty: f64 = by_period.values().map(|v| v.qty).sum(); + if api_qty == 0.0 { + continue; + } + + let mut e = StorePatchEntry { codein: codein.clone(), ..Default::default() }; + let mut last_stock = 0.0; + for period in periods { + let value = by_period.get(period); + e.sales.insert(period.clone(), value.map(|v| v.qty).unwrap_or(0.0)); + e.receptions.insert(period.clone(), value.map(|v| v.receptions).unwrap_or(0.0)); + if let Some(v) = value { + last_stock = v.stock; + } + e.stock.insert(period.clone(), last_stock); + e.quantite += value.map(|v| v.qty).unwrap_or(0.0); + e.ca += value.map(|v| v.ca).unwrap_or(0.0); + e.marge += value.map(|v| v.marge).unwrap_or(0.0); + } + resultat.push(e); + } + resultat +} + /// Articles à compléter : connus de FF (noid) et sans aucune vente SQL dans ce magasin. pub fn store_patch_candidates<'a>(rows: &'a [ProductRow], magasin: &str) -> Vec<&'a ProductRow> { rows.iter() @@ -77,6 +141,31 @@ pub fn apply_store_patch_in_place(rows: &mut [ProductRow], magasin: &str, period mod tests { use super::*; + #[test] + fn complement_depuis_l_api() { + let periods = vec!["202601".to_string(), "202602".to_string(), "202603".to_string()]; + let entree = |mois: &str, site: &str, qte: &str| -> FfMensuelEntry { + serde_json::from_value(serde_json::json!({ + "mois": mois, "site": site, "stock_fin_mois": "5.000", + "ventes": { "qte_vendue": qte, "ca_ht": "-26.66", "marge": "12.76" }, "receptions": null + })) + .unwrap() + }; + let mut mensuel = HashMap::new(); + mensuel.insert( + "A1".to_string(), + vec![entree("2026-01", "292", "-2.000"), entree("2026-02", "579", "-9"), entree("2025-01", "292", "-4")], + ); + mensuel.insert("A2".to_string(), vec![entree("2026-01", "292", "0")]); + let e = build_store_patch_entries(&["A1".into(), "A2".into(), "A3".into()], "292", &periods, &mensuel); + assert_eq!(e.len(), 1, "A2 sans vente, A3 sans donnée"); + assert_eq!(e[0].quantite, 2.0); + assert!((e[0].ca - 26.66).abs() < 1e-9); + assert_eq!(e[0].sales["202602"], 0.0); + assert_eq!(e[0].stock["202603"], 5.0, "stock reporté"); + assert_eq!(serde_json::to_value(&e[0]).unwrap()["receptions"], serde_json::json!({ "202601": 0, "202602": 0, "202603": 0 })); + } + #[test] fn complement_reporte_sur_les_totaux() { let periods = vec!["202601".to_string(), "202602".to_string()]; diff --git a/server/src/grid/tests_db.rs b/server/src/grid/tests_db.rs index 61583b7..6d4696c 100644 --- a/server/src/grid/tests_db.rs +++ b/server/src/grid/tests_db.rs @@ -60,6 +60,8 @@ async fn grid_rows_snapshots_gammes_a_valider() { std::fs::create_dir_all(&dir).unwrap(); std::env::set_var("DATA_DIR", &dir); std::env::set_var("DATABASE_URL", &url); + // Une autre base a pu servir à un test précédent du même processus. + crate::db::refresh(); let pool = crate::db::pool().unwrap(); sqlx::raw_sql(include_str!("../../sql/init.sql")).execute(&pool).await.unwrap(); nettoyer(&pool).await; diff --git a/server/src/grid/tests_ff.rs b/server/src/grid/tests_ff.rs new file mode 100644 index 0000000..14076d1 --- /dev/null +++ b/server/src/grid/tests_ff.rs @@ -0,0 +1,302 @@ +//! Test d'intégration du calcul de la grille (phase 2) sur une base de test +//! contenant les tables FF (`ff_fixture`), avec une fausse API FF Nancy. +//! +//! ```sh +//! GRID_FF_TEST_DATABASE_URL=postgres://postgres:postgres@localhost/ff_fixture cargo test grid::tests_ff -- --nocapture +//! ``` +//! +//! Écrit dans `grid_rows`, `session_snapshots`, `gammes_a_valider`, `users` et +//! `api_keys` de la base de test (nettoyés à la fin). + +use super::enrich::enrich_rows; +use super::months; +use super::product_rows::*; +use super::rows_cache; +use super::store::*; +use crate::auth::Claims; +use axum::body::Body; +use axum::extract::Path; +use axum::http::Request; +use axum::routing::get; +use axum::{Json, Router}; +use serde_json::{json, Value}; +use tower::ServiceExt; + +const CLE_API: &str = "cf_cle-de-test-grille-phase-2"; + +/// Fausse API FF : seul l'historique mensuel sert au complément par magasin. +async fn fausse_api(mois: String) -> String { + let app = Router::new().route( + "/api/articles/{id}/mensuel", + get(move |Path(id): Path| { + let mois = mois.clone(); + async move { + Json(json!({ "data": [ + { "mois": mois, "site": "579", "stock_fin_mois": "7.000", + "ventes": { "qte_vendue": "-3.000", "ca_ht": "-30.00", "marge": "9.5" }, + "receptions": { "qte_recue": "2" }, "id": id }, + { "mois": mois, "site": "292", "stock_fin_mois": "1", + "ventes": { "qte_vendue": "-100", "ca_ht": "-1", "marge": "0" } } + ] })) + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + format!("http://{addr}") +} + +async fn appel(req: Request) -> (u16, Vec) { + let res = super::router().oneshot(req).await.unwrap(); + let status = res.status().as_u16(); + (status, axum::body::to_bytes(res.into_body(), usize::MAX).await.unwrap().to_vec()) +} + +fn get_cle(uri: &str) -> Request { + Request::builder().uri(uri).header("x-api-key", CLE_API).body(Body::empty()).unwrap() +} + +fn json_de(b: &[u8]) -> Value { + serde_json::from_slice(b).unwrap_or_else(|e| panic!("JSON illisible ({e}) : {}", String::from_utf8_lossy(b))) +} + +async fn nettoyer(pool: &sqlx::PgPool) { + for sql in [ + "delete from grid_rows where code_fournisseur in ('F001', 'F002', 'F404')", + "delete from session_snapshots where code_fournisseur in ('F001', 'F002')", + "delete from gammes_a_valider where code_fournisseur in ('F001', 'F002')", + "delete from api_keys where name = 'test-grille'", + "delete from users where username = 'test-grille'", + ] { + sqlx::query(sql).execute(pool).await.unwrap(); + } +} + +#[tokio::test] +async fn calcul_de_la_grille_et_routes() { + let Ok(url) = std::env::var("GRID_FF_TEST_DATABASE_URL") else { + eprintln!("GRID_FF_TEST_DATABASE_URL absent : test d'intégration du calcul ignoré."); + return; + }; + let _verrou = rows_cache::TEST_LOCK.lock(); + let dir = std::env::temp_dir().join(format!("grid-ff-tests-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + std::env::set_var("DATA_DIR", &dir); + std::env::set_var("DATABASE_URL", &url); + std::env::set_var("AUTH_SECRET", "secret-de-test"); + crate::db::refresh(); + let periods = months::get_last12_months(months::today()); + let dernier_mois = months::ff_month_to_qlik(periods.last().unwrap()); + std::env::set_var("FF_API_BASE_URL", fausse_api(dernier_mois).await); + crate::ff::api::reset_ff_api_base_cache(); + + let pool = crate::db::pool().unwrap(); + sqlx::raw_sql(include_str!("../../sql/init.sql")).execute(&pool).await.unwrap(); + nettoyer(&pool).await; + rows_cache::invalidate_grid_rows_cache(None); + + // ── calcul complet ───────────────────────────────────────────────────── + let rows = get_product_rows(GetProductRowsInput::total("F001", false)).await.unwrap(); + let lignes = rows.snapshot(); + println!("── lignes calculées\n{}", serde_json::to_string_pretty(&lignes).unwrap()); + assert!(!lignes.is_empty()); + // `articles.codein` est un char(10) dans la base de test : codes complétés d'espaces. + let tapis = lignes.iter().find(|r| r.codein.trim() == "100001").expect("article 100001"); + let code_tapis = tapis.codein.clone(); + let tapis_url = urlencoding::encode(&code_tapis).into_owned(); + assert_eq!(tapis.sales12m.len(), 12); + assert_eq!(tapis.code_centrale.as_deref(), Some("10000111111")); + assert_eq!(tapis.ca_reseau, Some(50000.0), "métriques réseau du cache Qlik"); + assert_eq!(tapis.taux_presence_reseau, Some(1.0)); + assert!(tapis.sales12m_by_store.as_ref().unwrap().contains_key("292")); + assert!(tapis.working_stores.contains(&"292".to_string())); + assert_eq!(tapis.stock_actuel, Some(4.0), "5 + -1 sur les deux sites"); + assert_eq!(tapis.prix_vente, Some(12.9), "le PV le plus élevé des deux magasins"); + assert!(tapis.total_quantite > 0.0 && tapis.total_ca > 0.0); + + // Persistance (lancée en arrière-plan) puis instantané. + attendre_persistance("F001").await; + let fraicheur = get_grid_freshness("F001").await.unwrap().expect("instantané écrit"); + assert_eq!(fraicheur.row_count as usize, lignes.len()); + + // Cache : même tableau partagé. + let encore = get_product_rows(GetProductRowsInput::total("F001", false)).await.unwrap(); + assert!(std::sync::Arc::ptr_eq(&rows, &encore)); + + // Instantané resservi après invalidation du cache, gammes du snapshot reportées. + sqlx::query("insert into session_snapshots (code_fournisseur, magasin, changes, type) values ('F001', 'TOTAL', $1, 'snapshot')") + .bind(json!({ code_tapis.clone(): { "before": null, "after": "Z" } })) + .execute(&pool) + .await + .unwrap(); + rows_cache::invalidate_grid_rows_cache(Some("F001")); + let servies = get_product_rows(GetProductRowsInput::total("F001", false)).await.unwrap(); + assert!(!std::sync::Arc::ptr_eq(&rows, &servies)); + assert_eq!(servies.len(), lignes.len()); + let s = servies.snapshot(); + assert_eq!(s.iter().find(|r| r.codein == code_tapis).unwrap().code_gamme.as_deref(), Some("Z")); + assert!(s.iter().all(|r| r.payload_version == Some(GRID_PAYLOAD_VERSION)), "lignes issues de grid_rows"); + + // Synchro : recalcul forcé, nombre de lignes. + assert_eq!(get_product_rows_for_sync("F001").await.unwrap(), lignes.len()); + + // ── complément par magasin (fausse API FF) ───────────────────────────── + let patch = get_store_patch("F001", "579").await.unwrap(); + println!("── complément 579\n{}", serde_json::to_string_pretty(&*patch).unwrap()); + assert_eq!(patch.magasin, "579"); + assert_eq!(patch.periods, periods); + assert!(!patch.entries.is_empty()); + assert!(patch.entries.iter().all(|e| e.quantite == 3.0 && e.ca == 30.0)); + let meme = get_store_patch("F001", "579").await.unwrap(); + assert!(std::sync::Arc::ptr_eq(&patch, &meme), "complément partagé"); + + // Vue d'un magasin : rattrapage appliqué aux lignes. + let vue_579 = + get_product_rows(GetProductRowsInput { code_fournisseur: "F001".into(), magasin: Some("579".into()), force_refresh: false }) + .await + .unwrap(); + assert!(vue_579.snapshot().iter().any(|r| r.quantite_by_store.as_ref().and_then(|q| q.get("579")).copied() == Some(3.0))); + + // ── enrichissement /api/v1 ───────────────────────────────────────────── + let enrichies = + enrich_rows(query_grid_rows(&GridQuery { code_fournisseur: Some("F001".into()), ..Default::default() }).await.unwrap().rows).await; + let e = enrichies.iter().find(|r| r.row.codein == code_tapis).unwrap(); + assert_eq!(e.network.as_ref().map(|n| n.ca_reseau), Some(50000.0)); + let v = serde_json::to_value(e).unwrap(); + assert!(v.get("codeGammeServeur").is_some() && v.get("gammeAValider").is_some() && v.get("trend").is_some()); + assert_eq!(v["codein"], json!(code_tapis)); + + // ── routes ───────────────────────────────────────────────────────────── + let (id_user,): (i32,) = + sqlx::query_as("insert into users (username, password_hash, role) values ('test-grille', 'x', 'admin') returning id") + .fetch_one(&pool) + .await + .unwrap(); + sqlx::query("insert into api_keys (name, key_prefix, key_hash, role) values ('test-grille', 'cf_cle', $1, 'user')") + .bind(crate::auth::api_key::hash_api_key(CLE_API)) + .execute(&pool) + .await + .unwrap(); + let claims = + Claims { id: id_user.to_string(), name: "test-grille".into(), role: "admin".into(), exp: chrono::Utc::now().timestamp() + 3600 }; + let cookie = format!("{}={}", crate::auth::session::COOKIE_NAME, crate::auth::session::encode(&claims)); + let get_session = |uri: &str| Request::builder().uri(uri).header("cookie", cookie.clone()).body(Body::empty()).unwrap(); + + // /api/grid/rows (NDJSON) + let (st, corps) = appel(get_session("/api/grid/rows?fournisseur=F001&magasin=TOTAL&chunkSize=1")).await; + assert_eq!(st, 200); + let messages: Vec = String::from_utf8(corps).unwrap().lines().map(|l| json_de(l.as_bytes())).collect(); + assert_eq!(messages[0], json!({ "type": "start", "loaded": 0, "total": null })); + let chunks: Vec<&Value> = messages.iter().filter(|m| m["type"] == "chunk").collect(); + // chunkSize=1 est relevé au minimum de 50 : un seul lot ici. + assert_eq!(chunks.len(), 1); + assert_eq!(chunks[0]["rows"].as_array().unwrap().len(), lignes.len()); + assert_eq!(chunks[0]["loaded"], lignes.len()); + assert_eq!(messages.last().unwrap(), &json!({ "type": "done", "loaded": lignes.len(), "total": lignes.len() })); + let (st, _) = appel(get_session("/api/grid/rows")).await; + assert_eq!(st, 400); + let (st, _) = appel(Request::builder().uri("/api/grid/rows?fournisseur=F001").body(Body::empty()).unwrap()).await; + assert_eq!(st, 401); + + // /api/grid/rows/store-patch + let (st, corps) = appel(get_session("/api/grid/rows/store-patch?fournisseur=F001&magasin=579")).await; + assert_eq!(st, 200); + assert_eq!(json_de(&corps)["magasin"], "579"); + let (st, corps) = appel(get_session("/api/grid/rows/store-patch?fournisseur=F001&magasin=TOTAL")).await; + assert_eq!((st, json_de(&corps)), (400, json!({ "error": "Paramètres fournisseur et magasin requis." }))); + + // /api/v1/grid : diffusion complète, puis page + let (st, corps) = appel(get_cle("/api/v1/grid?fournisseur=F001")).await; + assert_eq!(st, 200); + let tout = json_de(&corps); + assert_eq!(tout["data"].as_array().unwrap().len(), lignes.len()); + assert_eq!(tout["meta"]["diffuse"], true); + assert_eq!(tout["meta"]["complet"], true); + assert_eq!(tout["pagination"]["total"], lignes.len()); + let (st, corps) = appel(get_cle("/api/v1/grid?fournisseur=F001&limit=1&fields=totalCa&sort=codein&order=asc")).await; + assert_eq!(st, 200); + let page = json_de(&corps); + assert_eq!(page["data"][0].as_object().unwrap().keys().collect::>(), vec!["totalCa", "codein"]); + assert_eq!(page["meta"]["complet"], json!(lignes.len() <= 1)); + if lignes.len() > 1 { + assert!(page["meta"]["avertissement"].as_str().unwrap().contains("page=2")); + } + let (st, corps) = appel(get_cle("/api/v1/grid?fournisseur=F002&compute=0")).await; + assert_eq!((st, json_de(&corps)["error"]["code"].clone()), (202, json!("not_ready"))); + let (st, corps) = appel(get_cle("/api/v1/grid?fournisseur=F002")).await; + let f002 = json_de(&corps); + assert_eq!(st, 200, "{f002}"); + assert_eq!(f002["meta"]["computedOnDemand"], true); + let (st, corps) = appel(get_cle("/api/v1/grid?fournisseur=F404")).await; + assert_eq!((st, json_de(&corps)["error"]["code"].clone()), (404, json!("not_found"))); + + // /api/v1/products/* + let (st, corps) = appel(get_cle("/api/v1/products/search?q=tapis&fields=libelle1")).await; + assert_eq!(st, 200); + let rech = json_de(&corps); + assert_eq!(rech["data"][0]["codein"], json!(code_tapis)); + assert_eq!(rech["meta"]["scope"], "tous fournisseurs"); + let (st, corps) = appel(get_cle(&format!("/api/v1/products/{tapis_url}?fournisseur=F001"))).await; + assert_eq!(st, 200); + let fiche = json_de(&corps); + assert_eq!(fiche["meta"]["hasNetwork"], true); + assert_eq!(fiche["data"]["network"]["caReseau"], 50000); + let (st, _) = appel(get_cle("/api/v1/products/999999")).await; + assert_eq!(st, 404); + + // Gammes : proposition à valider, puis lot avec un inconnu (tout ou rien). + let put = Request::builder() + .method("PUT") + .uri(format!("/api/v1/products/{tapis_url}/gamme")) + .header("x-api-key", CLE_API) + .header("content-type", "application/json") + .body(Body::from(r#"{"gamme":"A","fournisseur":"F001"}"#)) + .unwrap(); + let (st, corps) = appel(put).await; + let rep = json_de(&corps); + assert_eq!(st, 200, "{rep}"); + assert_eq!(rep["data"]["statut"], "a_valider"); + assert_eq!(rep["data"]["codeFournisseur"], "F001"); + let a_valider = super::gammes_a_valider::lister_gammes_a_valider("F001").await.unwrap(); + assert_eq!((a_valider[0].codein.as_str(), a_valider[0].auteur.as_deref()), (code_tapis.as_str(), Some("test-grille"))); + let post = |corps: &str| { + Request::builder().method("POST").uri("/api/v1/gammes").header("x-api-key", CLE_API).body(Body::from(corps.to_string())).unwrap() + }; + let (st, corps) = appel(post( + &json!({ "fournisseur": "F001", "changes": [{ "codein": code_tapis, "gamme": "B" }, { "codein": "XXX", "gamme": "A" }] }) + .to_string(), + )) + .await; + let rep = json_de(&corps); + assert_eq!(st, 404); + assert_eq!(rep["error"]["details"]["inconnus"], json!(["XXX"])); + let (st, corps) = appel(post("pas du json")).await; + assert_eq!( + (st, json_de(&corps)["error"]["message"].clone()), + (400, json!("Corps de requête JSON attendu (Content-Type: application/json).")) + ); + let (st, corps) = appel(post(&json!({ "fournisseur": "F001", "changes": [{ "codein": code_tapis, "gamme": "Z" }] }).to_string())).await; + let rep = json_de(&corps); + assert_eq!(st, 200, "{rep}"); + assert_eq!(rep["meta"]["demandes"], 1); + + // /api/v1/fournisseurs, /api/v1/network + let (st, corps) = appel(get_cle("/api/v1/fournisseurs")).await; + assert_eq!(st, 200); + let fous = json_de(&corps); + assert!(fous["data"].as_array().unwrap().iter().any(|f| f["code"] == "F001" && f["hasSnapshot"] == true)); + let (st, corps) = appel(get_cle("/api/v1/fournisseurs?withData=1&search=f00")).await; + assert_eq!(st, 200); + assert_eq!(json_de(&corps)["meta"]["scope"], "instantané uniquement"); + let (st, corps) = appel(get_cle("/api/v1/network/10000111111")).await; + assert_eq!(st, 200); + let net = json_de(&corps); + assert_eq!(net["data"].as_object().unwrap().keys().next().unwrap(), "codeCentrale"); + assert_eq!(net["meta"]["source"], "cache"); + let (st, _) = appel(get_cle("/api/v1/network/inconnu")).await; + assert_eq!(st, 404); + + nettoyer(&pool).await; + rows_cache::invalidate_grid_rows_cache(None); +}