Jobs et tâches planifiées

La Cafête utilise Apalis (backed by Redis) pour la file d'exécution des workers et tokio_cron_scheduler pour les déclencheurs cron qui poussent les jobs vers Apalis.

Lancement

cargo run -p caffette-worker

Jobs planifiés

Job	Schedule (cron)	Fréquence	Advisory lock
outbox_consumer	*/10 * * * * *	10 s	—
process_uncollected	*/10 * * * * *	10 s	104
deactivate_products	0 */5 * * * *	5 min	100
reconcile_payments	0 */5 * * * *	5 min	110
auto_close_registers	0 */5 * * * *	5 min	108
daily_reports	0 */5 * * * *	5 min	109
check_expiry	0 0 * * * *	1 h	—
reconcile_stock	0 0 */6 * * *	6 h	—
reconcile_ledger	0 0 3 * * *	03:00	—
extract_anomalies	0 0 6 * * *	06:00	200

Crons à 5 minutes

Plusieurs jobs sont programmés toutes les 5 minutes. La logique métier (auto_close_registers, daily_reports) vérifie elle-même si l'heure courante correspond à l'heure configurée pour l'organisation avant d'agir.

Architecture

flowchart LR
    S[tokio_cron_scheduler] -->|Push job| R[(Apalis / Redis)]
    R --> W[Worker pool]
    W --> J1[outbox_consumer]
    W --> J2[extract_anomalies]
    W --> J3[reconcile_stock]
    W --> J4[...]
    J1 --> PG[(PostgreSQL)]
    J2 --> PG
    J3 --> PG

Advisory locks

Les jobs sensibles utilisent pg_try_advisory_lock(1, <id>) pour empêcher les exécutions concurrentes entre plusieurs instances du worker :

Lock ID	Job
100	deactivate_products
104	process_uncollected
108	auto_close_registers
109	daily_reports
110	reconcile_payments
200	extract_anomalies

Détail des jobs

outbox_consumer

Traite les événements de la table outbox_events (pending → processing → published/failed). Route chaque événement vers son handler : RechargeCompleted et AnomalyDetected créent des notifications ; PurchaseCompleted et ReservationCreated sont journalisés.

process_uncollected

Détecte les réservations non retirées et génère les notifications/anomalies associées. Lock 104.

deactivate_products

Désactive les produits expirés. Lock 100.

reconcile_payments

Réconcilie les sessions de paiement Mobile Money en attente avec le fournisseur. Lock 110.

auto_close_registers

Ferme automatiquement les caisses ayant dépassé l'heure de fermeture configurée. Lock 108.

daily_reports

Envoie les rapports quotidiens par email (protégé par les settings daily_report_*). Lock 109.

check_expiry

Détecte les produits périmés, crée un mouvement de rebut et une anomalie expired_product. Horaire.

reconcile_stock

Compare products.stock_quantity avec la somme des stock_movements et crée une anomalie stock_inconsistency en cas d'écart. Toutes les 6 h.

reconcile_ledger

Contrôle detective en lecture seule : compare le solde de chaque carte avec la somme de ses transactions (transaction REPEATABLE READ). En cas d'écart, logge une erreur (ERROR) — n'écrit pas d'anomalie. Quotidien à 03:00.

extract_anomalies

flowchart TD
    A[06:00] --> B[pg_try_advisory_lock 1,200]
    B --> C[Pour chaque org active]
    C --> D[Lire default_anomaly_threshold_cents]
    D --> E[SELECT cards WHERE balance_cents < seuil<br/>ET non notifiée depuis 8 jours]
    E --> F[Sévérité: high si <0, medium si <seuil/2, sinon low]
    F --> G[INSERT anomaly + event AnomalyDetected]
    G --> H[Outbox → notification email]

Lock 200.

Logs

tail -f logs/worker.log | grep "anomaly"
grep "ERROR" logs/worker.log

Alertes recommandées

Condition	Action
extract_anomalies échoue 3 jours de suite	Email admin
outbox_consumer a > 100 events failed	Dashboard warning
reconcile_stock détecte un écart	Investiguer les mouvements manuels

---

Voir aussi

• Anomalies — types et cycle de vie
• Notifications — Setup