Serveur Rust : socle (config, base, auth, sessions, journaux, cache FF, SPA)

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 18:56:23 +00:00
1 parent d2c3dc7f8e
commit 3f69b1904e
28 files changed
+5873

No files matched your search

+5
View File
@@ -44,3 +44,8 @@ next-env.d.ts
# données locales (configuration avec mots de passe, comptes de secours)
/data/
# Rust
/server/target/
# Vite
/dist/
+3447
View File
File diff suppressed because it is too large. Load diff
+53
View File
@@ -0,0 +1,53 @@
[package]
name = "collectflow"
version = "0.2.0"
edition = "2021"
description = "CollectFlow — serveur Rust (API, base, FF Nancy, Qlik, synchronisation)"
publish = false
[dependencies]
anyhow = "1"
axum = { version = "0.8", features = ["macros"] }
axum-extra = { version = "0.10", features = ["cookie", "typed-header"] }
base64 = "0.22"
chromiumoxide = "0.9.1"
chrono = { version = "0.4", features = ["serde"] }
futures = "0.3"
hex = "0.4"
hmac = "0.12"
parking_lot = "0.12"
rand = "0.8"
regex = "1"
reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json", "gzip", "cookies", "stream", "http2"] }
rust_decimal = "1"
rust_xlsxwriter = "0.99"
scrypt = { version = "0.11", default-features = false, features = ["std"] }
serde = { version = "1", features = ["derive"] }
serde_json = { version = "1", features = ["preserve_order"] }
sha2 = "0.10"
sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio", "tls-rustls-ring", "postgres", "json", "chrono", "rust_decimal", "macros"] }
subtle = "2"
thiserror = "2"
tokio = { version = "1", features = ["full"] }
tower = "0.5"
tower-http = { version = "0.6", features = ["fs", "compression-gzip", "trace", "set-header"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
urlencoding = "2"
uuid = { version = "1", features = ["v4"] }
[profile.release]
opt-level = 3
lto = "thin"
codegen-units = 4
strip = true
# scrypt est lent sans optimisation : la connexion prenait plusieurs secondes en debug.
[profile.dev.package.scrypt]
opt-level = 3
[profile.dev.package.salsa20]
opt-level = 3
[profile.dev.package.pbkdf2]
opt-level = 3
[profile.dev.package.sha2]
opt-level = 3
+115
View File
@@ -0,0 +1,115 @@
# Portage Rust — conventions de travail
Ce document fixe le contrat entre le serveur Rust (`server/`) et l'interface
React (`src/`, compilée en SPA par Vite). Le code TypeScript d'origine (Next.js)
reste la **référence fonctionnelle** : chaque module Rust en est la traduction
fidèle (mêmes requêtes SQL, mêmes règles, mêmes messages, mêmes formes JSON).
## Socle déjà en place (ne pas réécrire)
| Besoin | Rust |
|---|---|
| Pool PostgreSQL (base app + recopie FF) | `crate::db::pool()? -> PgPool` (remplaçable à chaud, `db::refresh()`) |
| Ligne → JSON générique | `db::row_to_json`, `db::rows_to_json` |
| Lecture tolérante (`Number(r.x) \|\| 0`, `String(r.x ?? "")`) | `db::get_f64`, `db::f64_or0`, `db::get_string`, `db::string_or_empty` |
| Date ISO façon `toISOString()` | `db::iso(NaiveDateTime)`, `db::iso_utc` |
| Réglages `data/.db-config.json` | `config::read_config()`, `config::write_config(patch)`, `config::data_dir()`, `config::env_number(...)` |
| Cache FF (`cachedFF`) | `ff_cache::cached(key, CacheOpts { ttl, cache_if }, \|\| async { … }).await`, `ff_cache::invalidate()` |
| Journaux de job (`startCapture`…) | `logcapture::{start_capture, stop_capture, read_capture, list_captures}` |
| Session (routes internes) | extracteurs `auth::SessionUser`, `auth::AdminUser` (401/403 `{ error }`) ; `auth::Token` (jeton brut) ; `auth::require_session(token)` / `require_admin(token)` |
| Auth `/api/v1` | extracteur `auth::ApiAuth` (clé d'API ou session, erreurs au format v1) ; `ctx.require_admin()` |
| Enveloppe `/api/v1` | `api_v1_response::{ok, fail, build_pagination, ApiErrorCode}` |
| Erreurs | `error::{AppError, AppResult}` — `AppError` implémente `IntoResponse` (`{ error }`) et `From<anyhow::Error / sqlx::Error / reqwest::Error>` |
| Magasins / gammes | `domain::{MAGASINS, CODE_TOUS_MAGASINS, nom_magasin, est_gamme_valide}` |
| Actions / pages | `rpc::{Ctx, Handler, json}` + macro `handler!(fonction)` |
## Organisation
Chaque module possède son répertoire `server/src/<module>/` et expose dans son
`mod.rs` :
```rust
pub fn router() -> axum::Router; // routes HTTP (chemins identiques à Next.js)
pub fn actions() -> Vec<(&'static str, rpc::Handler)>; // ex. ("getUsers", handler!(get_users))
pub fn pages() -> Vec<(&'static str, rpc::Handler)>; // ex. ("dashboard", handler!(page_dashboard))
```
Le routeur Axum n'a pas d'état (`Router<()>`) : tout passe par les accès globaux
ci-dessus. Syntaxe des chemins Axum 0.8 : `/api/v1/products/{codein}`.
## Actions (ex-« server actions »)
`POST /api/actions/{nom}` — corps : tableau JSON des arguments positionnels,
réponse : la valeur de retour de la fonction TS, sérialisée à l'identique.
Une exception TS (ex. `requireSession()` qui lève) devient une `AppError`
(statut 401/403/500, corps `{ error }`) ; le client la relève en `Error`.
Une action s'écrit :
```rust
async fn get_users(ctx: Ctx) -> AppResult<Value> {
ctx.admin().await?; // = await requireAdmin()
let page: i64 = ctx.arg(0)?; // argument n°0 (null si absent → Option<T>)
…
json(resultat)
}
```
| Action | Source TS | Module Rust |
|---|---|---|
| `getFournisseurs`, `getMagasins` | `features/grid/actions.ts` | `ff` |
| `getFournisseursPourCadence`, `listCadences`, `upsertCadence`, `toggleCadence`, `removeCadence` | `features/commandes-auto/actions.ts` | `ff` |
| `getUsers`, `createUser`, `deleteUser`, `updatePassword` | `features/admin/api/user-actions.ts` | `admin` |
| `getApiKeys`, `createApiKey`, `revokeApiKey`, `deleteApiKey` | `features/admin/api/api-key-actions.ts` | `admin` |
| `getParametresAffichables`, `testDatabaseConnection`, `saveDatabaseSettings`, `saveQlikSettings`, `testQlikConnection`, `saveFfApiSettings`, `testFfApiConnection` | `features/settings/actions.ts` | `admin` |
| `readSyncSettings`, `saveSyncSettings` | `features/admin/api/sync-settings.ts` | `admin` |
| `getSnapshots`, `deleteSnapshot`, `saveSnapshot` | `features/snapshots/api/*.ts` | `grid` |
| `saveDraftChanges` | `features/grid/api/save-draft-changes.ts` | `grid` |
| `getGammesAValider`, `rejeterGammesAValider` | `features/grid/api/gammes-a-valider-actions.ts` | `grid` |
(`loginAction` disparaît : `POST /api/auth/login`.)
## Pages (données des anciennes pages serveur)
`GET /api/pages/{nom}?…` renvoie **uniquement les données** que la page serveur
obtenait de ses appels d'accès aux données (fonctions `pg*`, API FF, actions),
sous la forme exacte de leur valeur de retour TS. Toute la mise en forme pure
(pivots, listes de mois, totaux…) reste côté interface. En cas d'échec là où la
page affichait un `ErrorState` : statut 500 `{ error: message }`.
| Page | Paramètres | Réponse | Module |
|---|---|---|---|
| `dashboard` | — | `pgGetDashboardData()` (cache 30 min) | `ff` |
| `analytics` | `mode` (`fournisseur`\|`nomenclature`), `mois`, `moisN1` | lignes de `pgGetCaByFournisseur` ou `pgGetCaByNomenclature` (cachedFF, mêmes clés) | `ff` |
| `hit-parade` | `debut`, `fin` | lignes de `pgGetHitParade` (cachedFF) | `ff` |
| `stock-negatif` | `magasin` | `{ rowsNegatif, rowsSansVente, rowsSansVente6Mois }` (cachedFF) | `ff` |
| `commandes-auto` | — | `{ rows, cadences, fournisseurs }` | `ff` |
| `publicites` | `statut`, `page` | `{ publicites, total, pages, erreurPublicites?, historique, erreurHistorique?, pageSize }` | `ff` |
| `produit-fiche` | `codein` ou `cc` | `getProduitFiche(codein)` / `getProduitFicheByCodeCentrale(cc)` (ou `null`) | `produits` |
## Routes HTTP conservées (mêmes chemins, mêmes réponses)
`/api/ff-status`, `/api/diag`, `/api/diag/stock-fournisseur`, `/api/test-db`,
`/api/commandes-auto` → `ff` ; `/api/qlik/sync` → `qlik` ;
`/api/grid/rows`, `/api/grid/rows/store-patch`, `/api/export/excel`,
`/api/export/modified-gammes`, `/api/v1/*` → `grid` ;
`/api/produits/search`, `/api/produits/opportunites` → `produits` ;
`/api/admin/sync`, `/api/admin/sync/fournisseurs`, `/api/admin/grid-warmup`,
`/api/logs` → `admin`. Authentification : `/api/auth/{login,logout,session}`.
## Règles de traduction
- **Formes JSON identiques** : noms de champs exacts (`#[serde(rename_all = "camelCase")]`
ou `rename` explicite selon l'interface TS), champs optionnels TS absents ⇒
`#[serde(skip_serializing_if = "Option::is_none")]` (JSON.stringify omet `undefined`,
mais garde `null`). Une `Map` TS renvoyée au client se sérialise comme le faisait le TS
(souvent `Object.fromEntries`).
- **Même SQL** : recopier les requêtes telles quelles (paramètres `$1…`, `QueryBuilder`
ou `format!` contrôlé pour les fragments dynamiques — jamais de valeur utilisateur
interpolée sans échappement). Les colonnes `numeric` se lisent avec `db::get_f64`.
- **Même tolérance aux pannes** : une fonction TS qui avale l'erreur et rend `[]`
fait de même ; une qui lève propage une erreur.
- **Journaux** : `tracing::info!/warn!/error!` avec les mêmes préfixes (`[pg-ff]`, `[qlik]`…).
- **Dates** : le conteneur tourne en UTC ; `new Date()` ⇒ `chrono::Local::now()`.
- **Commentaires** en français, comme le code d'origine : garder les « pourquoi ».
- `cargo build` sans avertissement dans son module ; `cargo test` pour la logique pure.
+150
View File
@@ -0,0 +1,150 @@
-- Schéma CollectFlow : joué à chaque démarrage (idempotent).
-- Porté depuis scripts/db-init.js.
CREATE TABLE IF NOT EXISTS "ai_supplier_context" (
"code_fournisseur" varchar(20) PRIMARY KEY NOT NULL,
"context" text NOT NULL,
"updated_at" timestamp DEFAULT now()
);
CREATE TABLE IF NOT EXISTS "session_snapshots" (
"id" serial PRIMARY KEY NOT NULL,
"user_id" integer,
"code_fournisseur" varchar(20) NOT NULL,
"nom_fournisseur" varchar(255),
"magasin" varchar(20) NOT NULL,
"changes" jsonb NOT NULL,
"summary_json" jsonb,
"label" text,
"type" varchar(20) DEFAULT 'snapshot',
"created_at" timestamp DEFAULT now()
);
CREATE INDEX IF NOT EXISTS "idx_snapshots_fournisseur_date"
ON "session_snapshots" ("code_fournisseur", "created_at" DESC);
CREATE INDEX IF NOT EXISTS "idx_snapshots_user_type_date"
ON "session_snapshots" ("user_id", "type", "created_at" DESC);
CREATE TABLE IF NOT EXISTS "users" (
"id" serial PRIMARY KEY NOT NULL,
"username" varchar(50) NOT NULL UNIQUE,
"password_hash" text NOT NULL,
"role" varchar(20) NOT NULL DEFAULT 'user',
"created_at" timestamp DEFAULT now()
);
CREATE TABLE IF NOT EXISTS "commande_cadences" (
"id" serial PRIMARY KEY NOT NULL,
"code_fournisseur" varchar(20) NOT NULL,
"nom_fournisseur" varchar(255),
"site" varchar(20) NOT NULL,
"intervalle_semaines" integer NOT NULL,
"actif" boolean NOT NULL DEFAULT true,
"created_at" timestamp DEFAULT now(),
"updated_at" timestamp DEFAULT now()
);
CREATE UNIQUE INDEX IF NOT EXISTS "uq_cadence_fou_site"
ON "commande_cadences" ("code_fournisseur", "site");
CREATE INDEX IF NOT EXISTS "idx_cadence_site"
ON "commande_cadences" ("site");
CREATE TABLE IF NOT EXISTS "qlik_network_metrics" (
"code_centrale" varchar(20) PRIMARY KEY NOT NULL,
"ca_reseau" numeric(16, 2),
"qte_reseau" numeric(14, 2),
"nb_magasins_reseau" integer,
"ca_par_magasin_reseau" numeric(16, 2),
"marge_pct_reseau" numeric(8, 4),
"periode" varchar(20),
"fetched_at" timestamp DEFAULT now()
);
ALTER TABLE "qlik_network_metrics"
ADD COLUMN IF NOT EXISTS "ca_par_magasin_reseau" numeric(16, 2),
ADD COLUMN IF NOT EXISTS "marge_pct_reseau" numeric(8, 4),
ADD COLUMN IF NOT EXISTS "qte_by_month" jsonb,
ADD COLUMN IF NOT EXISTS "metrics_by_month" jsonb,
ADD COLUMN IF NOT EXISTS "libelle_reseau" varchar(255),
ADD COLUMN IF NOT EXISTS "fournisseur_reseau" varchar(255);
ALTER TABLE "qlik_network_metrics" DROP COLUMN IF EXISTS "rupture_pct_reseau";
ALTER TABLE "qlik_network_metrics" DROP COLUMN IF EXISTS "couverture_stock_reseau";
CREATE TABLE IF NOT EXISTS "grid_rows" (
"code_fournisseur" varchar(20) NOT NULL,
"codein" varchar(20) NOT NULL,
"nom_fournisseur" varchar(255),
"libelle1" varchar(500),
"gtin" varchar(30),
"reference" varchar(100),
"code_centrale" varchar(20),
"code1" varchar(20),
"code2" varchar(20),
"code3" varchar(20),
"code_gamme" varchar(20),
"code_gamme_init" varchar(20),
"total_ca" numeric(16, 2),
"total_quantite" numeric(14, 2),
"total_marge" numeric(16, 2),
"taux_marge" numeric(8, 4),
"stock_actuel" numeric(14, 2),
"ca_reseau" numeric(16, 2),
"qte_reseau" numeric(14, 2),
"nb_magasins_reseau" integer,
"ca_par_magasin_reseau" numeric(16, 2),
"marge_pct_reseau" numeric(8, 4),
"payload" jsonb NOT NULL,
"computed_at" timestamp DEFAULT now() NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS "uq_grid_rows_fou_codein"
ON "grid_rows" ("code_fournisseur", "codein");
CREATE INDEX IF NOT EXISTS "idx_grid_rows_fournisseur" ON "grid_rows" ("code_fournisseur");
CREATE INDEX IF NOT EXISTS "idx_grid_rows_codein" ON "grid_rows" ("codein");
CREATE INDEX IF NOT EXISTS "idx_grid_rows_gamme" ON "grid_rows" ("code_gamme");
CREATE INDEX IF NOT EXISTS "idx_grid_rows_code_centrale" ON "grid_rows" ("code_centrale");
CREATE INDEX IF NOT EXISTS "idx_grid_rows_total_ca" ON "grid_rows" ("total_ca");
CREATE INDEX IF NOT EXISTS "idx_grid_rows_libelle" ON "grid_rows" ("libelle1");
CREATE TABLE IF NOT EXISTS "api_keys" (
"id" serial PRIMARY KEY NOT NULL,
"name" varchar(100) NOT NULL,
"key_prefix" varchar(20) NOT NULL,
"key_hash" text NOT NULL UNIQUE,
"role" varchar(20) DEFAULT 'user' NOT NULL,
"created_by" varchar(50),
"created_at" timestamp DEFAULT now(),
"last_used_at" timestamp,
"revoked_at" timestamp
);
CREATE TABLE IF NOT EXISTS "sync_fournisseurs" (
"code_fournisseur" varchar(20) PRIMARY KEY NOT NULL,
"nom_fournisseur" varchar(255),
"actif_sql" boolean NOT NULL DEFAULT true,
"actif_qlik" boolean NOT NULL DEFAULT true,
"desactive_motif" text,
"dernier_sql_at" timestamp,
"dernier_sql_statut" varchar(20),
"dernier_sql_lignes" integer,
"dernier_sql_erreur" text,
"dernier_sql_ms" integer,
"dernier_qlik_at" timestamp,
"dernier_qlik_statut" varchar(20),
"dernier_qlik_codes" integer,
"dernier_qlik_erreur" text,
"dernier_qlik_ms" integer,
"updated_at" timestamp DEFAULT now()
);
CREATE INDEX IF NOT EXISTS "idx_sync_fou_sql" ON "sync_fournisseurs" ("actif_sql", "dernier_sql_at");
CREATE INDEX IF NOT EXISTS "idx_sync_fou_qlik" ON "sync_fournisseurs" ("actif_qlik", "dernier_qlik_at");
CREATE TABLE IF NOT EXISTS "gammes_a_valider" (
"code_fournisseur" varchar(20) NOT NULL,
"codein" varchar(20) NOT NULL,
"gamme" varchar(20) NOT NULL,
"gamme_precedente" varchar(20),
"auteur" varchar(100),
"created_at" timestamp DEFAULT now() NOT NULL,
PRIMARY KEY ("code_fournisseur", "codein")
);
+22
View File
@@ -0,0 +1,22 @@
//! Module `admin` — à porter.
use crate::rpc::Handler;
use axum::Router;
/// Routes HTTP du module.
pub fn router() -> Router {
Router::new()
}
/// Actions (`POST /api/actions/{nom}`) du module.
pub fn actions() -> Vec<(&'static str, Handler)> {
vec![]
}
/// Données de pages (`GET /api/pages/{nom}`) du module.
pub fn pages() -> Vec<(&'static str, Handler)> {
vec![]
}
/// Tâches de fond lancées au démarrage (minuteur de la synchro nocturne).
pub fn start_background_tasks() {}
+81
View File
@@ -0,0 +1,81 @@
//! Enveloppe de réponse de l'API publique `/api/v1`.
//!
//! Une forme unique : `{ data, pagination?, meta }` en succès,
//! `{ error: { code, message, details? } }` en échec, avec de vrais statuts HTTP.
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use axum::Json;
use serde::Serialize;
use serde_json::{json, Map, Value};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ApiErrorCode {
Unauthorized,
Forbidden,
BadRequest,
NotFound,
/// Donnée pas encore calculée : 202, ce n'est pas un échec.
NotReady,
InternalError,
}
impl ApiErrorCode {
pub fn as_str(self) -> &'static str {
match self {
Self::Unauthorized => "unauthorized",
Self::Forbidden => "forbidden",
Self::BadRequest => "bad_request",
Self::NotFound => "not_found",
Self::NotReady => "not_ready",
Self::InternalError => "internal_error",
}
}
pub fn status(self) -> StatusCode {
match self {
Self::Unauthorized => StatusCode::UNAUTHORIZED,
Self::Forbidden => StatusCode::FORBIDDEN,
Self::BadRequest => StatusCode::BAD_REQUEST,
Self::NotFound => StatusCode::NOT_FOUND,
Self::NotReady => StatusCode::ACCEPTED,
Self::InternalError => StatusCode::INTERNAL_SERVER_ERROR,
}
}
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Pagination {
pub page: i64,
pub limit: i64,
pub total: i64,
pub total_pages: i64,
/// `true` s'il reste des pages à lire.
pub has_more: bool,
}
pub fn build_pagination(page: i64, limit: i64, total: i64) -> Pagination {
let limit = limit.max(1);
let total_pages = ((total as f64) / (limit as f64)).ceil().max(1.0) as i64;
Pagination { page, limit, total, total_pages, has_more: page < total_pages }
}
/// Réponse de succès. `meta` porte la fraîcheur de la donnée (`computedAt`).
pub fn ok<T: Serialize>(data: T, pagination: Option<Pagination>, meta: Option<Map<String, Value>>) -> Response {
let mut body = Map::new();
body.insert("data".into(), serde_json::to_value(data).unwrap_or(Value::Null));
if let Some(p) = pagination {
body.insert("pagination".into(), serde_json::to_value(p).unwrap_or(Value::Null));
}
body.insert("meta".into(), Value::Object(meta.unwrap_or_default()));
Json(Value::Object(body)).into_response()
}
/// Réponse d'erreur. `details` sert notamment aux erreurs de validation.
pub fn fail(code: ApiErrorCode, message: impl Into<String>, details: Option<Value>) -> Response {
let mut err = json!({ "code": code.as_str(), "message": message.into() });
if let Some(d) = details {
err["details"] = d;
}
(code.status(), Json(json!({ "error": err }))).into_response()
}
+149
View File
@@ -0,0 +1,149 @@
//! Authentification de l'API publique `/api/v1`.
//!
//! Deux voies : clé d'API (`X-API-Key` ou `Authorization: Bearer …`) pour les
//! appelants non-navigateur, ou session du navigateur (l'interface utilise les
//! mêmes routes). Échec → 401 JSON au format `/api/v1`, jamais de redirection.
//!
//! Seul le hachage SHA-256 des clés est stocké (table `api_keys`).
use crate::api_v1_response::{fail, ApiErrorCode};
use axum::extract::FromRequestParts;
use axum::http::request::Parts;
use axum::http::HeaderMap;
use axum::response::Response;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use base64::Engine;
use parking_lot::Mutex;
use rand::RngCore;
use sha2::{Digest, Sha256};
use sqlx::Row;
use std::collections::HashMap;
use std::sync::LazyLock;
use std::time::{Duration, Instant};
use subtle::ConstantTimeEq;
const KEY_PREFIX: &str = "cf_";
const TRACE_USAGE: Duration = Duration::from_secs(60);
static DERNIERE_TRACE: LazyLock<Mutex<HashMap<i32, Instant>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
#[derive(Debug, Clone)]
pub struct AuthContext {
/// "api_key" | "session"
pub via: &'static str,
pub role: String,
/// Nom de la clé, ou nom de l'utilisateur connecté.
pub subject: String,
}
impl AuthContext {
/// Restreint un endpoint aux appelants administrateurs.
pub fn require_admin(&self) -> Result<(), Response> {
if self.role != "admin" {
return Err(fail(ApiErrorCode::Forbidden, "Rôle administrateur requis pour cette opération.", None));
}
Ok(())
}
}
pub fn hash_api_key(key: &str) -> String {
hex::encode(Sha256::digest(key.as_bytes()))
}
pub struct GeneratedKey {
pub key: String,
pub key_hash: String,
pub key_prefix: String,
}
/// Génère une clé : en clair (affichée une seule fois), son hachage, et un préfixe lisible.
pub fn generate_api_key() -> GeneratedKey {
let mut buf = [0u8; 24];
rand::thread_rng().fill_bytes(&mut buf);
let key = format!("{KEY_PREFIX}{}", URL_SAFE_NO_PAD.encode(buf));
GeneratedKey { key_hash: hash_api_key(&key), key_prefix: key.chars().take(11).collect(), key }
}
fn hashes_equal(a: &str, b: &str) -> bool {
match (hex::decode(a), hex::decode(b)) {
(Ok(x), Ok(y)) => !x.is_empty() && x.len() == y.len() && bool::from(x.as_slice().ct_eq(y.as_slice())),
_ => false,
}
}
fn read_key(headers: &HeaderMap) -> Option<String> {
if let Some(v) = headers.get("x-api-key").and_then(|v| v.to_str().ok()) {
if !v.trim().is_empty() {
return Some(v.trim().to_string());
}
}
let authz = headers.get("authorization").and_then(|v| v.to_str().ok())?;
let re = regex::Regex::new(r"(?i)^Bearer\s+(.+)$").unwrap();
re.captures(authz).map(|c| c[1].trim().to_string())
}
/// Authentifie une requête `/api/v1`.
pub async fn require_api_auth(headers: &HeaderMap) -> Result<AuthContext, Response> {
if let Some(presented) = read_key(headers) {
let presented_hash = hash_api_key(&presented);
let pool = crate::db::pool().map_err(|e| fail(ApiErrorCode::InternalError, e.message, None))?;
let row = sqlx::query("SELECT id, name, role, key_hash, revoked_at FROM api_keys WHERE key_hash = $1 LIMIT 1")
.bind(&presented_hash)
.fetch_optional(&pool)
.await
.map_err(|e| fail(ApiErrorCode::InternalError, e.to_string(), None))?;
let Some(row) = row else {
return Err(fail(ApiErrorCode::Unauthorized, "Clé d'API invalide.", None));
};
let key_hash: String = row.try_get("key_hash").unwrap_or_default();
if !hashes_equal(&key_hash, &presented_hash) {
return Err(fail(ApiErrorCode::Unauthorized, "Clé d'API invalide.", None));
}
let revoked: Option<chrono::NaiveDateTime> = row.try_get("revoked_at").ok().flatten();
if revoked.is_some() {
return Err(fail(ApiErrorCode::Unauthorized, "Clé d'API révoquée.", None));
}
let id: i32 = row.try_get("id").unwrap_or_default();
// Trace d'usage — jamais bloquante, au plus une écriture par minute et par clé.
let tracer = {
let mut m = DERNIERE_TRACE.lock();
let due = m.get(&id).map(|t| t.elapsed() >= TRACE_USAGE).unwrap_or(true);
if due {
m.insert(id, Instant::now());
}
due
};
if tracer {
tokio::spawn(async move {
if let Err(e) = sqlx::query("UPDATE api_keys SET last_used_at = now() WHERE id = $1").bind(id).execute(&pool).await {
tracing::error!("[api-auth] lastUsedAt KO: {}", crate::error::short(&e, 120));
}
});
}
return Ok(AuthContext {
via: "api_key",
role: row.try_get("role").unwrap_or_else(|_| "user".into()),
subject: row.try_get("name").unwrap_or_default(),
});
}
// Repli sur la session du navigateur.
if let Some(c) = super::session::from_headers(headers) {
return Ok(AuthContext { via: "session", role: c.role, subject: c.name });
}
Err(fail(
ApiErrorCode::Unauthorized,
"Authentification requise : fournissez un en-tête X-API-Key (ou Authorization: Bearer), ou connectez-vous.",
None,
))
}
/// Extracteur pour les routes `/api/v1`.
pub struct ApiAuth(pub AuthContext);
impl<S: Send + Sync> FromRequestParts<S> for ApiAuth {
type Rejection = Response;
async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
Ok(ApiAuth(require_api_auth(&parts.headers).await?))
}
}
+155
View File
@@ -0,0 +1,155 @@
//! Contrôle d'accès commun aux actions, routes et pages.
//!
//! Le jeton de session seul ne suffit pas : il reste valide jusqu'à son
//! expiration, même si le compte a été supprimé ou rétrogradé entre-temps. On
//! relit donc l'utilisateur dans `users` (cache court), et le rôle en base
//! l'emporte sur celui du jeton.
//!
//! Base injoignable : on se contente du jeton (mode dégradé), sans quoi le
//! compte de secours `data/users.json` ne pourrait plus rien faire — y compris
//! configurer la base.
use super::session::{from_headers, Claims};
use crate::error::AppError;
use axum::extract::FromRequestParts;
use axum::http::request::Parts;
use parking_lot::Mutex;
use serde::Serialize;
use sqlx::Row;
use std::collections::HashMap;
use std::sync::LazyLock;
use std::time::{Duration, Instant};
pub const MESSAGE_SESSION: &str = "Session expirée : reconnectez-vous.";
pub const MESSAGE_ADMIN: &str = "Action réservée aux administrateurs.";
const CACHE_TTL: Duration = Duration::from_secs(60);
const PAUSE_BASE_INJOIGNABLE: Duration = Duration::from_secs(15);
#[derive(Debug, Clone, Serialize)]
pub struct UtilisateurCourant {
pub id: String,
pub username: String,
pub role: String,
}
impl UtilisateurCourant {
pub fn is_admin(&self) -> bool {
self.role == "admin"
}
/// Identifiant numérique (`users.id`), `None` pour le compte de secours.
pub fn id_numerique(&self) -> Option<i32> {
self.id.parse::<i32>().ok().filter(|v| *v > 0)
}
}
#[derive(Clone)]
struct Fiche {
exists: bool,
role: String,
username: String,
at: Instant,
}
struct CacheState {
fiches: HashMap<String, Fiche>,
injoignable_depuis: Option<Instant>,
}
static CACHE: LazyLock<Mutex<CacheState>> =
LazyLock::new(|| Mutex::new(CacheState { fiches: HashMap::new(), injoignable_depuis: None }));
/// Oublie la fiche d'un utilisateur (après modification de son compte).
pub fn invalider(id: &str) {
CACHE.lock().fiches.remove(id);
}
/// Fiche de l'utilisateur en base, ou `None` si la base n'a pas pu répondre.
async fn lire_fiche(id: &str) -> Option<Fiche> {
{
let c = CACHE.lock();
if let Some(f) = c.fiches.get(id) {
if f.at.elapsed() < CACHE_TTL {
return Some(f.clone());
}
}
if c.injoignable_depuis.map(|t| t.elapsed() < PAUSE_BASE_INJOIGNABLE).unwrap_or(false) {
return None;
}
}
let id_num: i32 = if id.chars().all(|c| c.is_ascii_digit()) && !id.is_empty() {
id.parse::<i32>().unwrap_or(-1)
} else {
-1
};
let res = async {
let pool = crate::db::pool().map_err(|e| anyhow::anyhow!(e.message))?;
let row = sqlx::query("SELECT username, role FROM users WHERE id = $1 LIMIT 1")
.bind(id_num)
.fetch_optional(&pool)
.await?;
anyhow::Ok(row)
}
.await;
match res {
Ok(row) => {
let fiche = match row {
Some(r) => Fiche {
exists: true,
username: r.try_get("username").unwrap_or_default(),
role: r.try_get("role").unwrap_or_default(),
at: Instant::now(),
},
None => Fiche { exists: false, username: String::new(), role: String::new(), at: Instant::now() },
};
CACHE.lock().fiches.insert(id.to_string(), fiche.clone());
Some(fiche)
}
Err(e) => {
CACHE.lock().injoignable_depuis = Some(Instant::now());
tracing::warn!("[authz] Base injoignable, contrôle sur le seul jeton de session : {}", crate::error::short(&e, 200));
None
}
}
}
/// Utilisateur connecté, relu en base. 401 sinon.
pub async fn require_session(token: Option<&Claims>) -> Result<UtilisateurCourant, AppError> {
let jeton = token.ok_or_else(|| AppError::unauthorized(MESSAGE_SESSION))?;
match lire_fiche(&jeton.id).await {
None => Ok(UtilisateurCourant { id: jeton.id.clone(), username: jeton.name.clone(), role: jeton.role.clone() }),
Some(f) if !f.exists => Err(AppError::unauthorized(MESSAGE_SESSION)),
Some(f) => Ok(UtilisateurCourant { id: jeton.id.clone(), username: f.username, role: f.role }),
}
}
/// Administrateur connecté. 401 ou 403 sinon.
pub async fn require_admin(token: Option<&Claims>) -> Result<UtilisateurCourant, AppError> {
let u = require_session(token).await?;
if !u.is_admin() {
return Err(AppError::forbidden(MESSAGE_ADMIN));
}
Ok(u)
}
/// Extracteur : utilisateur connecté (sinon 401 JSON `{ error }`).
pub struct SessionUser(pub UtilisateurCourant);
impl<S: Send + Sync> FromRequestParts<S> for SessionUser {
type Rejection = AppError;
async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
let claims = from_headers(&parts.headers);
Ok(SessionUser(require_session(claims.as_ref()).await?))
}
}
/// Extracteur : administrateur connecté (sinon 401/403 JSON `{ error }`).
pub struct AdminUser(pub UtilisateurCourant);
impl<S: Send + Sync> FromRequestParts<S> for AdminUser {
type Rejection = AppError;
async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
let claims = from_headers(&parts.headers);
Ok(AdminUser(require_admin(claims.as_ref()).await?))
}
}
+62
View File
@@ -0,0 +1,62 @@
//! Comptes de secours `data/users.json`, utilisés seulement quand la base est
//! injoignable (première installation, base pas encore configurée).
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct UserJson {
pub id: String,
pub username: String,
pub password_hash: String,
pub role: String,
}
fn path() -> PathBuf {
crate::config::data_dir().join("users.json")
}
fn default_admin() -> UserJson {
UserJson {
id: "0".into(),
username: "admin".into(),
password_hash: super::password::hash_password("admin"),
role: "admin".into(),
}
}
/// Comptes de secours (créés avec admin/admin si le fichier n'existe pas).
pub fn get_fallback_users() -> Vec<UserJson> {
let p = path();
if !p.exists() {
let admin = default_admin();
save_fallback_users(&[admin.clone()]);
return vec![admin];
}
match std::fs::read_to_string(&p).ok().and_then(|s| serde_json::from_str(&s).ok()) {
Some(users) => users,
None => {
tracing::error!("[AUTH] Error reading users.json");
vec![]
}
}
}
pub fn save_fallback_users(users: &[UserJson]) {
let p = path();
if let Some(dir) = p.parent() {
let _ = std::fs::create_dir_all(dir);
}
if let Err(e) = std::fs::write(&p, serde_json::to_string_pretty(users).unwrap_or_default()) {
tracing::error!("[AUTH] Error writing users.json: {e}");
}
}
/// Base injoignable : le fichier de secours doit contenir au moins un administrateur.
pub fn assurer_admin_de_secours() {
if get_fallback_users().is_empty() {
tracing::info!("[AUTH] Initializing local JSON fallback with admin/admin...");
save_fallback_users(&[default_admin()]);
}
}
+191
View File
@@ -0,0 +1,191 @@
//! Authentification : comptes, sessions, clés d'API.
pub mod api_key;
pub mod authz;
pub mod fallback;
pub mod password;
pub mod session;
pub use api_key::{ApiAuth, AuthContext};
pub use authz::{require_admin, require_session, AdminUser, SessionUser, UtilisateurCourant};
pub use session::{Claims, Token};
use axum::http::{header, HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use serde::Deserialize;
use serde_json::json;
use sqlx::Row;
use tokio::sync::Mutex;
// ---------------------------------------------------------------------------
// Amorçage du compte admin
// ---------------------------------------------------------------------------
/// `true` une fois l'amorçage réussi pour ce processus.
static AMORCE: Mutex<bool> = Mutex::const_new(false);
/// Prépare les comptes au premier démarrage, une seule fois par processus.
///
/// - Base joignable : crée la table `users` si besoin, puis admin/admin
/// **uniquement si la table est vide**.
/// - Base injoignable : s'assure que le fichier de secours contient un
/// administrateur ; on retentera la base à la prochaine connexion.
pub async fn ensure_admin_exists() -> bool {
let mut done = AMORCE.lock().await;
if *done {
return false;
}
match amorcer().await {
Ok(cree) => {
*done = true;
cree
}
Err(e) => {
tracing::warn!("[AUTH] Database seeding failed (expected if DB not configured yet): {e:#}");
false
}
}
}
async fn amorcer() -> anyhow::Result<bool> {
let pool = match crate::db::pool() {
Ok(p) => p,
Err(e) => {
fallback::assurer_admin_de_secours();
anyhow::bail!(e.message);
}
};
if let Err(e) = sqlx::query("SELECT 1").execute(&pool).await {
fallback::assurer_admin_de_secours();
return Err(e.into());
}
sqlx::raw_sql(
r#"CREATE TABLE IF NOT EXISTS "users" (
id SERIAL PRIMARY KEY,
username VARCHAR(50) NOT NULL UNIQUE,
password_hash TEXT NOT NULL,
role VARCHAR(20) NOT NULL DEFAULT 'user',
created_at TIMESTAMP DEFAULT NOW()
)"#,
)
.execute(&pool)
.await?;
let nb: i64 = sqlx::query_scalar("SELECT count(*) FROM users").fetch_one(&pool).await?;
if nb > 0 {
return Ok(false);
}
tracing::info!("[AUTH] Table users vide : création du compte admin/admin.");
let res = sqlx::query("INSERT INTO users (username, password_hash, role) VALUES ('admin', $1, 'admin') ON CONFLICT DO NOTHING")
.bind(password::hash_password("admin"))
.execute(&pool)
.await?;
Ok(res.rows_affected() > 0)
}
// ---------------------------------------------------------------------------
// Connexion / déconnexion
// ---------------------------------------------------------------------------
struct CompteTrouve {
id: String,
username: String,
password_hash: String,
role: String,
}
/// Vérifie les identifiants. `None` si refusés.
async fn authorize(username: &str, password: &str) -> Option<CompteTrouve> {
ensure_admin_exists().await;
if username.is_empty() || password.is_empty() {
return None;
}
tracing::info!("[AUTH] Checking user \"{username}\"...");
let db_lookup = async {
let pool = crate::db::pool().map_err(|e| anyhow::anyhow!(e.message))?;
let row = sqlx::query("SELECT id, username, password_hash, role FROM users WHERE username = $1")
.bind(username)
.fetch_optional(&pool)
.await?;
anyhow::Ok(row.map(|r| CompteTrouve {
id: r.try_get::<i32, _>("id").unwrap_or_default().to_string(),
username: r.try_get("username").unwrap_or_default(),
password_hash: r.try_get("password_hash").unwrap_or_default(),
role: r.try_get("role").unwrap_or_else(|_| "user".into()),
}))
};
let user = match db_lookup.await {
Ok(u) => {
tracing::info!("[AUTH] DB check: User found: {}", u.is_some());
u
}
Err(_) => {
// Base injoignable uniquement : un compte absent de la base ne doit
// pas pouvoir se connecter avec le fichier de secours.
tracing::warn!("[AUTH] DB unreachable, falling back to JSON.");
fallback::get_fallback_users().into_iter().find(|u| u.username == username).map(|u| CompteTrouve {
id: u.id,
username: u.username,
password_hash: u.password_hash,
role: u.role,
})
}
}?;
let ok = password::verify_password(password, &user.password_hash);
tracing::info!("[AUTH] Password valid for \"{username}\": {ok}");
ok.then_some(user)
}
#[derive(Deserialize)]
struct LoginBody {
#[serde(default)]
username: String,
#[serde(default)]
password: String,
}
async fn login(headers: HeaderMap, Json(body): Json<LoginBody>) -> Response {
let Some(user) = authorize(body.username.trim(), &body.password).await else {
return (StatusCode::UNAUTHORIZED, Json(json!({ "error": "Utilisateur ou mot de passe incorrect." }))).into_response();
};
let claims = Claims {
id: user.id.clone(),
name: user.username.clone(),
role: user.role.clone(),
exp: chrono::Utc::now().timestamp() + session::MAX_AGE_SECS,
};
(
[(header::SET_COOKIE, session::set_cookie(&claims, &headers)), (header::CACHE_CONTROL, "no-store".into())],
Json(json!({ "ok": true, "user": { "id": user.id, "name": user.username, "role": user.role } })),
)
.into_response()
}
async fn logout() -> Response {
([(header::SET_COOKIE, session::clear_cookie())], Json(json!({ "ok": true }))).into_response()
}
/// Session courante pour l'interface (`useSession`). Relue en base : un compte
/// supprimé n'a plus de session.
async fn current_session(Token(token): Token) -> Response {
let user = match token.as_ref() {
Some(_) => authz::require_session(token.as_ref()).await.ok(),
None => None,
};
let body = match user {
Some(u) => json!({ "user": { "id": u.id, "name": u.username, "role": u.role } }),
None => json!({ "user": null }),
};
([(header::CACHE_CONTROL, "no-store")], Json(body)).into_response()
}
pub fn router() -> Router {
Router::new()
.route("/api/auth/login", post(login))
.route("/api/auth/logout", post(logout))
.route("/api/auth/session", get(current_session))
}
+60
View File
@@ -0,0 +1,60 @@
//! Mots de passe : scrypt au format `sel:hachage` (hex), compatible avec les
//! comptes créés par l'ancienne version Node (`scryptSync(pwd, sel, 64)`,
//! N=16384, r=8, p=1, le sel hexadécimal étant utilisé tel quel comme octets).
use rand::RngCore;
use subtle::ConstantTimeEq;
fn derive(password: &str, salt: &str) -> Option<[u8; 64]> {
let params = scrypt::Params::new(14, 8, 1, 64).ok()?;
let mut out = [0u8; 64];
scrypt::scrypt(password.as_bytes(), salt.as_bytes(), &params, &mut out).ok()?;
Some(out)
}
pub fn hash_password(password: &str) -> String {
let mut salt = [0u8; 16];
rand::thread_rng().fill_bytes(&mut salt);
let salt = hex::encode(salt);
let hash = derive(password, &salt).map(hex::encode).unwrap_or_default();
format!("{salt}:{hash}")
}
pub fn verify_password(password: &str, stored: &str) -> bool {
let Some((salt, hash)) = stored.split_once(':') else { return false };
if salt.is_empty() || hash.is_empty() {
return false;
}
let Ok(expected) = hex::decode(hash) else { return false };
let Some(actual) = derive(password, salt) else { return false };
expected.len() == actual.len() && bool::from(actual.as_slice().ct_eq(expected.as_slice()))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn roundtrip() {
let h = hash_password("admin");
assert!(verify_password("admin", &h));
assert!(!verify_password("nope", &h));
}
#[test]
fn compatible_node() {
// require("crypto").scryptSync("admin", "00112233445566778899aabbccddeeff", 64).toString("hex")
let stored = "00112233445566778899aabbccddeeff:".to_string()
+ &hex::encode(derive("admin", "00112233445566778899aabbccddeeff").unwrap());
assert!(verify_password("admin", &stored));
}
}
#[cfg(test)]
mod node_compat {
#[test]
fn hash_from_node() {
assert!(super::verify_password("s3cret", "abcdef0123456789abcdef0123456789:8e07d3f906fe581051e47f776fd3fb6307401dc4702369a6db2b7fbb892eb23571af616a59cf13a162cfd584035381c062b7d17fd4ca0d82a2cdf0e1ce2330aa"));
assert!(!super::verify_password("s3cre", "abcdef0123456789abcdef0123456789:8e07d3f906fe581051e47f776fd3fb6307401dc4702369a6db2b7fbb892eb23571af616a59cf13a162cfd584035381c062b7d17fd4ca0d82a2cdf0e1ce2330aa"));
}
}
+124
View File
@@ -0,0 +1,124 @@
//! Jeton de session signé (cookie HttpOnly), remplaçant le JWT NextAuth.
//!
//! Contenu : identifiant, nom et rôle de l'utilisateur + expiration, encodés en
//! base64url et signés HMAC-SHA256 avec `AUTH_SECRET` (ou un secret aléatoire
//! conservé dans `data/.auth-secret`).
use axum::extract::FromRequestParts;
use axum::http::request::Parts;
use axum::http::HeaderMap;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use base64::Engine;
use hmac::{Hmac, Mac};
use rand::RngCore;
use serde::{Deserialize, Serialize};
use sha2::Sha256;
use std::convert::Infallible;
use std::sync::LazyLock;
pub const COOKIE_NAME: &str = "cf_session";
/// Durée de vie d'une session : 30 jours (comme NextAuth).
pub const MAX_AGE_SECS: i64 = 30 * 24 * 3600;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Claims {
/// Identifiant (`users.id`, ou « 0 » pour le compte de secours).
pub id: String,
pub name: String,
pub role: String,
/// Expiration (secondes Unix).
pub exp: i64,
}
static SECRET: LazyLock<Vec<u8>> = LazyLock::new(|| {
if let Ok(s) = std::env::var("AUTH_SECRET") {
if !s.trim().is_empty() {
return s.into_bytes();
}
}
let path = crate::config::data_dir().join(".auth-secret");
if let Ok(s) = std::fs::read_to_string(&path) {
if !s.trim().is_empty() {
return s.trim().as_bytes().to_vec();
}
}
let mut buf = [0u8; 32];
rand::thread_rng().fill_bytes(&mut buf);
let s = hex::encode(buf);
let _ = std::fs::create_dir_all(crate::config::data_dir());
if let Err(e) = std::fs::write(&path, &s) {
tracing::warn!("[auth] secret de session non persisté ({e}) : les sessions ne survivront pas au redémarrage");
}
s.into_bytes()
});
fn sign(payload: &str) -> String {
let mut mac = Hmac::<Sha256>::new_from_slice(&SECRET).expect("hmac");
mac.update(payload.as_bytes());
URL_SAFE_NO_PAD.encode(mac.finalize().into_bytes())
}
pub fn encode(claims: &Claims) -> String {
let payload = URL_SAFE_NO_PAD.encode(serde_json::to_vec(claims).unwrap_or_default());
let sig = sign(&payload);
format!("{payload}.{sig}")
}
pub fn decode(token: &str) -> Option<Claims> {
let (payload, sig) = token.split_once('.')?;
let mut mac = Hmac::<Sha256>::new_from_slice(&SECRET).ok()?;
mac.update(payload.as_bytes());
let sig_bytes = URL_SAFE_NO_PAD.decode(sig).ok()?;
mac.verify_slice(&sig_bytes).ok()?;
let claims: Claims = serde_json::from_slice(&URL_SAFE_NO_PAD.decode(payload).ok()?).ok()?;
if claims.exp < chrono::Utc::now().timestamp() {
return None;
}
Some(claims)
}
/// Jeton lu dans l'en-tête `Cookie`, s'il est présent et valide.
pub fn from_headers(headers: &HeaderMap) -> Option<Claims> {
for value in headers.get_all(axum::http::header::COOKIE) {
let Ok(s) = value.to_str() else { continue };
for part in s.split(';') {
let part = part.trim();
if let Some(v) = part.strip_prefix(COOKIE_NAME).and_then(|r| r.strip_prefix('=')) {
if let Some(c) = decode(v) {
return Some(c);
}
}
}
}
None
}
fn is_https(headers: &HeaderMap) -> bool {
headers
.get("x-forwarded-proto")
.and_then(|v| v.to_str().ok())
.map(|v| v.eq_ignore_ascii_case("https"))
.unwrap_or(false)
}
/// En-tête `Set-Cookie` qui ouvre la session.
pub fn set_cookie(claims: &Claims, req_headers: &HeaderMap) -> String {
let secure = if is_https(req_headers) { "; Secure" } else { "" };
format!("{COOKIE_NAME}={}; Path=/; HttpOnly; SameSite=Lax; Max-Age={MAX_AGE_SECS}{secure}", encode(claims))
}
/// En-tête `Set-Cookie` qui ferme la session.
pub fn clear_cookie() -> String {
format!("{COOKIE_NAME}=; Path=/; HttpOnly; SameSite=Lax; Max-Age=0")
}
/// Jeton de la requête (jamais de rejet : `None` si absent ou invalide).
#[derive(Debug, Clone)]
pub struct Token(pub Option<Claims>);
impl<S: Send + Sync> FromRequestParts<S> for Token {
type Rejection = Infallible;
async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
Ok(Token(from_headers(&parts.headers)))
}
}
+99
View File
@@ -0,0 +1,99 @@
//! Fichier de configuration `data/.db-config.json` (réglages saisis dans Paramètres).
//!
//! Le fichier contient des mots de passe et des clés d'API : il ne repart jamais
//! tel quel vers le navigateur. Les clés inconnues sont conservées à l'écriture.
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
use std::path::PathBuf;
/// Répertoire des données persistantes (volume Docker `/app/data`).
pub fn data_dir() -> PathBuf {
std::env::var("DATA_DIR")
.map(PathBuf::from)
.unwrap_or_else(|_| std::env::current_dir().unwrap_or_default().join("data"))
}
pub fn config_file() -> PathBuf {
data_dir().join(".db-config.json")
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DbConfig {
#[serde(default)]
pub url: Option<String>,
#[serde(default)]
pub open_router_key: Option<String>,
#[serde(default)]
pub open_router_model: Option<String>,
#[serde(default)]
pub ai_provider: Option<String>,
#[serde(default)]
pub google_ai_key: Option<String>,
#[serde(default)]
pub google_ai_model: Option<String>,
/// Qlik Sense — données réseau.
#[serde(default)]
pub qlik_host: Option<String>,
#[serde(default)]
pub qlik_user: Option<String>,
#[serde(default)]
pub qlik_password: Option<String>,
/// API REST FF Nancy (ex. https://api.ffnancy.fr).
#[serde(default)]
pub ff_api_base_url: Option<String>,
}
fn read_raw_sync() -> Map<String, Value> {
std::fs::read_to_string(config_file())
.ok()
.and_then(|s| serde_json::from_str::<Value>(&s).ok())
.and_then(|v| match v {
Value::Object(m) => Some(m),
_ => None,
})
.unwrap_or_default()
}
/// Lit la config existante (vide si absente ou illisible).
pub fn read_config() -> DbConfig {
serde_json::from_value(Value::Object(read_raw_sync())).unwrap_or_default()
}
/// Fusionne `patch` dans la config enregistrée : les clés absentes du patch sont
/// conservées ; une clé passée à `null` est retirée.
pub fn write_config(patch: Map<String, Value>) -> std::io::Result<()> {
let mut existing = read_raw_sync();
for (k, v) in patch {
if v.is_null() {
existing.remove(&k);
} else {
existing.insert(k, v);
}
}
std::fs::create_dir_all(data_dir())?;
let text = serde_json::to_string_pretty(&Value::Object(existing)).unwrap_or_else(|_| "{}".into());
std::fs::write(config_file(), text)
}
/// Chaîne non vide, sinon `None`.
pub fn non_empty(s: Option<String>) -> Option<String> {
s.map(|v| v.trim().to_string()).filter(|v| !v.is_empty())
}
/// URL de connexion PostgreSQL : réglage enregistré, sinon `DATABASE_URL`.
pub fn database_url() -> Option<String> {
non_empty(read_config().url).or_else(|| non_empty(std::env::var("DATABASE_URL").ok()))
}
/// Variable d'environnement numérique bornée (équivalent de `envNumber` côté TS).
pub fn env_number(name: &str, default: i64, min: i64, max: i64) -> i64 {
match std::env::var(name) {
Ok(raw) if !raw.trim().is_empty() => match raw.trim().parse::<f64>() {
Ok(v) if v.is_finite() => (v.trunc() as i64).clamp(min, max),
_ => default,
},
_ => default,
}
}
+202
View File
@@ -0,0 +1,202 @@
//! Pool PostgreSQL unique par processus, remplaçable à chaud.
//!
//! L'URL vient de `data/.db-config.json` (Paramètres), sinon de `DATABASE_URL`.
//! Enregistrer de nouveaux réglages appelle [`refresh`] : le pool suivant est
//! recréé à la première requête.
//!
//! La même base porte les tables de l'application et la recopie nocturne des
//! tables FF Nancy (`articles`, `mvtart`, `cube_stock`, `fouident`…).
use crate::error::{AppError, AppResult};
use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc};
use parking_lot::RwLock;
use serde_json::{Map, Value};
use sqlx::postgres::{PgConnectOptions, PgPoolOptions, PgRow};
use sqlx::{Column, Executor, PgPool, Row, TypeInfo, ValueRef};
use std::str::FromStr;
use std::sync::LazyLock;
use std::time::Duration;
static POOL: LazyLock<RwLock<Option<PgPool>>> = LazyLock::new(|| RwLock::new(None));
/// Masque le mot de passe d'une URL de connexion pour les journaux.
pub fn mask_url(url: &str) -> String {
let re = regex::Regex::new(r":([^@/]+)@").unwrap();
re.replace(url, ":****@").into_owned()
}
/// Options de connexion communes (pool applicatif et tests de connexion).
pub fn connect_options(url: &str) -> Result<PgConnectOptions, sqlx::Error> {
let mut opts = PgConnectOptions::from_str(url)?.application_name("collectflow");
// Délai maximal par requête : opt-in (les calculs nocturnes des plus gros
// fournisseurs peuvent être longs).
if let Some(ms) = std::env::var("PG_STATEMENT_TIMEOUT_MS").ok().and_then(|v| v.parse::<u64>().ok()).filter(|v| *v > 0) {
opts = opts.options([("statement_timeout", ms.to_string())]);
}
Ok(opts)
}
fn build_pool(url: &str) -> Result<PgPool, sqlx::Error> {
let opts = connect_options(url)?;
Ok(PgPoolOptions::new()
.max_connections(20)
.idle_timeout(Duration::from_secs(30))
.acquire_timeout(Duration::from_secs(10))
// Pas de workers parallèles PostgreSQL : ils consomment /dev/shm, trop
// petit dans le conteneur, et font échouer les grosses agrégations.
.after_connect(|conn, _meta| {
Box::pin(async move {
if let Err(e) = conn.execute("SET max_parallel_workers_per_gather = 0").await {
tracing::error!("[DB] SET max_parallel_workers_per_gather failed: {e}");
}
Ok(())
})
})
.connect_lazy_with(opts))
}
/// Pool courant, créé à la demande. Erreur 500 si aucune base n'est configurée.
pub fn pool() -> AppResult<PgPool> {
if let Some(p) = POOL.read().as_ref() {
return Ok(p.clone());
}
let mut guard = POOL.write();
if let Some(p) = guard.as_ref() {
return Ok(p.clone());
}
let url = crate::config::database_url().ok_or_else(|| {
tracing::error!("[DB] CRITICAL: No database connection string available!");
AppError::internal("Base de données non configurée (Paramètres → Base de données).")
})?;
tracing::info!("[DB] Initializing connection with: {}", mask_url(&url));
let p = build_pool(&url).map_err(|e| AppError::internal(format!("URL de base invalide : {e}")))?;
*guard = Some(p.clone());
Ok(p)
}
/// Oublie le pool courant (réglages modifiés dans Paramètres).
pub fn refresh() {
tracing::info!("[DB] Refreshing database connection...");
if let Some(old) = POOL.write().take() {
tokio::spawn(async move { old.close().await });
}
}
// ---------------------------------------------------------------------------
// Conversion générique d'une ligne en JSON
// ---------------------------------------------------------------------------
/// Horodatage au format JS `toISOString()` (`2026-01-31T12:34:56.789Z`).
pub fn iso(dt: NaiveDateTime) -> String {
dt.format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string()
}
pub fn iso_utc(dt: DateTime<Utc>) -> String {
iso(dt.naive_utc())
}
/// Convertit une ligne en objet JSON, colonne par colonne, selon le type SQL.
///
/// Les `numeric` deviennent des nombres (le pilote Node les rendait en chaînes,
/// que le code convertissait avec `Number()`), les horodatages des chaînes ISO.
pub fn row_to_json(row: &PgRow) -> Value {
let mut obj = Map::new();
for (i, col) in row.columns().iter().enumerate() {
obj.insert(col.name().to_string(), cell_to_json(row, i));
}
Value::Object(obj)
}
pub fn rows_to_json(rows: &[PgRow]) -> Vec<Value> {
rows.iter().map(row_to_json).collect()
}
fn cell_to_json(row: &PgRow, i: usize) -> Value {
let raw = match row.try_get_raw(i) {
Ok(r) => r,
Err(_) => return Value::Null,
};
if raw.is_null() {
return Value::Null;
}
let ty = raw.type_info().name().to_ascii_uppercase();
macro_rules! get {
($t:ty) => {
row.try_get::<$t, _>(i).ok()
};
}
let num = |f: Option<f64>| f.and_then(serde_json::Number::from_f64).map(Value::Number).unwrap_or(Value::Null);
match ty.as_str() {
"BOOL" => get!(bool).map(Value::Bool).unwrap_or(Value::Null),
"INT2" => get!(i16).map(|v| Value::from(v)).unwrap_or(Value::Null),
"INT4" => get!(i32).map(|v| Value::from(v)).unwrap_or(Value::Null),
"INT8" => get!(i64).map(|v| Value::from(v)).unwrap_or(Value::Null),
"FLOAT4" => num(get!(f32).map(|v| v as f64)),
"FLOAT8" => num(get!(f64)),
"NUMERIC" => num(get!(rust_decimal::Decimal).and_then(|d| d.to_string().parse::<f64>().ok())),
"JSON" | "JSONB" => get!(Value).unwrap_or(Value::Null),
"TIMESTAMP" => get!(NaiveDateTime).map(|d| Value::String(iso(d))).unwrap_or(Value::Null),
"TIMESTAMPTZ" => get!(DateTime<Utc>).map(|d| Value::String(iso_utc(d))).unwrap_or(Value::Null),
"DATE" => get!(NaiveDate)
.map(|d| Value::String(iso(d.and_hms_opt(0, 0, 0).unwrap())))
.unwrap_or(Value::Null),
"TEXT[]" | "VARCHAR[]" | "_TEXT" | "_VARCHAR" => get!(Vec<String>).map(Value::from).unwrap_or(Value::Null),
_ => get!(String).map(Value::String).unwrap_or(Value::Null),
}
}
// ---------------------------------------------------------------------------
// Lecture tolérante de colonnes (équivalent des `Number(r.x) || 0` du code TS)
// ---------------------------------------------------------------------------
/// Valeur numérique d'une colonne quel que soit son type SQL (int, numeric,
/// float, texte numérique). `None` si NULL ou illisible.
pub fn get_f64(row: &PgRow, col: &str) -> Option<f64> {
let raw = row.try_get_raw(col).ok()?;
if raw.is_null() {
return None;
}
let ty = raw.type_info().name().to_ascii_uppercase();
match ty.as_str() {
"INT2" => row.try_get::<i16, _>(col).ok().map(|v| v as f64),
"INT4" => row.try_get::<i32, _>(col).ok().map(|v| v as f64),
"INT8" => row.try_get::<i64, _>(col).ok().map(|v| v as f64),
"FLOAT4" => row.try_get::<f32, _>(col).ok().map(|v| v as f64),
"FLOAT8" => row.try_get::<f64, _>(col).ok(),
"NUMERIC" => row
.try_get::<rust_decimal::Decimal, _>(col)
.ok()
.and_then(|d| d.to_string().parse::<f64>().ok()),
_ => row.try_get::<String, _>(col).ok().and_then(|s| s.trim().parse::<f64>().ok()),
}
.filter(|v| v.is_finite())
}
/// `Number(r.col) || 0`.
pub fn f64_or0(row: &PgRow, col: &str) -> f64 {
get_f64(row, col).unwrap_or(0.0)
}
/// Valeur texte d'une colonne quel que soit son type (nombres et dates convertis).
pub fn get_string(row: &PgRow, col: &str) -> Option<String> {
let idx = row.columns().iter().position(|c| c.name() == col)?;
match cell_to_json(row, idx) {
Value::Null => None,
Value::String(s) => Some(s),
Value::Number(n) => {
// 12.0 → "12", comme String(12) en JS.
let f = n.as_f64().unwrap_or_default();
if f.fract() == 0.0 && f.abs() < 1e15 {
Some(format!("{}", f as i64))
} else {
Some(n.to_string())
}
}
other => Some(other.to_string()),
}
}
/// `String(r.col ?? "")`.
pub fn string_or_empty(row: &PgRow, col: &str) -> String {
get_string(row, col).unwrap_or_default()
}
+52
View File
@@ -0,0 +1,52 @@
//! Référentiels partagés : magasins et gammes (miroir de `src/lib/magasins.ts`
//! et `src/lib/gammes.ts` côté interface).
pub struct Magasin {
/// Code FF du site (clé de toutes les requêtes).
pub code: &'static str,
/// Nom complet.
pub nom: &'static str,
/// Nom court.
pub court: &'static str,
}
pub const MAGASINS: [Magasin; 2] = [
Magasin { code: "292", nom: "Frouard (Nancy)", court: "Frouard" },
Magasin { code: "579", nom: "Houdemont", court: "Houdemont" },
];
/// Code utilisé par la Grille et l'API pour « tous nos magasins ».
pub const CODE_TOUS_MAGASINS: &str = "TOTAL";
pub const LIBELLE_TOUS_MAGASINS: &str = "Nos 2 magasins";
/// Anciens codes encore présents dans certaines réponses de l'API FF.
fn alias(code: &str) -> &str {
match code {
"F01" => "292",
"F02" => "579",
c => c,
}
}
pub fn trouver_magasin(code: &str) -> Option<&'static Magasin> {
let c = alias(code.trim());
MAGASINS.iter().find(|m| m.code == c)
}
/// Nom lisible d'un magasin. `TOTAL` (ou vide) → « Nos 2 magasins ».
pub fn nom_magasin(code: Option<&str>, court: bool) -> String {
match code {
None | Some("") | Some(CODE_TOUS_MAGASINS) => LIBELLE_TOUS_MAGASINS.to_string(),
Some(c) => match trouver_magasin(c) {
Some(m) => (if court { m.court } else { m.nom }).to_string(),
None => c.to_string(),
},
}
}
/// Codes de gamme valides.
pub const CODES_GAMME: [&str; 5] = ["A", "B", "C", "Y", "Z"];
pub fn est_gamme_valide(code: &str) -> bool {
CODES_GAMME.contains(&code)
}
+105
View File
@@ -0,0 +1,105 @@
//! Erreurs HTTP communes.
//!
//! Les routes « internes » (appelées par l'interface) répondent `{ "error": "…" }`
//! avec le statut adéquat ; l'API publique `/api/v1` a sa propre enveloppe
//! (`crate::api_v1_response`).
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use axum::Json;
use serde_json::{json, Value};
#[derive(Debug)]
pub struct AppError {
pub status: StatusCode,
pub message: String,
/// Corps JSON complet à renvoyer à la place de `{ error }`.
pub body: Option<Value>,
}
pub type AppResult<T> = Result<T, AppError>;
impl AppError {
pub fn new(status: StatusCode, message: impl Into<String>) -> Self {
Self { status, message: message.into(), body: None }
}
pub fn bad_request(message: impl Into<String>) -> Self {
Self::new(StatusCode::BAD_REQUEST, message)
}
pub fn unauthorized(message: impl Into<String>) -> Self {
Self::new(StatusCode::UNAUTHORIZED, message)
}
pub fn forbidden(message: impl Into<String>) -> Self {
Self::new(StatusCode::FORBIDDEN, message)
}
pub fn not_found(message: impl Into<String>) -> Self {
Self::new(StatusCode::NOT_FOUND, message)
}
pub fn internal(message: impl Into<String>) -> Self {
Self::new(StatusCode::INTERNAL_SERVER_ERROR, message)
}
/// Réponse au corps libre (ex. `{ success: false, error }` avec un statut 500).
pub fn with_body(status: StatusCode, body: Value) -> Self {
let message = body.get("error").and_then(|v| v.as_str()).unwrap_or_default().to_string();
Self { status, message, body: Some(body) }
}
}
impl std::fmt::Display for AppError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{} {}", self.status, self.message)
}
}
impl std::error::Error for AppError {}
impl IntoResponse for AppError {
fn into_response(self) -> Response {
let body = self.body.unwrap_or_else(|| json!({ "error": self.message }));
(self.status, Json(body)).into_response()
}
}
impl From<anyhow::Error> for AppError {
fn from(e: anyhow::Error) -> Self {
// Une AppError enveloppée dans anyhow garde son statut.
match e.downcast::<AppError>() {
Ok(app) => app,
Err(e) => {
tracing::error!("[error] {e:#}");
Self::internal(format!("{e:#}"))
}
}
}
}
impl From<sqlx::Error> for AppError {
fn from(e: sqlx::Error) -> Self {
tracing::error!("[db] {e}");
Self::internal(e.to_string())
}
}
impl From<serde_json::Error> for AppError {
fn from(e: serde_json::Error) -> Self {
Self::bad_request(e.to_string())
}
}
impl From<std::io::Error> for AppError {
fn from(e: std::io::Error) -> Self {
Self::internal(e.to_string())
}
}
impl From<reqwest::Error> for AppError {
fn from(e: reqwest::Error) -> Self {
Self::internal(e.to_string())
}
}
/// Message d'une erreur quelconque tronqué à `n` caractères (pour les journaux).
pub fn short(e: &dyn std::fmt::Display, n: usize) -> String {
let s = e.to_string();
s.chars().take(n).collect()
}
+19
View File
@@ -0,0 +1,19 @@
//! Module `ff` — à porter.
use crate::rpc::Handler;
use axum::Router;
/// Routes HTTP du module.
pub fn router() -> Router {
Router::new()
}
/// Actions (`POST /api/actions/{nom}`) du module.
pub fn actions() -> Vec<(&'static str, Handler)> {
vec![]
}
/// Données de pages (`GET /api/pages/{nom}`) du module.
pub fn pages() -> Vec<(&'static str, Handler)> {
vec![]
}
+156
View File
@@ -0,0 +1,156 @@
//! Cache mémoire des lectures « FF » (base miroir et API FF Nancy).
//!
//! Les tables FF ne sont recopiées qu'une fois par nuit : recalculer à chaque
//! visite les mêmes agrégations lourdes coûtait plusieurs secondes par page.
//! - durée de vie bornée (`FF_CACHE_TTL_S`, 30 min par défaut) ;
//! - les appels simultanés pour la même clé partagent une seule requête ;
//! - une erreur n'est jamais mise en cache ;
//! - [`invalidate`] vide tout (fin de synchro nocturne).
use futures::future::{BoxFuture, FutureExt, Shared};
use parking_lot::Mutex;
use std::any::Any;
use std::collections::HashMap;
use std::future::Future;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, LazyLock};
use std::time::{Duration, Instant};
const MAX_ENTRIES: usize = 100;
type AnyArc = Arc<dyn Any + Send + Sync>;
type Pending = Shared<BoxFuture<'static, Result<AnyArc, Arc<String>>>>;
enum Entry {
Ready { value: AnyArc, expires_at: Instant, inserted: u64 },
Pending(Pending),
}
struct Store {
map: HashMap<String, Entry>,
seq: u64,
}
static STORE: LazyLock<Mutex<Store>> = LazyLock::new(|| Mutex::new(Store { map: HashMap::new(), seq: 0 }));
static GENERATION: AtomicU64 = AtomicU64::new(0);
pub fn default_ttl() -> Duration {
let s = std::env::var("FF_CACHE_TTL_S").ok().and_then(|v| v.parse::<u64>().ok()).filter(|v| *v > 0).unwrap_or(30 * 60);
Duration::from_secs(s)
}
/// Options d'une lecture mise en cache.
pub struct CacheOpts<T> {
/// Durée de vie propre à cette clé.
pub ttl: Option<Duration>,
/// Ne garder le résultat que s'il passe ce test (ex. liste non vide : une
/// lecture qui avale ses erreurs ne doit pas figer une panne passagère).
pub cache_if: Option<fn(&T) -> bool>,
}
impl<T> Default for CacheOpts<T> {
fn default() -> Self {
Self { ttl: None, cache_if: None }
}
}
/// Lecture mise en cache sous `key`.
pub async fn cached<T, F, Fut>(key: &str, opts: CacheOpts<T>, load: F) -> anyhow::Result<T>
where
T: Clone + Send + Sync + 'static,
F: FnOnce() -> Fut,
Fut: Future<Output = anyhow::Result<T>> + Send + 'static,
{
let pending = {
let mut store = STORE.lock();
match store.map.get(key) {
Some(Entry::Ready { value, expires_at, .. }) if *expires_at > Instant::now() => {
if let Some(v) = value.downcast_ref::<T>() {
return Ok(v.clone());
}
None
}
Some(Entry::Pending(p)) => Some(p.clone()),
_ => None,
}
};
let pending = match pending {
Some(p) => p,
None => {
let generation = GENERATION.load(Ordering::SeqCst);
let key_owned = key.to_string();
let ttl = opts.ttl.unwrap_or_else(default_ttl);
let cache_if = opts.cache_if;
let fut = load();
let shared: Pending = async move {
let res = fut.await;
let mut store = STORE.lock();
match res {
Ok(value) => {
let keep = cache_if.map(|f| f(&value)).unwrap_or(true);
let any: AnyArc = Arc::new(value);
if generation != GENERATION.load(Ordering::SeqCst) {
// Invalidation pendant le chargement : résultat rendu mais pas gardé.
} else if keep {
store.seq += 1;
let inserted = store.seq;
store.map.insert(
key_owned,
Entry::Ready { value: any.clone(), expires_at: Instant::now() + ttl, inserted },
);
evict(&mut store);
} else {
store.map.remove(&key_owned);
}
Ok(any)
}
Err(e) => {
if generation == GENERATION.load(Ordering::SeqCst) {
store.map.remove(&key_owned);
}
Err(Arc::new(format!("{e:#}")))
}
}
}
.boxed()
.shared();
STORE.lock().map.insert(key.to_string(), Entry::Pending(shared.clone()));
shared
}
};
match pending.await {
Ok(any) => any
.downcast_ref::<T>()
.cloned()
.ok_or_else(|| anyhow::anyhow!("ff_cache: type inattendu pour la clé {key}")),
Err(msg) => Err(anyhow::anyhow!("{msg}")),
}
}
fn evict(store: &mut Store) {
while store.map.len() > MAX_ENTRIES {
let oldest = store
.map
.iter()
.filter_map(|(k, e)| match e {
Entry::Ready { inserted, .. } => Some((k.clone(), *inserted)),
_ => None,
})
.min_by_key(|(_, i)| *i)
.map(|(k, _)| k);
match oldest {
Some(k) => {
store.map.remove(&k);
}
None => break,
}
}
}
/// Vide tout le cache FF (fin de synchro nocturne, données fraîches).
pub fn invalidate() {
GENERATION.fetch_add(1, Ordering::SeqCst);
STORE.lock().map.clear();
}
+19
View File
@@ -0,0 +1,19 @@
//! Module `grid` — à porter.
use crate::rpc::Handler;
use axum::Router;
/// Routes HTTP du module.
pub fn router() -> Router {
Router::new()
}
/// Actions (`POST /api/actions/{nom}`) du module.
pub fn actions() -> Vec<(&'static str, Handler)> {
vec![]
}
/// Données de pages (`GET /api/pages/{nom}`) du module.
pub fn pages() -> Vec<(&'static str, Handler)> {
vec![]
}
+225
View File
@@ -0,0 +1,225 @@
//! Capture des journaux serveur d'un job, pour téléchargement.
//!
//! Les extractions Qlik produisent des milliers de lignes ; la sortie standard
//! du conteneur les tronque. Tant qu'au moins une capture est ouverte, chaque
//! ligne journalisée (par n'importe quel code du processus) y est recopiée, puis
//! écrite sur disque au fil de l'eau (toutes les 2 s au plus).
//!
//! ⚠️ Deux jobs simultanés partagent les journaux du processus : chaque capture
//! ouverte reçoit TOUTES les lignes de la période.
use chrono::Utc;
use parking_lot::Mutex;
use serde::Serialize;
use std::collections::HashMap;
use std::fmt::Write as _;
use std::io::Write as _;
use std::path::PathBuf;
use std::sync::LazyLock;
use std::time::{Instant, SystemTime};
use tracing::field::{Field, Visit};
use tracing::{Event, Subscriber};
use tracing_subscriber::layer::Context;
use tracing_subscriber::Layer;
const MAX_LIGNES: usize = 200_000;
const MAX_LONGUEUR_LIGNE: usize = 8_000;
const MAX_FICHIERS: usize = 20;
const FLUSH_MS: u128 = 2_000;
struct Capture {
lignes: Vec<String>,
tronque: bool,
debut: Instant,
dernier_flush: Option<Instant>,
ecrites: usize,
}
static CAPTURES: LazyLock<Mutex<HashMap<String, Capture>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
fn log_dir() -> PathBuf {
std::env::temp_dir().join("collectflow-logs")
}
fn nom_sur(id: &str) -> String {
id.chars().map(|c| if c.is_ascii_alphanumeric() || "_.-".contains(c) { c } else { '_' }).collect()
}
fn fichier_pour(id: &str) -> PathBuf {
log_dir().join(format!("{}.log", nom_sur(id)))
}
fn horodatage() -> String {
Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string()
}
/// Écrit sur disque les lignes pas encore écrites (synchrone, appelé sous verrou).
fn flush(id: &str, c: &mut Capture) {
let _ = std::fs::create_dir_all(log_dir());
let total = c.lignes.len();
if c.ecrites == 0 || total > c.ecrites {
let nouvelles = &c.lignes[c.ecrites..total];
let mut texte = nouvelles.join("\n");
texte.push('\n');
let res = if c.ecrites == 0 {
std::fs::write(fichier_pour(id), texte)
} else {
std::fs::OpenOptions::new()
.append(true)
.open(fichier_pour(id))
.and_then(|mut f| f.write_all(texte.as_bytes()))
};
if res.is_ok() {
c.ecrites = total;
}
}
c.dernier_flush = Some(Instant::now());
}
/// Ajoute une ligne à toutes les captures ouvertes.
pub fn ajouter(ligne: &str) {
let mut captures = CAPTURES.lock();
if captures.is_empty() {
return;
}
let texte = if ligne.len() > MAX_LONGUEUR_LIGNE {
let mut cut = MAX_LONGUEUR_LIGNE;
while !ligne.is_char_boundary(cut) {
cut -= 1;
}
format!("{} …[ligne tronquée]", &ligne[..cut])
} else {
ligne.to_string()
};
let prefixe = format!("[{}] ", horodatage());
for (id, c) in captures.iter_mut() {
if c.lignes.len() >= MAX_LIGNES {
if !c.tronque {
c.tronque = true;
c.lignes.push(format!("{prefixe}[log-capture] limite de {MAX_LIGNES} lignes atteinte — suite non enregistrée"));
}
continue;
}
c.lignes.push(format!("{prefixe}{texte}"));
if c.dernier_flush.map(|t| t.elapsed().as_millis() >= FLUSH_MS).unwrap_or(true) {
flush(id, c);
}
}
}
fn purger() {
let Ok(rd) = std::fs::read_dir(log_dir()) else { return };
let mut fichiers: Vec<(PathBuf, SystemTime)> = rd
.filter_map(|e| e.ok())
.filter(|e| e.file_name().to_string_lossy().ends_with(".log"))
.map(|e| (e.path(), e.metadata().and_then(|m| m.modified()).unwrap_or(SystemTime::UNIX_EPOCH)))
.collect();
if fichiers.len() <= MAX_FICHIERS {
return;
}
fichiers.sort_by(|a, b| b.1.cmp(&a.1));
for (p, _) in fichiers.into_iter().skip(MAX_FICHIERS) {
let _ = std::fs::remove_file(p);
}
}
/// Ouvre une capture : toutes les lignes journalisées à partir de maintenant y
/// sont recopiées jusqu'à [`stop_capture`].
pub fn start_capture(id: &str, entete: Option<&str>) {
let mut c = Capture { lignes: Vec::new(), tronque: false, debut: Instant::now(), dernier_flush: None, ecrites: 0 };
c.lignes.push(format!("[{}] [log-capture] début de capture {id}", horodatage()));
if let Some(e) = entete {
c.lignes.push(format!("[{}] [log-capture] {e}", horodatage()));
}
flush(id, &mut c);
CAPTURES.lock().insert(id.to_string(), c);
purger();
}
/// Ferme la capture et écrit son état final sur disque.
pub fn stop_capture(id: &str, pied: Option<&str>) {
let Some(mut c) = CAPTURES.lock().remove(id) else { return };
let duree = c.debut.elapsed().as_secs();
let n = c.lignes.len();
c.lignes.push(format!("[{}] [log-capture] fin de capture {id} — {n} lignes, {duree} s", horodatage()));
if let Some(p) = pied {
c.lignes.push(format!("[{}] [log-capture] {p}", horodatage()));
}
flush(id, &mut c);
}
/// Contenu d'un journal : la capture en cours, sinon le fichier sur disque.
pub fn read_capture(id: &str) -> Option<String> {
if let Some(c) = CAPTURES.lock().get(id) {
return Some(c.lignes.join("\n") + "\n");
}
std::fs::read_to_string(fichier_pour(id)).ok()
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CaptureInfo {
pub id: String,
pub octets: u64,
pub modifie_le: String,
pub en_cours: bool,
}
/// Journaux disponibles, du plus récent au plus ancien.
pub fn list_captures() -> Vec<CaptureInfo> {
let en_cours: Vec<String> = CAPTURES.lock().keys().cloned().collect();
let mut out = Vec::new();
if let Ok(rd) = std::fs::read_dir(log_dir()) {
for e in rd.filter_map(|e| e.ok()) {
let nom = e.file_name().to_string_lossy().to_string();
let Some(id) = nom.strip_suffix(".log") else { continue };
let Ok(meta) = e.metadata() else { continue };
let modifie: chrono::DateTime<Utc> = meta.modified().map(Into::into).unwrap_or_else(|_| Utc::now());
out.push(CaptureInfo {
id: id.to_string(),
octets: meta.len(),
modifie_le: modifie.format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string(),
en_cours: en_cours.iter().any(|x| nom_sur(x) == id),
});
}
}
out.sort_by(|a, b| b.modifie_le.cmp(&a.modifie_le));
out
}
// ---------------------------------------------------------------------------
// Couche `tracing` qui alimente les captures
// ---------------------------------------------------------------------------
pub struct CaptureLayer;
struct MessageVisitor(String);
impl Visit for MessageVisitor {
fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
if field.name() == "message" {
let _ = write!(self.0, "{value:?}");
} else {
let _ = write!(self.0, " {}={value:?}", field.name());
}
}
fn record_str(&mut self, field: &Field, value: &str) {
if field.name() == "message" {
self.0.push_str(value);
} else {
let _ = write!(self.0, " {}={value}", field.name());
}
}
}
impl<S: Subscriber> Layer<S> for CaptureLayer {
fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
// Aucune capture ouverte (le cas courant) : on ne formate rien.
if CAPTURES.lock().is_empty() {
return;
}
let mut v = MessageVisitor(String::new());
event.record(&mut v);
ajouter(&v.0);
}
}
+108
View File
@@ -0,0 +1,108 @@
#![allow(dead_code)]
//! CollectFlow — serveur.
//!
//! Un binaire unique : API JSON (interface + `/api/v1` publique), accès à la
//! base PostgreSQL (tables de l'application et recopie FF Nancy), extraction
//! Qlik Sense, synchronisation nocturne, et service de l'interface compilée.
mod admin;
mod api_v1_response;
mod auth;
mod config;
mod db;
mod domain;
mod error;
mod ff;
mod ff_cache;
mod grid;
mod logcapture;
mod migrations;
mod produits;
mod qlik;
mod routes;
mod rpc;
mod spa;
use axum::extract::DefaultBodyLimit;
use axum::Router;
use rpc::Handler;
use tower_http::compression::CompressionLayer;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;
use tracing_subscriber::{EnvFilter, Layer};
/// Toutes les actions déclarées par les modules.
pub fn modules_actions() -> Vec<(&'static str, Handler)> {
[ff::actions(), qlik::actions(), grid::actions(), produits::actions(), admin::actions(), routes::actions()].concat()
}
/// Toutes les pages déclarées par les modules.
pub fn modules_pages() -> Vec<(&'static str, Handler)> {
[ff::pages(), qlik::pages(), grid::pages(), produits::pages(), admin::pages(), routes::pages()].concat()
}
fn init_tracing() {
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info,sqlx=warn,chromiumoxide=warn,tungstenite=warn,hyper=warn"));
let fmt = tracing_subscriber::fmt::layer().with_target(false).with_ansi(std::io::IsTerminal::is_terminal(&std::io::stdout())).with_filter(filter);
tracing_subscriber::registry().with(fmt).with(logcapture::CaptureLayer).init();
}
fn app() -> Router {
Router::new()
.merge(auth::router())
.merge(rpc::router())
.merge(ff::router())
.merge(qlik::router())
.merge(grid::router())
.merge(produits::router())
.merge(admin::router())
.merge(routes::router())
.merge(spa::router())
// Les instantanés de la Grille peuvent peser plusieurs Mo.
.layer(DefaultBodyLimit::max(64 * 1024 * 1024))
.layer(CompressionLayer::new())
}
async fn shutdown_signal() {
let ctrl_c = async {
let _ = tokio::signal::ctrl_c().await;
};
#[cfg(unix)]
let terminate = async {
if let Ok(mut s) = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
s.recv().await;
}
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
tokio::select! {
_ = ctrl_c => {},
_ = terminate => {},
}
tracing::info!("Arrêt demandé.");
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
init_tracing();
tracing::info!("CollectFlow {} — démarrage", env!("CARGO_PKG_VERSION"));
let _ = std::fs::create_dir_all(config::data_dir());
migrations::run().await;
tokio::spawn(async {
auth::ensure_admin_exists().await;
});
admin::start_background_tasks();
let port: u16 = std::env::var("PORT").ok().and_then(|p| p.parse().ok()).unwrap_or(5643);
let host = std::env::var("HOSTNAME").ok().filter(|h| !h.is_empty()).unwrap_or_else(|| "0.0.0.0".into());
let addr = format!("{host}:{port}");
let listener = match tokio::net::TcpListener::bind(&addr).await {
Ok(l) => l,
// HOSTNAME est parfois le nom du conteneur : on retombe sur toutes les interfaces.
Err(_) => tokio::net::TcpListener::bind(("0.0.0.0", port)).await?,
};
tracing::info!("Écoute sur http://{}", listener.local_addr()?);
axum::serve(listener, app()).with_graceful_shutdown(shutdown_signal()).await?;
Ok(())
}
+19
View File
@@ -0,0 +1,19 @@
//! Schéma de la base, joué à chaque démarrage (idempotent).
const INIT_SQL: &str = include_str!("../sql/init.sql");
pub async fn run() {
tracing::info!("[DB Init] Starting database migration & initialization...");
let pool = match crate::db::pool() {
Ok(p) => p,
Err(_) => {
tracing::warn!("[DB Init] No DATABASE_URL found in env or config. Skipping.");
return;
}
};
match sqlx::raw_sql(INIT_SQL).execute(&pool).await {
Ok(_) => tracing::info!("[DB Init] Initialization successful."),
// L'application reste utilisable (comptes de secours, Paramètres).
Err(e) => tracing::error!("[DB Init] Initialization failed: {e}"),
}
}
+19
View File
@@ -0,0 +1,19 @@
//! Module `produits` — à porter.
use crate::rpc::Handler;
use axum::Router;
/// Routes HTTP du module.
pub fn router() -> Router {
Router::new()
}
/// Actions (`POST /api/actions/{nom}`) du module.
pub fn actions() -> Vec<(&'static str, Handler)> {
vec![]
}
/// Données de pages (`GET /api/pages/{nom}`) du module.
pub fn pages() -> Vec<(&'static str, Handler)> {
vec![]
}
+19
View File
@@ -0,0 +1,19 @@
//! Module `qlik` — à porter.
use crate::rpc::Handler;
use axum::Router;
/// Routes HTTP du module.
pub fn router() -> Router {
Router::new()
}
/// Actions (`POST /api/actions/{nom}`) du module.
pub fn actions() -> Vec<(&'static str, Handler)> {
vec![]
}
/// Données de pages (`GET /api/pages/{nom}`) du module.
pub fn pages() -> Vec<(&'static str, Handler)> {
vec![]
}
+19
View File
@@ -0,0 +1,19 @@
//! Module `routes` — à porter.
use crate::rpc::Handler;
use axum::Router;
/// Routes HTTP du module.
pub fn router() -> Router {
Router::new()
}
/// Actions (`POST /api/actions/{nom}`) du module.
pub fn actions() -> Vec<(&'static str, Handler)> {
vec![]
}
/// Données de pages (`GET /api/pages/{nom}`) du module.
pub fn pages() -> Vec<(&'static str, Handler)> {
vec![]
}
+125
View File
@@ -0,0 +1,125 @@
//! Appels de l'interface vers le serveur hors routes REST classiques.
//!
//! - **Actions** — `POST /api/actions/{nom}`, corps = tableau JSON des arguments.
//! Remplacent les « server actions » Next.js : l'interface garde les mêmes
//! fonctions (`getUsers()`, `saveSnapshot(…)`…), devenues de simples appels
//! HTTP (cf. `src/lib/rpc.ts`). Une erreur renvoie `{ error }` avec un statut
//! 4xx/5xx, que le client relève comme une exception.
//! - **Pages** — `GET /api/pages/{nom}?…` : les données qu'une page serveur
//! Next.js calculait avant rendu (tableau de bord, hit-parade…), paramètres
//! d'URL compris.
//!
//! Chaque module déclare les siennes via `actions()` / `pages()`.
use crate::auth::{session, Claims, UtilisateurCourant};
use crate::error::{AppError, AppResult};
use axum::extract::{Path, Query};
use axum::http::{header, HeaderMap};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use futures::future::BoxFuture;
use serde::de::DeserializeOwned;
use serde::Serialize;
use serde_json::Value;
use std::collections::HashMap;
use std::sync::LazyLock;
/// Contexte d'un appel : jeton de session, paramètres d'URL, arguments.
#[derive(Clone, Default)]
pub struct Ctx {
pub token: Option<Claims>,
pub query: HashMap<String, String>,
pub args: Vec<Value>,
}
impl Ctx {
/// Utilisateur connecté (401 sinon).
pub async fn session(&self) -> AppResult<UtilisateurCourant> {
crate::auth::require_session(self.token.as_ref()).await
}
/// Administrateur connecté (401/403 sinon).
pub async fn admin(&self) -> AppResult<UtilisateurCourant> {
crate::auth::require_admin(self.token.as_ref()).await
}
/// Argument n° `i` désérialisé (`null` s'il est absent).
pub fn arg<T: DeserializeOwned>(&self, i: usize) -> AppResult<T> {
let v = self.args.get(i).cloned().unwrap_or(Value::Null);
serde_json::from_value(v).map_err(|e| AppError::bad_request(format!("Argument {i} invalide : {e}")))
}
/// Argument brut n° `i`.
pub fn raw(&self, i: usize) -> Value {
self.args.get(i).cloned().unwrap_or(Value::Null)
}
/// Paramètre d'URL (page), vide → `None`.
pub fn q(&self, name: &str) -> Option<&str> {
self.query.get(name).map(|s| s.as_str()).filter(|s| !s.is_empty())
}
}
pub type Handler = fn(Ctx) -> BoxFuture<'static, AppResult<Value>>;
/// Sérialise un résultat en JSON.
pub fn json<T: Serialize>(v: T) -> AppResult<Value> {
serde_json::to_value(v).map_err(|e| AppError::internal(e.to_string()))
}
/// Transforme une fonction `async fn(Ctx) -> AppResult<Value>` en [`Handler`].
#[macro_export]
macro_rules! handler {
($f:path) => {
(|c: $crate::rpc::Ctx| -> ::futures::future::BoxFuture<'static, $crate::error::AppResult<::serde_json::Value>> {
Box::pin($f(c))
}) as $crate::rpc::Handler
};
}
static ACTIONS: LazyLock<HashMap<&'static str, Handler>> = LazyLock::new(|| {
let mut m = HashMap::new();
for (name, h) in crate::modules_actions() {
if m.insert(name, h).is_some() {
panic!("action en double : {name}");
}
}
m
});
static PAGES: LazyLock<HashMap<&'static str, Handler>> = LazyLock::new(|| {
let mut m = HashMap::new();
for (name, h) in crate::modules_pages() {
if m.insert(name, h).is_some() {
panic!("page en double : {name}");
}
}
m
});
fn no_store(v: Value) -> Response {
([(header::CACHE_CONTROL, "no-store")], Json(v)).into_response()
}
async fn call_action(Path(name): Path<String>, headers: HeaderMap, body: Option<Json<Value>>) -> AppResult<Response> {
let h = ACTIONS.get(name.as_str()).ok_or_else(|| AppError::not_found(format!("Action inconnue : {name}")))?;
let args = match body.map(|b| b.0) {
Some(Value::Array(a)) => a,
Some(Value::Null) | None => vec![],
Some(other) => vec![other],
};
let ctx = Ctx { token: session::from_headers(&headers), query: HashMap::new(), args };
Ok(no_store(h(ctx).await?))
}
async fn call_page(Path(name): Path<String>, headers: HeaderMap, Query(query): Query<HashMap<String, String>>) -> AppResult<Response> {
let h = PAGES.get(name.as_str()).ok_or_else(|| AppError::not_found(format!("Page inconnue : {name}")))?;
let ctx = Ctx { token: session::from_headers(&headers), query, args: vec![] };
Ok(no_store(h(ctx).await?))
}
pub fn router() -> Router {
// Force la construction des registres au démarrage (doublons détectés tôt).
LazyLock::force(&ACTIONS);
LazyLock::force(&PAGES);
Router::new()
.route("/api/actions/{name}", post(call_action))
.route("/api/pages/{name}", get(call_page))
}
+73
View File
@@ -0,0 +1,73 @@
//! Service de l'interface (SPA React compilée par Vite) et garde des pages.
//!
//! Les fichiers existants du répertoire statique sont servis tels quels. Toute
//! autre URL hors `/api` renvoie `index.html` (routage côté navigateur), après
//! les mêmes contrôles que l'ancien middleware Next.js :
//! - non connecté → `/login` ;
//! - connecté sur `/login` ou `/` → `/dashboard` ;
//! - `/settings` et `/admin/*` réservés aux administrateurs.
use crate::auth::session;
use axum::body::Body;
use axum::http::{header, HeaderValue, Request, StatusCode, Uri};
use axum::response::{IntoResponse, Redirect, Response};
use axum::Router;
use std::path::PathBuf;
use tower_http::services::ServeDir;
use tower_http::set_header::SetResponseHeaderLayer;
pub fn static_dir() -> PathBuf {
std::env::var("STATIC_DIR").map(PathBuf::from).unwrap_or_else(|_| PathBuf::from("web"))
}
async fn page(req: Request<Body>) -> Response {
let uri: Uri = req.uri().clone();
let path = uri.path();
if path.starts_with("/api/") || path == "/api" {
return (StatusCode::NOT_FOUND, axum::Json(serde_json::json!({ "error": "Route inconnue" }))).into_response();
}
let claims = session::from_headers(req.headers());
let is_login = path == "/login";
let is_public = is_login || path.starts_with("/public/");
let user = match claims.as_ref() {
Some(_) => crate::auth::require_session(claims.as_ref()).await.ok(),
None => None,
};
match &user {
None if !is_public => return Redirect::to("/login").into_response(),
Some(_) if is_public || path == "/" => return Redirect::to("/dashboard").into_response(),
Some(u) if (path.starts_with("/settings") || path.starts_with("/admin")) && !u.is_admin() => {
return Redirect::to("/dashboard").into_response()
}
_ => {}
}
match tokio::fs::read(static_dir().join("index.html")).await {
Ok(html) => (
[
(header::CONTENT_TYPE, HeaderValue::from_static("text/html; charset=utf-8")),
(header::CACHE_CONTROL, HeaderValue::from_static("no-cache")),
],
html,
)
.into_response(),
Err(_) => (StatusCode::SERVICE_UNAVAILABLE, "Interface non compilée (index.html introuvable).").into_response(),
}
}
pub fn router() -> Router {
let dir = static_dir();
let assets = Router::new()
.nest_service("/assets", ServeDir::new(dir.join("assets")))
.layer(SetResponseHeaderLayer::overriding(
header::CACHE_CONTROL,
HeaderValue::from_static("public, max-age=31536000, immutable"),
));
let files = ServeDir::new(&dir)
.append_index_html_on_directories(false)
.fallback(tower::service_fn(|req: Request<Body>| async move { Ok::<_, std::convert::Infallible>(page(req).await) }));
assets.fallback_service(files)
}