Skip to Content
TecnicaFunzionalitaSistema di Notifiche - Job Schedulati con BullMQ

Sistema di Notifiche - Job Schedulati con BullMQ

🎯 Overview

Le notifiche possono essere generate sia da eventi in tempo reale (import fatture, sync Cassa in Cloud) che da job schedulati in background usando BullMQ.

📋 Job Schedulati Attivi

Daily Checks (Controlli Giornalieri)

Schedule: Ogni giorno alle 2:00 AM
Queue: daily-checks
Worker: dailyChecksWorker

Controlla automaticamente:

  • ✅ Stock bassi: Genera notifiche per semilavorati con giacenza < 10 unità
  • ✅ Food cost elevato: Genera notifiche per prodotti con food cost > 40%

🔧 Architettura

┌─────────────────────┐
│ setupScheduler.ts   │ ← Configura repeatable job (cron pattern)
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│ Redis Queue         │ ← Memorizza job schedulati
│ (daily-checks)      │
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│ dailyChecksWorker   │ ← Processa job quando triggherati
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│ dailyChecks.ts      │ ← Logica business (query DB, genera notifiche)
└─────────────────────┘

📂 File Coinvolti

1. Job Logic

File: src/jobs/dailyChecks.ts

export async function checkInventoryJob(tenant_id?: string) {
  // Query prodotti con stock basso
  const lowStockProducts = await SellingProduct.find({
    isSemifinished: true,
    stockEstimate: { $lt: 10 },
  });
 
  // Genera notifiche
  for (const product of lowStockProducts) {
    await notifyLowStock(
      product.tenant_id,
      product.name,
      product.stockEstimate,
      product._id.toString(),
    );
  }
}
 
export async function analyzeFoodCostJob(tenant_id?: string) {
  // Query prodotti con food cost elevato
  // Genera notifiche per food cost > 40%
}

2. Worker (BullMQ)

File: src/workers/dailyChecksWorker.ts

const dailyChecksWorker = new Worker<DailyChecksJobData>(
  "daily-checks",
  async (job: Job) => {
    const { tenant_id, checks } = job.data;
 
    // Esegue i controlli
    const results = await dailyChecksJob(tenant_id, checks);
 
    return results;
  },
  { connection, concurrency: 2 },
);

3. Scheduler Setup

File: src/jobs/setupScheduler.ts

export async function setupDailyChecksScheduler() {
  await queues.dailyChecks.add(
    "daily-checks-scheduled",
    {
      checks: ["inventory", "foodcost"],
    },
    {
      repeat: {
        pattern: "0 2 * * *", // Cron: ogni giorno alle 2:00 AM
      },
      jobId: "daily-checks-recurring",
    },
  );
}

4. Queue Configuration

File: src/lib/queue.ts

export const queues = {
  dailyChecks: new Queue("daily-checks", { connection }),
  // ... altre queue
};
 
export interface DailyChecksJobData {
  tenant_id?: string; // Opzionale: specifico tenant o tutti
  checks?: Array<"inventory" | "foodcost">;
}
 
export async function addDailyChecksJob(data: DailyChecksJobData) {
  return queues.dailyChecks.add("daily-checks", data, {
    attempts: 2,
    backoff: { type: "exponential", delay: 10000 },
  });
}

5. API Endpoint (Manuale)

File: src/app/api/jobs/daily-checks/route.ts

// POST /api/jobs/daily-checks
// Permette di triggerare manualmente i controlli
export async function POST(request: Request) {
  const { tenant_id } = await requireAuth(request);
 
  const job = await addDailyChecksJob({ tenant_id });
 
  return NextResponse.json({ jobId: job.id });
}

🚀 Setup e Avvio

1. Avvia i Worker

npm run workers
# oppure in produzione
npm run workers:prod

Il comando avvia:

  • Tutti i worker BullMQ (incluso dailyChecksWorker)
  • Lo scheduler automatico (setupScheduler)

2. Verifica Scheduler

# Entra in Redis CLI
redis-cli
 
# Vedi le chiavi delle queue
KEYS bull:daily-checks:*
 
# Vedi i job schedulati
ZRANGE bull:daily-checks:repeat 0 -1 WITHSCORES

3. Test Manuale

# Triggera i controlli subito (per tutti i tenant)
curl -X POST http://localhost:3000/api/jobs/daily-checks \
  -H "Authorization: Bearer YOUR_JWT" \
  -H "Content-Type: application/json"
 
# Solo per un tenant specifico
curl -X POST http://localhost:3000/api/jobs/daily-checks \
  -H "Authorization: Bearer YOUR_JWT" \
  -H "Content-Type: application/json" \
  -d '{"tenant_id":"67614d6a885c080b3cd45e9d"}'
 
# Solo controllo inventario
curl -X POST http://localhost:3000/api/jobs/daily-checks \
  -H "Authorization: Bearer YOUR_JWT" \
  -H "Content-Type: application/json" \
  -d '{"checks":["inventory"]}'

⚙️ Configurazione

Cambia Schedule

Modifica src/jobs/setupScheduler.ts:

repeat: {
  pattern: "0 2 * * *",  // 2:00 AM ogni giorno
  // pattern: "0 */6 * * *",  // Ogni 6 ore
  // pattern: "*/5 * * * *",  // Ogni 5 minuti (test)
}

Aggiungi Nuovi Controlli

  1. Aggiungi funzione in src/jobs/dailyChecks.ts
  2. Importa helper da src/lib/notifications.ts
  3. Aggiorna dailyChecksJob() per chiamare la nuova funzione
  4. Aggiorna tipo checks in DailyChecksJobData

Esempio:

export async function checkInvoicesDueJob(tenant_id?: string) {
  const dueInvoices = await Invoice.find({
    dueDate: { $lte: new Date() },
    status: "unpaid",
  });
 
  for (const invoice of dueInvoices) {
    await notifySystemMessage(
      invoice.tenant_id,
      "Fattura in scadenza",
      `Fattura ${invoice.number} scade oggi`,
      "high",
    );
  }
}

🔍 Monitoring

Dashboard (Future)

# Installa Bull Board (opzionale)
npm install @bull-board/express @bull-board/api

Log dei Job

I worker loggano automaticamente:

✅ [DailyChecksWorker] Job 123 completed
❌ [DailyChecksWorker] Job 456 failed: Connection timeout

Redis Insight

Usa Redis Insight  per visualizzare:

  • Job in coda
  • Job completati/falliti
  • Repeatable jobs
  • Job data e results

🎯 Esempi di Notifiche Generate

Stock Basso

{
  title: "Stock Basso",
  message: "Mozzarella di Bufala ha solo 5 unità rimanenti",
  type: "warning",
  priority: "medium",
  category: "inventory",
  actionUrl: "/semilavorati/67614e6b885c080b3cd45f2a",
  actionLabel: "Vedi Prodotto"
}

Food Cost Elevato

{
  title: "Food Cost Elevato",
  message: "Pizza Margherita ha un food cost del 45.2%",
  type: "warning",
  priority: "high",
  category: "costs",
  actionUrl: "/prodotti-finiti/67614e6b885c080b3cd45f31",
  actionLabel: "Vedi Prodotto"
}

📊 Metriche

Il sistema traccia:

  • Numero di prodotti controllati
  • Notifiche generate
  • Errori durante l’elaborazione
  • Tempo di esecuzione
{
  inventory: {
    success: true,
    productsChecked: 45,
    notificationsCreated: 3
  },
  foodCost: {
    success: true,
    productsAnalyzed: 128,
    notificationsCreated: 7
  }
}

🔐 Multi-Tenant

Il job supporta due modalità:

  1. Singolo Tenant (specificando tenant_id)

    • Utile per esecuzioni manuali
    • Permette controlli on-demand

  2. Tutti i Tenant (senza tenant_id)

    • Usato dallo scheduler automatico
    • Processa ogni prodotto col proprio tenant_id
// Tutti i tenant
await addDailyChecksJob({});
 
// Singolo tenant
await addDailyChecksJob({ tenant_id: "123" });

🚨 Gestione Errori

  • ✅ Retry automatico: 2 tentativi con backoff esponenziale
  • ✅ Non-blocking: Errori in un controllo non fermano gli altri
  • ✅ Logging: Tutti gli errori sono loggati
  • ✅ Isolamento: Errori di notifica non fermano il job
try {
  await notifyLowStock(...);
} catch (notifErr) {
  console.error("Errore notifica:", notifErr);
  // Il job continua
}

📝 Checklist Deployment

  • Redis in esecuzione (redis-server)
  • MongoDB connesso
  • Worker avviati (npm run workers:prod)
  • Scheduler configurato (verifica con Redis CLI)
  • Variabili d’ambiente settate (REDIS_HOST, MONGODB_URI)
  • Monitoring attivo (logs, Redis Insight)
  • Test manuale eseguito con successo

🎉 Vantaggi BullMQ

✅ Persistenza: Job sopravvivono a restart
✅ Retry: Gestione automatica dei fallimenti
✅ Scheduling: Cron patterns flessibili
✅ Concurrency: Processa job in parallelo
✅ Monitoring: Dashboard e Redis tools
✅ Scalabilità: Worker distribuiti su più server

Last updated on
Sistema di NotificheGestione Punti Vendita (Outlets) con Cassa in Cloud