Skip to Content
TecnicaInfrastrutturaBullMQ Background Jobs System

BullMQ Background Jobs System

Panoramica

Sistema di gestione job in background basato su BullMQ e Redis per operazioni asincrone lunghe.

Architettura

Code Disponibili

  1. ai-analysis - Analisi AI (ricette, ottimizzazione costi, sprechi)
  2. invoice-import - Importazione fatture XML/PDF
  3. cassaincloud-sync - Sincronizzazione dati da Cassa in Cloud
  4. foodcost-recalculation - Ricalcolo food cost prodotti

Componenti

src/
├── lib/
│   └── queue.ts              # Configurazione code e helper
├── workers/
│   ├── index.ts              # Manager per avviare tutti i worker
│   ├── aiAnalysisWorker.ts
│   ├── invoiceImportWorker.ts
│   ├── cassaInCloudSyncWorker.ts
│   └── foodcostRecalculationWorker.ts
├── app/api/jobs/
│   └── route.ts              # API per creare e monitorare job
└── components/jobs/
    ├── JobMonitor.tsx        # UI per monitorare job
    └── JobLauncher.tsx       # UI per avviare job

Setup

1. Installazione Redis

# macOS con Homebrew
brew install redis
brew services start redis
 
# Docker
docker run -d -p 6379:6379 redis:alpine
 
# Linux
sudo apt-get install redis-server
sudo systemctl start redis

2. Configurazione Ambiente

Aggiungi a .env.local:

REDIS_HOST=localhost
REDIS_PORT=6379

3. Avviare i Worker

# Development
npm run workers
 
# Production
npm run workers:prod

Oppure con PM2 (configurato in ecosystem.config.js):

pm2 start ecosystem.config.js

Utilizzo

Creare un Job (API)

POST /api/jobs

const response = await fetch("/api/jobs", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    type: "ai-analysis",
    data: {
      analysisType: "recipe",
      productId: "507f1f77bcf86cd799439011",
    },
  }),
});
 
const { jobId, queueName } = await response.json();

Creare un Job (Programmaticamente)

import { addAIAnalysisJob } from "@/lib/queue";
 
const job = await addAIAnalysisJob({
  tenant_id: "tenant123",
  userId: "user456",
  analysisType: "recipe",
  productId: "product789",
});
 
console.log("Job creato:", job.id);

Monitorare un Job

GET /api/jobs?jobId=xxx&queue=aiAnalysis

const response = await fetch(`/api/jobs?jobId=${jobId}&queue=aiAnalysis`);
const jobStatus = await response.json();
 
console.log("Stato:", jobStatus.state); // active, completed, failed
console.log("Progresso:", jobStatus.progress); // 0-100
console.log("Risultato:", jobStatus.returnvalue);

Ottenere tutti i Job di una Coda

GET /api/jobs?queue=aiAnalysis&status=active

const response = await fetch(`/api/jobs?queue=aiAnalysis&status=completed`);
const { jobs, total } = await response.json();

Tipi di Job

1. AI Analysis

Tipi di analisi:

  • recipe - Analisi ricetta con suggerimenti alternativi
  • cost-optimization - Ottimizzazione costi fornitori
  • waste-analysis - Analisi sprechi e inefficienze

Esempio:

await addAIAnalysisJob({
  tenant_id: "...",
  userId: "...",
  analysisType: "recipe",
  productId: "507f1f77bcf86cd799439011",
  parameters: {
    considerSeasonality: true,
    maxAlternatives: 5,
  },
});

Output:

{
  "success": true,
  "analysisType": "recipe",
  "productId": "...",
  "suggestions": [
    {
      "ingredient": "Pomodoro",
      "alternative": "Pomodoro pelato",
      "savings": 0.5
    }
  ],
  "estimatedSavings": 5.5
}

2. Invoice Import

Importa fatture da file XML (FatturaPA) o PDF.

Esempio:

await addInvoiceImportJob({
  tenant_id: "...",
  userId: "...",
  fileUrl: "https://storage.example.com/invoices/IT123.xml",
  supplierId: "supplier123",
});

Output:

{
  "success": true,
  "invoiceId": "invoice789",
  "invoiceNumber": "FT-001",
  "supplier": "Fornitore Test",
  "totalAmount": 1000
}

Trigger automatico: Al termine, viene creato automaticamente un job di ricalcolo food cost.

3. Cassa in Cloud Sync

Sincronizza dati da Cassa in Cloud.

Tipi di sincronizzazione:

  • products - Solo prodotti
  • categories - Solo categorie
  • outlets - Solo punti vendita
  • sales - Solo vendite
  • full - Sincronizzazione completa

Esempio:

await addCassaInCloudSyncJob({
  tenant_id: "...",
  userId: "...",
  syncType: "full",
  dateFrom: "2024-01-01",
  dateTo: "2024-12-31",
  idsSalesPoint: [123, 456],
});

Output:

{
  "success": true,
  "syncType": "full",
  "results": {
    "outlets": { "fetched": 5, "created": 1, "updated": 4 },
    "categories": { "fetched": 25, "created": 2, "updated": 23 },
    "products": { "fetched": 150, "created": 10, "updated": 140 },
    "sales": { "fetched": 5420, "created": 5420 }
  },
  "syncedAt": "2024-11-22T10:30:00Z"
}

4. Foodcost Recalculation

Ricalcola i costi dei prodotti finiti.

Esempio:

await addFoodcostRecalculationJob({
  tenant_id: "...",
  userId: "...",
  recalculateAll: true,
  triggeredBy: "invoice-import", // o "price-update" o "manual"
});

Output:

{
  "success": true,
  "triggeredBy": "invoice-import",
  "total": 50,
  "recalculated": 48,
  "errors": 2,
  "changes": [
    {
      "productId": "...",
      "oldCost": 10.5,
      "newCost": 11.2,
      "variance": 6.67
    }
  ],
  "significantChanges": [
    /* variazioni > 5% */
  ]
}

Componenti UI

JobMonitor

Componente per monitorare i job in tempo reale.

import { JobMonitor } from "@/components/jobs/JobMonitor";
 
<JobMonitor />;

Features:

  • Selezione coda e stato
  • Auto-refresh ogni 5 secondi per job attivi
  • Progress bar per job in corso
  • Visualizzazione errori e risultati

JobLauncher

Componente per avviare job facilmente.

import { JobLauncher } from "@/components/jobs/JobLauncher";
 
<JobLauncher />;

Features:

  • Form per ogni tipo di job
  • Validazione input
  • Feedback immediato

Configurazione Worker

Concorrenza

// In worker file
const worker = new Worker("queue-name", handler, {
  connection,
  concurrency: 2, // Numero di job in parallelo
});

Retry e Backoff

await queue.add("job-name", data, {
  attempts: 3, // Numero tentativi
  backoff: {
    type: "exponential",
    delay: 5000, // Delay iniziale in ms
  },
});

Limiter

const worker = new Worker("queue-name", handler, {
  connection,
  limiter: {
    max: 10, // Massimo 10 job
    duration: 60000, // per minuto
  },
});

Monitoraggio e Debugging

Eventi Worker

worker.on("completed", job => {
  console.log(`Job ${job.id} completato`);
});
 
worker.on("failed", (job, err) => {
  console.error(`Job ${job?.id} fallito:`, err);
});
 
worker.on("progress", (job, progress) => {
  console.log(`Job ${job.id} progresso: ${progress}%`);
});
 
worker.on("error", err => {
  console.error("Worker error:", err);
});

Logs dentro il Job

async (job: Job) => {
  await job.log("Inizio elaborazione...");
  await job.updateProgress(50);
  await job.log("Metà del lavoro completata");
  // ...
};

BullMQ Board (Dashboard Web)

Installa BullMQ Board per UI di monitoraggio:

npm install @bull-board/express @bull-board/api

Scripts NPM

Aggiungi a package.json:

{
  "scripts": {
    "workers": "tsx watch src/workers/index.ts",
    "workers:prod": "tsx src/workers/index.ts"
  }
}

Gestione Errori

Retry Automatico

Job falliti vengono ritentati automaticamente secondo la configurazione:

{
  attempts: 3,
  backoff: {
    type: "exponential", // o "fixed"
    delay: 5000
  }
}

Dead Letter Queue

Dopo tutti i tentativi falliti, il job va in “failed” state e può essere reinserito manualmente:

const failedJobs = await queue.getFailed();
for (const job of failedJobs) {
  await job.retry();
}

Best Practices

  1. Progress Updates: Aggiorna il progress regolarmente per UX migliore
  2. Idempotenza: I job devono essere idempotenti (eseguibili più volte senza effetti collaterali)
  3. Timeout: Imposta timeout realistici per evitare job bloccati
  4. Logging: Logga informazioni utili per debugging
  5. Tenant Isolation: Ogni job deve includere tenant_id per isolamento multi-tenant
  6. Cleanup: Pulisci job completati/falliti periodicamente

Troubleshooting

Redis non raggiungibile

Error: connect ECONNREFUSED 127.0.0.1:6379

Soluzione: Verifica che Redis sia avviato:

redis-cli ping  # Deve rispondere "PONG"

Worker non processano job

Verifica:

  1. Worker è in esecuzione? ps aux | grep workers
  2. Redis è accessibile?
  3. Controlla i log del worker

Job bloccati in “active”

Soluzione:

// Pulisci job attivi stalled
await queue.clean(0, 1000, "active");

Produzione

PM2 Configuration

// ecosystem.config.js
module.exports = {
  apps: [
    {
      name: "foodcost-workers",
      script: "src/workers/index.ts",
      interpreter: "tsx",
      instances: 1,
      autorestart: true,
      watch: false,
      max_memory_restart: "1G",
      env: {
        NODE_ENV: "production",
      },
    },
  ],
};

Docker

# Worker container
FROM node:20-alpine
WORKDIR /app
COPY package*.json ./
RUN npm ci --production
COPY . .
CMD ["npm", "run", "workers:prod"]

Scalabilità

Per gestire più carico, avvia più istanze worker:

# PM2
pm2 start ecosystem.config.js -i 4
 
# Docker Compose
docker-compose up --scale workers=4

Ogni worker processerà job in parallelo dalla stessa coda Redis.

Last updated on
Bull Board - Implementazione Completata ✅BullMQ & Bull Board - Quick Start