Synchro nocturne, préchauffage de la grille (branchement à venir) et README

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LG2LLJuswrrTTPKJmunEVf
This commit is contained in:
Claude committed 2026-10-12 19:06:06 +00:00
1 parent 87d5803b55
commit b80e8f6f42
6 files changed
+976 -36

No files matched your search

+39 -35
View File
@@ -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. 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 ## Prérequis
- **Node.js** 18+ (pour le développement local) - **Docker** et **Docker Compose** (déploiement)
- **Docker** et **Docker Compose** (pour le déploiement conteneurisé) - **PostgreSQL** 16+ (base séparée, non fournie par le conteneur)
- **PostgreSQL** 16+ - Pour développer : **Rust** stable (`rustup`) et **Node.js** 22+
## Lancer avec Docker ## Lancer avec Docker
Le projet est configuré pour tourner dans un conteneur Docker optimisé (mode standalone de Next.js). 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.
**Note :** Vous devez disposer d'une base de données PostgreSQL séparée (le conteneur ne lance que l'application web).
### 1. Démarrer l'application
À la racine du projet, lancez :
```bash ```bash
npm run docker:build # Une seule fois, si le paquet GHCR est privé (jeton GitHub avec le droit read:packages) :
# ou directement echo <JETON> | docker login ghcr.io -u <utilisateur-github> --password-stdin
docker-compose up -d --build
```
L'application sera accessible sur [http://localhost:5643](http://localhost:5643).
*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`).* # Démarrer / mettre à jour :
docker compose up -d
### 2. Arrêter l'application
```bash
npm run docker:down
# ou directement
docker-compose down
``` ```
## 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 ```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 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 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) : ## Structure
- `src/features/*` : Logique métier (Business) isolée par feature (ex: `grid`, `snapshots`)
- `src/types/*` : Interfaces TypeScript (Model) - `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).
- `src/app/*` : Routeurs Next.js UI et endpoints d'API (Application/API) - `server/sql/init.sql` — schéma joué (idempotent) à chaque démarrage.
- `src/db/*` : Schémas Drizzle ORM et connexions (Data) - `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).
+158
View File
@@ -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<String>,
#[serde(skip_serializing_if = "Option::is_none")]
started_at: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
finished_at: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option<String>,
last_errors: Vec<String>,
}
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<Arc<Mutex<Option<WarmupJob>>>> = 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<String> = 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<Value> {
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<HashMap<String, String>>) -> Json<Value> {
{
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::<f64>().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())
}
+4 -1
View File
@@ -4,6 +4,7 @@ pub mod api_keys;
pub mod logs; pub mod logs;
mod qlik_bridge; mod qlik_bridge;
pub mod settings; pub mod settings;
pub mod sync_fournisseurs;
pub mod sync_settings; pub mod sync_settings;
pub mod users; pub mod users;
@@ -16,7 +17,9 @@ use serde_json::{Map, Value};
/// Routes HTTP du module. /// Routes HTTP du module.
pub fn router() -> Router { 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<Value> { async fn read_sync_settings_action(ctx: Ctx) -> AppResult<Value> {
+148
View File
@@ -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<String>,
pub actif_sql: bool,
pub actif_qlik: bool,
pub desactive_motif: Option<String>,
pub dernier_sql_at: Option<String>,
pub dernier_sql_statut: Option<String>,
pub dernier_sql_lignes: Option<i32>,
pub dernier_sql_erreur: Option<String>,
pub dernier_sql_ms: Option<i32>,
pub dernier_qlik_at: Option<String>,
pub dernier_qlik_statut: Option<String>,
pub dernier_qlik_codes: Option<i32>,
pub dernier_qlik_erreur: Option<String>,
pub dernier_qlik_ms: Option<i32>,
pub updated_at: Option<String>,
}
fn ts(r: &PgRow, col: &str) -> Option<String> {
r.try_get::<Option<NaiveDateTime>, _>(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<Json<Value>> {
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<SyncFournisseurRow> = 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<String>) -> 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<Json<Value>> {
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<String> = match body.get("codes").and_then(|v| v.as_array()) {
Some(a) => {
let codes: Vec<String> = 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<String>| 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::<sqlx::Postgres>::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() })))
}
+126
View File
@@ -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<Json<Value>> {
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<Value> {
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<i64, String> {
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::<f64>().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<Json<Value>> {
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<Value>)> {
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() }))))
}
}
}
+501
View File
@@ -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<String>,
pub fin_at: Option<String>,
pub fournisseur_courant: Option<String>,
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<String>,
}
static ETAT: LazyLock<Mutex<SyncSchedulerState>> = 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<Local>) -> String {
crate::db::iso_utc(d.with_timezone(&chrono::Utc))
}
// ─── Fenêtre horaire ─────────────────────────────────────────────────────────
fn a_heure(maintenant: DateTime<Local>, heure: i64) -> DateTime<Local> {
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<Local>) -> Option<DateTime<Local>> {
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<usize> {
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::<sqlx::Postgres>::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<String>,
}
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<Local>) -> 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<Local>) -> anyhow::Result<Option<Cible>> {
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<Local>) -> anyhow::Result<Option<Cible>> {
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>) -> 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<String>) {
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<Regex> = 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<String>) {
let t0 = Instant::now();
{
let mut e = ETAT.lock();
e.fournisseur_courant = Some(libelle(code, nom));
e.etape = Some("qlik");
}
let res: anyhow::Result<Option<usize>> = async {
let articles = crate::ff::pg::pg_get_articles_by_fournisseur(code).await?;
let mut vus = HashSet::new();
let codes: Vec<String> = 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<String> = 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> {
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());
}
}