diff --git a/README.md b/README.md index ed21264..e12980f 100644 --- a/README.md +++ b/README.md @@ -2,56 +2,60 @@ CollectFlow est une application de révision d'assortiment et d'analyse de performances des produits en point de vente, conçue pour consolider les données issues de plusieurs fournisseurs et magasins. +## Architecture + +| Partie | Technologie | Dossier | +|---|---|---| +| Serveur (API, base, FF Nancy, Qlik, synchro nocturne) | **Rust** — Axum, sqlx, chromiumoxide | `server/` | +| Interface | React 19 + Tailwind 4, compilée en fichiers statiques par Vite | `src/` | + +Un seul binaire Rust sert l'API JSON (`/api/...`, dont l'API publique `/api/v1`) et l'interface compilée. Plus de Node.js à l'exécution. + ## Prérequis -- **Node.js** 18+ (pour le développement local) -- **Docker** et **Docker Compose** (pour le déploiement conteneurisé) -- **PostgreSQL** 16+ +- **Docker** et **Docker Compose** (déploiement) +- **PostgreSQL** 16+ (base séparée, non fournie par le conteneur) +- Pour développer : **Rust** stable (`rustup`) et **Node.js** 22+ ## Lancer avec Docker -Le projet est configuré pour tourner dans un conteneur Docker optimisé (mode standalone de Next.js). -**Note :** Vous devez disposer d'une base de données PostgreSQL séparée (le conteneur ne lance que l'application web). +L'image est construite par GitHub Actions à chaque mise à jour de `main` et publiée sur `ghcr.io/r0m1k3/collectflow:latest`. Le `docker-compose.yml` la récupère à chaque démarrage (`pull_policy: always`) : **une nouvelle version s'installe au prochain `docker compose up -d`**, sans rien compiler sur le serveur. -### 1. Démarrer l'application -À la racine du projet, lancez : ```bash -npm run docker:build -# ou directement -docker-compose up -d --build -``` -L'application sera accessible sur [http://localhost:5643](http://localhost:5643). +# Une seule fois, si le paquet GHCR est privé (jeton GitHub avec le droit read:packages) : +echo | docker login ghcr.io -u --password-stdin -*L'application sera attachée au réseau Docker externe `nginx_default` afin d'être exposée derrière votre reverse proxy Nginx. Assurez-vous que ce réseau existe (`docker network create nginx_default`).* - -### 2. Arrêter l'application -```bash -npm run docker:down -# ou directement -docker-compose down +# Démarrer / mettre à jour : +docker compose up -d ``` -## Développement Local (Sans Docker) +L'application est accessible sur [http://localhost:5643](http://localhost:5643). Elle est attachée au réseau Docker externe `nginx_default` pour être exposée derrière le reverse proxy Nginx (`docker network create nginx_default` s'il n'existe pas). -Si vous préférez développer en local, vous devrez configurer votre propre base de données PostgreSQL. +Construire l'image localement plutôt que la télécharger : `docker compose build`. + +### Données et réglages + +Le volume `collectflow-data` (`/app/data`) conserve les réglages saisis dans **Paramètres** (`.db-config.json` : base, Qlik, API FF, synchro nocturne), les comptes de secours (`users.json`) et le secret de session (`.auth-secret`, si `AUTH_SECRET` n'est pas défini). Il est repris tel quel depuis l'ancienne version Node : comptes et mots de passe restent valables (seules les sessions ouvertes sont à refaire). + +Variables d'environnement utiles : `DATABASE_URL`, `AUTH_SECRET`, `FF_API_BASE_URL`, `QLIK_HOST` / `QLIK_USER` / `QLIK_PWD` (et les réglages fins `QLIK_*`), `PG_STATEMENT_TIMEOUT_MS`, `FF_CACHE_TTL_S`, `RUST_LOG` (niveau de journalisation, `info` par défaut). + +## Développement local -1. Installer les dépendances : ```bash +# Serveur (port 5643) — tables créées et compte admin/admin initialisé au premier démarrage +cd server +DATABASE_URL=postgres://postgres:postgres@localhost/collectflow cargo run + +# Interface (port 5173, /api relayé vers le serveur) npm install -``` - -2. Configurer la connexion DB : -Allez sur la page des Paramètres (`/settings`) dans l'application pour configurer l'accès à votre PostgreSQL local. - -3. Lancer le serveur de développement : -```bash npm run dev ``` -## Structure du Projet (BMAD) +Vérifications : `cargo test` (dans `server/`), `npm run lint`, `npm run build`. -Ce projet respecte l'architecture **BMAD** (Business, Model, Application/API, Data) : -- `src/features/*` : Logique métier (Business) isolée par feature (ex: `grid`, `snapshots`) -- `src/types/*` : Interfaces TypeScript (Model) -- `src/app/*` : Routeurs Next.js UI et endpoints d'API (Application/API) -- `src/db/*` : Schémas Drizzle ORM et connexions (Data) +## Structure + +- `server/src/` — serveur Rust, un module par domaine : `auth` (sessions, clés d'API), `ff` (base miroir et API FF Nancy), `qlik` (données réseau), `grid` (révision d'assortiment, API `/api/v1`), `produits`, `admin` (comptes, paramètres, synchronisation nocturne). +- `server/sql/init.sql` — schéma joué (idempotent) à chaque démarrage. +- `src/` — interface React : `features/*` par domaine, `components/` partagés, `pages/` et `router.tsx`. +- `server/PORTING.md` — contrat entre l'interface et le serveur (actions, données de pages, routes). diff --git a/server/src/admin/grid_warmup.rs b/server/src/admin/grid_warmup.rs new file mode 100644 index 0000000..71f4181 --- /dev/null +++ b/server/src/admin/grid_warmup.rs @@ -0,0 +1,158 @@ +//! Préchauffage de l'instantané de grille (`grid_rows`). +//! +//! L'API `/api/v1` ne calcule jamais : elle lit un instantané écrit quand la +//! Grille est ouverte. Sans préchauffage, un consommateur externe ne verrait que +//! les fournisseurs déjà consultés à la main. Ce job calcule chaque fournisseur +//! une fois, séquentiellement (le calcul est lourd en SQL). + +use crate::auth::AdminUser; +use axum::extract::Query; +use axum::Json; +use parking_lot::Mutex; +use serde::Serialize; +use serde_json::{json, Value}; +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, LazyLock}; + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +struct WarmupJob { + /// "idle" | "running" | "success" | "error" + status: &'static str, + total: usize, + done: usize, + skipped: usize, + failed: usize, + #[serde(skip_serializing_if = "Option::is_none")] + current_fournisseur: Option, + #[serde(skip_serializing_if = "Option::is_none")] + started_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + finished_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, + last_errors: Vec, +} + +impl WarmupJob { + fn idle() -> Self { + Self { + status: "idle", + total: 0, + done: 0, + skipped: 0, + failed: 0, + current_fournisseur: None, + started_at: None, + finished_at: None, + error: None, + last_errors: vec![], + } + } +} + +static JOB: LazyLock>>> = LazyLock::new(|| Arc::new(Mutex::new(None))); + +fn public_job() -> Value { + let job = JOB.lock().clone().unwrap_or_else(WarmupJob::idle); + let mut v = serde_json::to_value(job).unwrap_or(json!({})); + v["success"] = json!(true); + v +} + +fn maintenant() -> String { + crate::db::iso_utc(chrono::Utc::now()) +} + +fn maj(f: impl FnOnce(&mut WarmupJob)) { + if let Some(j) = JOB.lock().as_mut() { + f(j); + } +} + +async fn run_warmup(stale_hours: f64) { + let res: anyhow::Result<()> = async { + let fournisseurs = crate::ff::pg::pg_get_fournisseurs(None).await; + // Les fournisseurs calculés récemment sont sautés : un second passage ne + // coûte presque rien, le bouton est rejouable sans crainte. + let existing = crate::grid::store::list_grid_suppliers().await?; + let fresh_limit = chrono::Utc::now().naive_utc() - chrono::Duration::milliseconds((stale_hours * 3_600_000.0) as i64); + let fresh: HashSet = existing + .iter() + .filter(|s| s.computed_at_naive().map(|d| d > fresh_limit).unwrap_or(false)) + .map(|s| s.code_fournisseur.clone()) + .collect(); + maj(|j| j.total = fournisseurs.len()); + tracing::info!("[grid-warmup] {} fournisseurs, {} déjà à jour (< {}h)", fournisseurs.len(), fresh.len(), stale_hours); + + for f in &fournisseurs { + if fresh.contains(&f.code) { + maj(|j| j.skipped += 1); + continue; + } + maj(|j| j.current_fournisseur = Some(format!("{} — {}", f.code, f.nom))); + match crate::grid::product_rows::get_product_rows_for_sync(&f.code).await { + Ok(_) => maj(|j| j.done += 1), + Err(e) => { + let msg = format!("{}: {}", f.code, crate::error::short(&e, 120)); + tracing::error!("[grid-warmup] {msg}"); + maj(|j| { + j.failed += 1; + if j.last_errors.len() < 10 { + j.last_errors.push(msg); + } + }); + } + } + } + Ok(()) + } + .await; + match res { + Ok(()) => { + maj(|j| { + j.current_fournisseur = None; + j.status = "success"; + j.finished_at = Some(maintenant()); + }); + if let Some(j) = JOB.lock().as_ref() { + tracing::info!("[grid-warmup] terminé — {} calculés, {} à jour, {} en échec", j.done, j.skipped, j.failed); + } + } + Err(e) => { + let msg = format!("{e:#}"); + tracing::error!("[grid-warmup] échec global: {msg}"); + maj(|j| { + j.status = "error"; + j.error = Some(msg); + j.finished_at = Some(maintenant()); + }); + } + } +} + +/// GET /api/admin/grid-warmup — avancement du préchauffage. +pub async fn get(_admin: AdminUser) -> Json { + Json(public_job()) +} + +/// POST /api/admin/grid-warmup?staleHours=24 — démarre en arrière-plan. +pub async fn post(_admin: AdminUser, Query(q): Query>) -> Json { + { + let mut job = JOB.lock(); + if job.as_ref().map(|j| j.status == "running").unwrap_or(false) { + drop(job); + return Json(public_job()); + } + // Comme `Number(param)` : absent ou vide → 0, illisible → défaut 24 h. + let raw = q + .get("staleHours") + .map(|s| if s.trim().is_empty() { 0.0 } else { s.trim().parse::().unwrap_or(f64::NAN) }) + .unwrap_or(0.0); + let stale_hours = if raw.is_finite() { raw.clamp(0.0, 720.0) } else { 24.0 }; + *job = Some(WarmupJob { status: "running", started_at: Some(maintenant()), ..WarmupJob::idle() }); + tracing::info!("[grid-warmup] démarré (staleHours={stale_hours})"); + tokio::spawn(run_warmup(stale_hours)); + } + Json(public_job()) +} diff --git a/server/src/admin/mod.rs b/server/src/admin/mod.rs index abca81b..84e7e96 100644 --- a/server/src/admin/mod.rs +++ b/server/src/admin/mod.rs @@ -4,6 +4,7 @@ pub mod api_keys; pub mod logs; mod qlik_bridge; pub mod settings; +pub mod sync_fournisseurs; pub mod sync_settings; pub mod users; @@ -16,7 +17,9 @@ use serde_json::{Map, Value}; /// Routes HTTP du module. pub fn router() -> Router { - Router::new().route("/api/logs", get(logs::get_logs)) + Router::new() + .route("/api/logs", get(logs::get_logs)) + .route("/api/admin/sync/fournisseurs", get(sync_fournisseurs::get).patch(sync_fournisseurs::patch)) } async fn read_sync_settings_action(ctx: Ctx) -> AppResult { diff --git a/server/src/admin/sync_fournisseurs.rs b/server/src/admin/sync_fournisseurs.rs new file mode 100644 index 0000000..ce94899 --- /dev/null +++ b/server/src/admin/sync_fournisseurs.rs @@ -0,0 +1,148 @@ +//! `/api/admin/sync/fournisseurs` — paramétrage par fournisseur de la synchro nocturne. + +use crate::auth::AdminUser; +use crate::error::{AppError, AppResult}; +use axum::http::StatusCode; +use axum::Json; +use chrono::NaiveDateTime; +use serde::Serialize; +use serde_json::{json, Value}; +use sqlx::postgres::PgRow; +use sqlx::Row; + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct SyncFournisseurRow { + pub code_fournisseur: String, + pub nom_fournisseur: Option, + pub actif_sql: bool, + pub actif_qlik: bool, + pub desactive_motif: Option, + pub dernier_sql_at: Option, + pub dernier_sql_statut: Option, + pub dernier_sql_lignes: Option, + pub dernier_sql_erreur: Option, + pub dernier_sql_ms: Option, + pub dernier_qlik_at: Option, + pub dernier_qlik_statut: Option, + pub dernier_qlik_codes: Option, + pub dernier_qlik_erreur: Option, + pub dernier_qlik_ms: Option, + pub updated_at: Option, +} + +fn ts(r: &PgRow, col: &str) -> Option { + r.try_get::, _>(col).ok().flatten().map(crate::db::iso) +} + +pub fn from_row(r: &PgRow) -> SyncFournisseurRow { + SyncFournisseurRow { + code_fournisseur: r.try_get("code_fournisseur").unwrap_or_default(), + nom_fournisseur: r.try_get("nom_fournisseur").ok().flatten(), + actif_sql: r.try_get("actif_sql").unwrap_or(true), + actif_qlik: r.try_get("actif_qlik").unwrap_or(true), + desactive_motif: r.try_get("desactive_motif").ok().flatten(), + dernier_sql_at: ts(r, "dernier_sql_at"), + dernier_sql_statut: r.try_get("dernier_sql_statut").ok().flatten(), + dernier_sql_lignes: r.try_get("dernier_sql_lignes").ok().flatten(), + dernier_sql_erreur: r.try_get("dernier_sql_erreur").ok().flatten(), + dernier_sql_ms: r.try_get("dernier_sql_ms").ok().flatten(), + dernier_qlik_at: ts(r, "dernier_qlik_at"), + dernier_qlik_statut: r.try_get("dernier_qlik_statut").ok().flatten(), + dernier_qlik_codes: r.try_get("dernier_qlik_codes").ok().flatten(), + dernier_qlik_erreur: r.try_get("dernier_qlik_erreur").ok().flatten(), + dernier_qlik_ms: r.try_get("dernier_qlik_ms").ok().flatten(), + updated_at: ts(r, "updated_at"), + } +} + +/// GET — la liste complète des fournisseurs paramétrés, avec l'état de leur +/// dernière synchronisation (quelques centaines de lignes : pas de pagination). +pub async fn get(_admin: AdminUser) -> AppResult> { + let pool = crate::db::pool()?; + let rows = sqlx::query("SELECT * FROM sync_fournisseurs ORDER BY coalesce(nom_fournisseur, code_fournisseur) ASC") + .fetch_all(&pool) + .await?; + let rows: Vec = rows.iter().map(from_row).collect(); + let resume = json!({ + "total": rows.len(), + "actifsSql": rows.iter().filter(|r| r.actif_sql).count(), + "actifsQlik": rows.iter().filter(|r| r.actif_qlik).count(), + "desactivesAuto": rows.iter().filter(|r| r.desactive_motif.as_deref().map(|m| !m.is_empty()).unwrap_or(false)).count(), + "enEchecSql": rows.iter().filter(|r| r.dernier_sql_statut.as_deref() == Some("echec")).count(), + "enEchecQlik": rows.iter().filter(|r| r.dernier_qlik_statut.as_deref() == Some("echec")).count(), + "jamaisSql": rows.iter().filter(|r| r.dernier_sql_at.is_none()).count(), + }); + Ok(Json(json!({ "success": true, "fournisseurs": rows, "resume": resume }))) +} + +fn invalide(details: Vec) -> AppError { + AppError::with_body(StatusCode::BAD_REQUEST, json!({ "error": "Paramètres invalides", "details": details })) +} + +/// PATCH — active ou désactive des fournisseurs, pour l'une ou l'autre source. +/// +/// Réactiver lève le motif de désactivation automatique : un fournisseur remis +/// en service par l'admin ne doit pas garder un « aucun article » trompeur. +pub async fn patch(_admin: AdminUser, body: axum::body::Bytes) -> AppResult> { + let body: Value = serde_json::from_slice(&body) + .map_err(|_| AppError::with_body(StatusCode::BAD_REQUEST, json!({ "error": "Corps JSON invalide" })))?; + let mut details = Vec::new(); + let codes: Vec = match body.get("codes").and_then(|v| v.as_array()) { + Some(a) => { + let codes: Vec = a.iter().filter_map(|v| v.as_str().map(String::from)).collect(); + if codes.len() != a.len() || codes.iter().any(|c| c.is_empty()) { + details.push("Invalid input".to_string()); + } + if codes.is_empty() { + details.push("Too small: expected array to have >=1 items".to_string()); + } + if codes.len() > 2000 { + details.push("Too big: expected array to have <=2000 items".to_string()); + } + codes + } + None => { + details.push("Invalid input: expected array".to_string()); + vec![] + } + }; + let bool_opt = |k: &str, details: &mut Vec| match body.get(k) { + None | Some(Value::Null) => None, + Some(Value::Bool(b)) => Some(*b), + Some(_) => { + details.push("Invalid input: expected boolean".to_string()); + None + } + }; + let actif_sql = bool_opt("actifSql", &mut details); + let actif_qlik = bool_opt("actifQlik", &mut details); + if details.is_empty() && actif_sql.is_none() && actif_qlik.is_none() { + details.push("Préciser au moins actifSql ou actifQlik".to_string()); + } + if !details.is_empty() { + return Err(invalide(details)); + } + + let reactivation = actif_sql == Some(true) || actif_qlik == Some(true); + let pool = crate::db::pool()?; + let mut qb = sqlx::QueryBuilder::::new("UPDATE sync_fournisseurs SET updated_at = now()"); + if let Some(v) = actif_sql { + qb.push(", actif_sql = ").push_bind(v); + } + if let Some(v) = actif_qlik { + qb.push(", actif_qlik = ").push_bind(v); + } + if reactivation { + qb.push(", desactive_motif = NULL"); + } + qb.push(" WHERE code_fournisseur = ANY(").push_bind(codes).push(") RETURNING code_fournisseur"); + let rows = qb.build().fetch_all(&pool).await?; + tracing::info!( + "[api/admin/sync] {} fournisseur(s) mis à jour (sql={} qlik={})", + rows.len(), + actif_sql.map(|b| b.to_string()).unwrap_or_else(|| "undefined".into()), + actif_qlik.map(|b| b.to_string()).unwrap_or_else(|| "undefined".into()) + ); + Ok(Json(json!({ "success": true, "misAJour": rows.len() }))) +} diff --git a/server/src/admin/sync_routes.rs b/server/src/admin/sync_routes.rs new file mode 100644 index 0000000..ee21b36 --- /dev/null +++ b/server/src/admin/sync_routes.rs @@ -0,0 +1,126 @@ +//! `/api/admin/sync` — réglages, état et pilotage de la synchronisation nocturne. + +use super::sync_scheduler::{debut_fenetre, demander_arret, get_sync_scheduler_state, lancer_round, seed_sync_fournisseurs}; +use super::sync_settings::{read_sync_settings, save_sync_settings}; +use crate::auth::AdminUser; +use crate::error::{AppError, AppResult}; +use axum::http::StatusCode; +use axum::Json; +use chrono::Local; +use serde_json::{json, Map, Value}; + +/// GET — réglages, état courant, et ouverture de fenêtre. +pub async fn get(_admin: AdminUser) -> AppResult> { + let settings = read_sync_settings(); + let maintenant = Local::now(); + let debut = debut_fenetre(&settings, maintenant); + Ok(Json(json!({ + "success": true, + "settings": settings, + "state": get_sync_scheduler_state(), + "dansLaFenetre": debut.is_some(), + "fenetreDebutAt": debut.map(|d| crate::db::iso_utc(d.with_timezone(&chrono::Utc))), + "maintenant": crate::db::iso_utc(maintenant.with_timezone(&chrono::Utc)), + }))) +} + +fn corps_json(body: &[u8]) -> AppResult { + serde_json::from_slice(body).map_err(|_| AppError::with_body(StatusCode::BAD_REQUEST, json!({ "error": "Corps JSON invalide" }))) +} + +/// Entier borné « coercé » comme `z.coerce.number().int().min().max()`. +fn coerce_int(v: &Value, min: i64, max: i64) -> Result { + let n = match v { + Value::Number(n) => n.as_f64(), + Value::String(s) if s.trim().is_empty() => Some(0.0), + Value::String(s) => s.trim().parse::().ok(), + Value::Bool(b) => Some(if *b { 1.0 } else { 0.0 }), + Value::Null => Some(0.0), + _ => None, + }; + let Some(n) = n.filter(|n| n.is_finite()) else { + return Err("Invalid input: expected number, received NaN".into()); + }; + if n.fract() != 0.0 { + return Err("Invalid input: expected int, received number".into()); + } + let n = n as i64; + if n < min { + return Err(format!("Too small: expected number to be >={min}")); + } + if n > max { + return Err(format!("Too big: expected number to be <={max}")); + } + Ok(n) +} + +/// PATCH — met à jour les réglages. +pub async fn patch(_admin: AdminUser, body: axum::body::Bytes) -> AppResult> { + let body = corps_json(&body)?; + let Some(obj) = body.as_object() else { + return Err(AppError::with_body( + StatusCode::BAD_REQUEST, + json!({ "error": "Paramètres invalides", "details": [": Invalid input: expected object"] }), + )); + }; + let mut patch = Map::new(); + let mut details = Vec::new(); + match obj.get("actif") { + None => {} + Some(Value::Bool(b)) => { + patch.insert("actif".into(), Value::Bool(*b)); + } + Some(_) => details.push("actif: Invalid input: expected boolean".to_string()), + } + for (k, min, max) in [("heureDebut", 0, 23), ("heureFin", 0, 23), ("qlikParNuit", 0, 200), ("sqlParQlik", 1, 500), ("qlikMinJours", 0, 365)] { + if let Some(v) = obj.get(k) { + match coerce_int(v, min, max) { + Ok(n) => { + patch.insert(k.into(), Value::from(n)); + } + Err(m) => details.push(format!("{k}: {m}")), + } + } + } + if !details.is_empty() { + return Err(AppError::with_body(StatusCode::BAD_REQUEST, json!({ "error": "Paramètres invalides", "details": details }))); + } + let settings = save_sync_settings(&patch)?; + Ok(Json(json!({ "success": true, "settings": settings }))) +} + +/// POST — démarre un round manuellement, l'arrête, ou resynchronise la liste. +pub async fn post(_admin: AdminUser, body: axum::body::Bytes) -> AppResult<(StatusCode, Json)> { + let body = corps_json(&body)?; + let action = body.get("action").and_then(|v| v.as_str()).unwrap_or_default(); + let forcer = match body.get("forcer") { + None => Some(false), + Some(Value::Bool(b)) => Some(*b), + Some(_) => None, + }; + if !matches!(action, "start" | "stop" | "seed") || forcer.is_none() { + return Err(AppError::with_body(StatusCode::BAD_REQUEST, json!({ "error": "Action inconnue (start, stop ou seed)" }))); + } + match action { + "seed" => { + let n = seed_sync_fournisseurs().await?; + Ok((StatusCode::OK, Json(json!({ "success": true, "fournisseurs": n })))) + } + "stop" => { + demander_arret(); + Ok((StatusCode::OK, Json(json!({ "success": true, "state": get_sync_scheduler_state() })))) + } + _ => { + let etat = get_sync_scheduler_state(); + if etat.en_cours { + return Ok((StatusCode::CONFLICT, Json(json!({ "error": "Un round est déjà en cours.", "state": etat })))); + } + // Lancé en tâche de fond : le round dure des minutes, le client suit via GET. + let forcer = forcer.unwrap_or(false); + tokio::spawn(async move { lancer_round("manuel", forcer).await }); + // Laisse au round le temps de passer « en cours » avant de répondre. + tokio::task::yield_now().await; + Ok((StatusCode::OK, Json(json!({ "success": true, "state": get_sync_scheduler_state() })))) + } + } +} diff --git a/server/src/admin/sync_scheduler.rs b/server/src/admin/sync_scheduler.rs new file mode 100644 index 0000000..7e002f4 --- /dev/null +++ b/server/src/admin/sync_scheduler.rs @@ -0,0 +1,501 @@ +//! Synchronisation nocturne des fournisseurs. +//! +//! Deux sources aux coûts très différents, donc deux files distinctes : +//! • **SQL** — l'instantané de grille (`get_product_rows`), ~43 s sur le plus +//! gros fournisseur. Alimente la Grille et `/api/v1`. +//! • **Qlik** — les métriques réseau, jusqu'à 8 min par fournisseur, sur un +//! serveur externe qui abandonne quand on le sollicite trop. Agrégats +//! mensuels : les rafraîchir chaque nuit n'apporterait rien. +//! +//! ORDRE DE PASSAGE — la file est « le plus ancien d'abord », en ignorant ceux +//! déjà traités depuis l'ouverture de la fenêtre courante. Sans curseur ni +//! compteur de tour : +//! • la nuit 2 reprend là où la nuit 1 s'est arrêtée ; +//! • quand plus personne n'est en attente, le tour est fini ; la nuit +//! suivante tout le monde redevient éligible ; +//! • un fournisseur en échec garde une date ancienne : il repasse en tête. +//! +//! La fenêtre ne peut pas interrompre un fournisseur en cours : elle signifie +//! « ne plus en **démarrer** après l'heure de fin ». + +use super::sync_settings::{read_sync_settings, SyncSettings}; +use chrono::{DateTime, Datelike, Duration, Local, NaiveDateTime, TimeZone, Timelike}; +use parking_lot::Mutex; +use regex::Regex; +use serde::Serialize; +use sqlx::Row; +use std::collections::HashSet; +use std::sync::LazyLock; +use std::time::Instant; + +// ─── État courant, en mémoire ──────────────────────────────────────────────── + +#[derive(Debug, Clone, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct SyncSchedulerState { + pub en_cours: bool, + /// "auto" = déclenché par la fenêtre nocturne, "manuel" = lancé depuis l'admin. + pub origine: Option<&'static str>, + pub debut_at: Option, + pub fin_at: Option, + pub fournisseur_courant: Option, + pub etape: Option<&'static str>, + pub sql_faits: u32, + pub sql_echecs: u32, + pub qlik_faits: u32, + pub qlik_echecs: u32, + /// Renseigné quand la file s'est vidée : le tour est bouclé. + pub tour_termine: bool, + pub arret_demande: bool, + pub derniere_erreur: Option, +} + +static ETAT: LazyLock> = LazyLock::new(|| Mutex::new(SyncSchedulerState::default())); + +pub fn get_sync_scheduler_state() -> SyncSchedulerState { + ETAT.lock().clone() +} + +pub fn demander_arret() { + let mut e = ETAT.lock(); + if e.en_cours { + e.arret_demande = true; + } +} + +fn iso_local(d: DateTime) -> String { + crate::db::iso_utc(d.with_timezone(&chrono::Utc)) +} + +// ─── Fenêtre horaire ───────────────────────────────────────────────────────── + +fn a_heure(maintenant: DateTime, heure: i64) -> DateTime { + let naive = maintenant.date_naive().and_hms_opt(heure.clamp(0, 23) as u32, 0, 0).unwrap(); + Local.from_local_datetime(&naive).earliest().unwrap_or(maintenant) +} + +/// Début de la fenêtre en cours, ou `None` si l'heure courante est en dehors. +/// +/// Gère le passage par minuit (ex. 22h → 5h) : entre 0h et l'heure de fin, la +/// fenêtre a commencé **la veille**. +pub fn debut_fenetre(s: &SyncSettings, maintenant: DateTime) -> Option> { + let h = maintenant.hour() as i64; + let debut_du_jour = a_heure(maintenant, s.heure_debut); + if s.heure_debut < s.heure_fin { + return (h >= s.heure_debut && h < s.heure_fin).then_some(debut_du_jour); + } + // Fenêtre à cheval sur minuit. + if h >= s.heure_debut { + return Some(debut_du_jour); + } + if h < s.heure_fin { + return Some(debut_du_jour - Duration::days(1)); + } + None +} + +// ─── Référentiel des fournisseurs ──────────────────────────────────────────── + +/// Aligne la table de paramétrage sur le référentiel FF : ajoute les nouveaux +/// fournisseurs (actifs par défaut), rafraîchit les noms. Ne supprime rien. +pub async fn seed_sync_fournisseurs() -> anyhow::Result { + let fournisseurs = crate::ff::pg::pg_get_fournisseurs(None).await; + if fournisseurs.is_empty() { + return Ok(0); + } + let pool = crate::db::pool()?; + for lot in fournisseurs.chunks(500) { + let mut qb = sqlx::QueryBuilder::::new("INSERT INTO sync_fournisseurs (code_fournisseur, nom_fournisseur) "); + qb.push_values(lot, |mut b, f| { + b.push_bind(&f.code).push_bind(&f.nom); + }); + // Seul le nom est réaligné : les choix d'activation appartiennent à l'admin. + qb.push(" ON CONFLICT (code_fournisseur) DO UPDATE SET nom_fournisseur = excluded.nom_fournisseur"); + qb.build().execute(&pool).await?; + } + Ok(fournisseurs.len()) +} + +// ─── Files d'attente ───────────────────────────────────────────────────────── + +struct Cible { + code: String, + nom: Option, +} + +fn cible(r: &sqlx::postgres::PgRow) -> Cible { + Cible { code: r.try_get("code_fournisseur").unwrap_or_default(), nom: r.try_get("nom_fournisseur").ok().flatten() } +} + +fn naive(d: DateTime) -> NaiveDateTime { + // Colonnes `timestamp` sans fuseau, écrites par `now()` du serveur (UTC) : + // on compare en UTC. + d.with_timezone(&chrono::Utc).naive_utc() +} + +/// Le plus ancien fournisseur SQL pas encore traité depuis l'ouverture de la fenêtre. +async fn prochain_sql(depuis: DateTime) -> anyhow::Result> { + let pool = crate::db::pool()?; + let row = sqlx::query( + "SELECT code_fournisseur, nom_fournisseur FROM sync_fournisseurs + WHERE actif_sql = true AND (dernier_sql_at IS NULL OR dernier_sql_at < $1) + ORDER BY dernier_sql_at ASC NULLS FIRST LIMIT 1", + ) + .bind(naive(depuis)) + .fetch_optional(&pool) + .await?; + Ok(row.as_ref().map(cible)) +} + +/// Délai avant de retenter un fournisseur Qlik en échec : la nuit suivante. +const QLIK_REPRISE_ECHEC_H: i64 = 18; + +/// Le plus ancien fournisseur Qlik à rafraîchir : données plus vieilles que +/// `qlikMinJours`, ou d'avant le mois en cours (la fenêtre de 12 mois a glissé), +/// ou dernière extraction en échec il y a plus de 18 h. +async fn prochain_qlik(qlik_min_jours: i64, maintenant: DateTime) -> anyhow::Result> { + let seuil_age = maintenant - Duration::days(qlik_min_jours); + let debut_mois = Local + .from_local_datetime(&maintenant.date_naive().with_day(1).unwrap().and_hms_opt(0, 0, 0).unwrap()) + .earliest() + .unwrap_or(maintenant); + let seuil = if seuil_age > debut_mois { seuil_age } else { debut_mois }; + let seuil_echec = maintenant - Duration::hours(QLIK_REPRISE_ECHEC_H); + let pool = crate::db::pool()?; + let row = sqlx::query( + "SELECT code_fournisseur, nom_fournisseur FROM sync_fournisseurs + WHERE actif_qlik = true AND ( + dernier_qlik_at IS NULL + OR dernier_qlik_at < $1 + OR (dernier_qlik_statut = 'echec' AND dernier_qlik_at < $2) + ) + ORDER BY dernier_qlik_at ASC NULLS FIRST LIMIT 1", + ) + .bind(naive(seuil)) + .bind(naive(seuil_echec)) + .fetch_optional(&pool) + .await?; + Ok(row.as_ref().map(cible)) +} + +// ─── Traitement d'un fournisseur ───────────────────────────────────────────── + +fn libelle(code: &str, nom: &Option) -> String { + match nom { + Some(n) if !n.is_empty() => format!("{code} — {n}"), + _ => code.to_string(), + } +} + +fn tronque(s: &str, n: usize) -> String { + s.chars().take(n).collect() +} + +/// Instantané SQL. `dernier_sql_at` est renseigné **même en cas d'échec** : +/// sans cela, un fournisseur en échec resterait le plus ancien et bloquerait la file. +async fn traiter_sql(code: &str, nom: &Option) { + let t0 = Instant::now(); + { + let mut e = ETAT.lock(); + e.fournisseur_courant = Some(libelle(code, nom)); + e.etape = Some("sql"); + } + let res = crate::grid::product_rows::get_product_rows_for_sync(code).await; + let ms = t0.elapsed().as_millis() as i32; + let pool = match crate::db::pool() { + Ok(p) => p, + Err(_) => return, + }; + match res { + Ok(n) => { + let vide = n == 0; + // Un fournisseur sans article n'a rien à donner : on le sort des deux + // files plutôt que de le réessayer chaque nuit (motif visible par l'admin). + let sql = if vide { + "UPDATE sync_fournisseurs SET dernier_sql_at = now(), dernier_sql_statut = 'vide', dernier_sql_lignes = $2, + dernier_sql_erreur = NULL, dernier_sql_ms = $3, actif_sql = false, actif_qlik = false, + desactive_motif = 'Aucun article — désactivé automatiquement', updated_at = now() WHERE code_fournisseur = $1" + } else { + "UPDATE sync_fournisseurs SET dernier_sql_at = now(), dernier_sql_statut = 'succes', dernier_sql_lignes = $2, + dernier_sql_erreur = NULL, dernier_sql_ms = $3, updated_at = now() WHERE code_fournisseur = $1" + }; + if let Err(e) = sqlx::query(sql).bind(code).bind(n as i32).bind(ms).execute(&pool).await { + tracing::error!("[sync-nuit] suivi SQL {code} non écrit : {e}"); + } + if vide { + tracing::info!("[sync-nuit] {code} sans article — désactivé automatiquement"); + } else { + ETAT.lock().sql_faits += 1; + } + } + Err(err) => { + let msg = format!("{err:#}"); + { + let mut e = ETAT.lock(); + e.sql_echecs += 1; + e.derniere_erreur = Some(format!("{code} (SQL) : {msg}")); + } + tracing::error!("[sync-nuit] SQL {code} en échec : {msg}"); + let _ = sqlx::query( + "UPDATE sync_fournisseurs SET dernier_sql_at = now(), dernier_sql_statut = 'echec', dernier_sql_erreur = $2, + dernier_sql_ms = $3, updated_at = now() WHERE code_fournisseur = $1", + ) + .bind(code) + .bind(tronque(&msg, 500)) + .bind(ms) + .execute(&pool) + .await; + } + } +} + +/// Format d'un code centrale plausible — même filtre que la synchro manuelle. +static CODE_CENTRALE: LazyLock = LazyLock::new(|| Regex::new(r"(?i)^[A-Z0-9][A-Z0-9_-]{1,29}$").unwrap()); + +/// Extraction Qlik. Même politique que le SQL sur l'horodatage en cas d'échec. +async fn traiter_qlik(code: &str, nom: &Option) { + let t0 = Instant::now(); + { + let mut e = ETAT.lock(); + e.fournisseur_courant = Some(libelle(code, nom)); + e.etape = Some("qlik"); + } + let res: anyhow::Result> = async { + let articles = crate::ff::pg::pg_get_articles_by_fournisseur(code).await?; + let mut vus = HashSet::new(); + let codes: Vec = articles + .iter() + .map(|a| a.code_centrale.clone().unwrap_or_default().trim().to_string()) + .filter(|c| !c.is_empty() && c != "-" && CODE_CENTRALE.is_match(c)) + .filter(|c| vus.insert(c.clone())) + .collect(); + if codes.is_empty() { + return Ok(None); + } + let n = super::qlik_bridge::extraire_et_enregistrer(codes, code).await?; + Ok(Some(n)) + } + .await; + let ms = t0.elapsed().as_millis() as i32; + let Ok(pool) = crate::db::pool() else { return }; + match res { + Ok(None) => { + let _ = sqlx::query( + "UPDATE sync_fournisseurs SET dernier_qlik_at = now(), dernier_qlik_statut = 'vide', dernier_qlik_codes = 0, + dernier_qlik_erreur = NULL, dernier_qlik_ms = $2, updated_at = now() WHERE code_fournisseur = $1", + ) + .bind(code) + .bind(ms) + .execute(&pool) + .await; + } + Ok(Some(n)) => { + let _ = sqlx::query( + "UPDATE sync_fournisseurs SET dernier_qlik_at = now(), dernier_qlik_statut = 'succes', dernier_qlik_codes = $2, + dernier_qlik_erreur = NULL, dernier_qlik_ms = $3, updated_at = now() WHERE code_fournisseur = $1", + ) + .bind(code) + .bind(n as i32) + .bind(ms) + .execute(&pool) + .await; + ETAT.lock().qlik_faits += 1; + tracing::info!("[sync-nuit] Qlik {code} — {n} codes en {} s", t0.elapsed().as_secs()); + } + Err(err) => { + let msg = format!("{err:#}"); + { + let mut e = ETAT.lock(); + e.qlik_echecs += 1; + e.derniere_erreur = Some(format!("{code} (Qlik) : {msg}")); + } + tracing::error!("[sync-nuit] Qlik {code} en échec : {msg}"); + let _ = sqlx::query( + "UPDATE sync_fournisseurs SET dernier_qlik_at = now(), dernier_qlik_statut = 'echec', dernier_qlik_erreur = $2, + dernier_qlik_ms = $3, updated_at = now() WHERE code_fournisseur = $1", + ) + .bind(code) + .bind(tronque(&msg, 500)) + .bind(ms) + .execute(&pool) + .await; + } + } +} + +// ─── Boucle de travail ─────────────────────────────────────────────────────── + +/// Traite les files jusqu'à épuisement, sortie de fenêtre, ou demande d'arrêt. +/// +/// Les deux files avancent en parallèle : une extraction Qlik est intercalée +/// tous les `sqlParQlik` fournisseurs SQL. Sans cet entrelacement, Qlik ne +/// démarrerait jamais tant que la file SQL n'est pas vide. +pub async fn lancer_round(origine: &'static str, forcer_hors_fenetre: bool) { + let s = read_sync_settings(); + let maintenant = Local::now(); + let debut = debut_fenetre(&s, maintenant); + if !forcer_hors_fenetre && debut.is_none() { + return; + } + { + let mut e = ETAT.lock(); + if e.en_cours { + return; + } + *e = SyncSchedulerState { + en_cours: true, + origine: Some(origine), + debut_at: Some(iso_local(maintenant)), + ..Default::default() + }; + } + // Un lancement manuel hors fenêtre borne la file sur « depuis ce matin ». + let depuis = debut.unwrap_or_else(|| a_heure(maintenant, 0)); + tracing::info!("[sync-nuit] round {origine} démarré (fenêtre {}h-{}h, Qlik ≤ {})", s.heure_debut, s.heure_fin, s.qlik_par_nuit); + + let res: anyhow::Result<()> = async { + seed_sync_fournisseurs().await?; + + // Garde anti-boucle : l'avancement de la file repose sur l'écriture de + // `dernier*_at`, volontairement non bloquante en cas d'erreur. + let mut tentes: HashSet = HashSet::new(); + let mut depuis_dernier_qlik: i64 = 0; + loop { + if ETAT.lock().arret_demande { + break; + } + if !forcer_hors_fenetre && debut_fenetre(&s, Local::now()).is_none() { + tracing::info!("[sync-nuit] fin de fenêtre — arrêt après le fournisseur en cours"); + break; + } + let qlik_possible = { + let e = ETAT.lock(); + ((e.qlik_faits + e.qlik_echecs) as i64) < s.qlik_par_nuit + }; + + // Qlik prioritaire quand le quota d'entrelacement est atteint. + if qlik_possible && depuis_dernier_qlik >= s.sql_par_qlik { + if let Some(c) = prochain_qlik(s.qlik_min_jours, Local::now()).await? { + if tentes.insert(format!("qlik:{}", c.code)) { + traiter_qlik(&c.code, &c.nom).await; + depuis_dernier_qlik = 0; + continue; + } + } + } + + if let Some(c) = prochain_sql(depuis).await? { + if !tentes.insert(format!("sql:{}", c.code)) { + tracing::error!( + "[sync-nuit] {} ressort de la file après avoir été traité : le suivi n'a pas pu être écrit. Arrêt du round pour ne pas boucler.", + c.code + ); + break; + } + traiter_sql(&c.code, &c.nom).await; + depuis_dernier_qlik += 1; + continue; + } + + // File SQL vide : on consacre le temps restant à Qlik. + if qlik_possible { + if let Some(c) = prochain_qlik(s.qlik_min_jours, Local::now()).await? { + if tentes.insert(format!("qlik:{}", c.code)) { + traiter_qlik(&c.code, &c.nom).await; + depuis_dernier_qlik = 0; + continue; + } + } + } + + ETAT.lock().tour_termine = true; + tracing::info!("[sync-nuit] plus rien en attente — tour terminé"); + break; + } + Ok(()) + } + .await; + + if let Err(e) = res { + let msg = format!("{e:#}"); + tracing::error!("[sync-nuit] round interrompu : {msg}"); + ETAT.lock().derniere_erreur = Some(msg); + } + // Les pages (stock, hit-parade, CA…) relisent la base fraîche. + crate::ff_cache::invalidate(); + let e = { + let mut e = ETAT.lock(); + e.en_cours = false; + e.fournisseur_courant = None; + e.etape = None; + e.fin_at = Some(iso_local(Local::now())); + e.clone() + }; + tracing::info!( + "[sync-nuit] round terminé — SQL {} ok / {} en échec, Qlik {} ok / {} en échec, tour bouclé={}", + e.sql_faits, + e.sql_echecs, + e.qlik_faits, + e.qlik_echecs, + e.tour_termine + ); +} + +/// Battement du minuteur : ne fait quelque chose que dans la fenêtre, et jamais +/// deux rounds à la fois. Silencieux hors fenêtre — appelé chaque minute. +pub async fn tick_scheduler() { + let s = read_sync_settings(); + if !s.actif || ETAT.lock().en_cours { + return; + } + if debut_fenetre(&s, Local::now()).is_none() { + return; + } + lancer_round("auto", false).await; +} + +/// Minuteur de la synchronisation nocturne : bat chaque minute et laisse +/// `tick_scheduler` décider s'il y a quelque chose à faire. Rien ne tourne si +/// l'application est arrêtée ; il se réarme tout seul au redémarrage. +pub fn start_sync_timer() { + tracing::info!("[sync-nuit] minuteur démarré (battement 60 s)"); + tokio::spawn(async { + let mut intervalle = tokio::time::interval(std::time::Duration::from_secs(60)); + intervalle.tick().await; + loop { + intervalle.tick().await; + // Un round dure longtemps : il tourne dans sa propre tâche pour qu'une + // panique éventuelle n'arrête pas le minuteur. + if let Err(e) = tokio::spawn(tick_scheduler()).await { + tracing::error!("[sync-nuit] battement en erreur : {e}"); + } + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::admin::sync_settings::SYNC_SETTINGS_DEFAUT; + + fn at(h: u32) -> DateTime { + Local.with_ymd_and_hms(2026, 3, 10, h, 30, 0).unwrap() + } + + #[test] + fn fenetre_simple() { + let s = SyncSettings { heure_debut: 1, heure_fin: 5, ..SYNC_SETTINGS_DEFAUT }; + assert!(debut_fenetre(&s, at(0)).is_none()); + assert_eq!(debut_fenetre(&s, at(3)).unwrap().hour(), 1); + assert!(debut_fenetre(&s, at(5)).is_none()); + } + + #[test] + fn fenetre_minuit() { + let s = SyncSettings { heure_debut: 22, heure_fin: 5, ..SYNC_SETTINGS_DEFAUT }; + assert_eq!(debut_fenetre(&s, at(23)).unwrap().day(), 10); + let d = debut_fenetre(&s, at(2)).unwrap(); + assert_eq!((d.day(), d.hour()), (9, 22)); + assert!(debut_fenetre(&s, at(12)).is_none()); + } +}