fix: ajouter try/catch dans les blocs else de migrateExistingItems pour gérer les violations UNIQUE

- Ajout de scheduleRssFetch() dans server/_core/index.ts
- Planificateur démarré au lancement du serveur
- Supporte les modes interval et scheduled depuis rss_settings
- Rechargement dynamique lors de la sauvegarde des paramètres RSS
- Supprime la dépendance à la tâche planifiée Manus externe

Dockerfile

@@ -0,0 +1,27 @@
+FROM node:22-slim
+
+RUN corepack enable && corepack prepare pnpm@latest --activate
+
+WORKDIR /app
+
+# Installer les dépendances
+COPY package.json pnpm-lock.yaml ./
+COPY patches/ ./patches/
+RUN pnpm install --frozen-lockfile
+
+# Copier les sources
+COPY . .
+
+# Compiler le frontend (vite) et le backend (esbuild)
+RUN pnpm run build
+
+# Copier les migrations drizzle
+COPY drizzle/ ./drizzle/
+COPY drizzle.config.ts ./
+
+ENV NODE_ENV=production
+ENV PORT=3000
+
+EXPOSE 3000
+
+CMD ["node", "dist/index.js"]

drizzle/0006_rss_settings_scheduled_time.sql

@@ -0,0 +1,9 @@
+-- Migration: ajout colonnes scheduledTime et autoFetchEnabled dans rss_settings
+-- Alignement du schéma Drizzle avec la table existante
+
+ALTER TABLE rss_settings 
+  ADD COLUMN IF NOT EXISTS scheduledTime VARCHAR(5) DEFAULT '06:00',
+  ADD COLUMN IF NOT EXISTS autoFetchEnabled TINYINT(1) NOT NULL DEFAULT 1;
+
+UPDATE rss_settings SET autoFetchEnabled = isEnabled WHERE 1=1;
+UPDATE rss_settings SET scheduledTime = '06:00' WHERE scheduledTime IS NULL OR scheduledTime = '';

drizzle/schema.ts

@@ -66,8 +66,8 @@ export const veilleItems = mysqlTable("veille_items", {
   categorie: varchar("categorie", { length: 128 }),
   niveau: varchar("niveau", { length: 128 }),
   territoire: varchar("territoire", { length: 255 }),
-  // Liste JSON des territoires (multi-département) ex: ["Isère","Savoie"]
-  territoires: json("territoires").$type<string[]>(),
+  // Liste JSON des territoires (multi-département) ex: ["Isère","Savoie"] — stocké en text pour compatibilité Drizzle/MySQL
+  territoires: text("territoires"),
   resume: text("resume"),
   source: varchar("source", { length: 512 }),
   passage: text("passage"),
@@ -91,8 +91,8 @@ export const aapItems = mysqlTable("aap_items", {
   categorie: mysqlEnum("categorie", ["Handicap", "PA", "Enfance", "Précarité", "Sanitaire", "Autre"]).notNull(),
   region: varchar("region", { length: 255 }),
   departement: varchar("departement", { length: 255 }),
-  // Liste JSON des départements (multi-département) ex: ["Isère (38)","Savoie (73)"]
-  departements: json("departements").$type<string[]>(),
+  // Liste JSON des départements (multi-département) ex: ["Isère (38)","Savoie (73)"] — stocké en text pour compatibilité Drizzle/MySQL
+  departements: text("departements"),
   dateCloture: timestamp("dateCloture"),
   datePublication: timestamp("datePublication"),
   lien: text("lien"),

server/_core/index.ts

@@ -12,7 +12,8 @@ import { runFullImport } from "../importer";
 import uploadRoutes from "../uploadRoutes";
 import scheduledRoutes from "../scheduledRoutes";
 import { ensureAdminExists } from "../localAuth";
-import { getSetting } from "../db";
+import { getSetting, getRssSettings } from "../db";
+import { runRssFetch } from "../rssEngine";
 
 function isPortAvailable(port: number): Promise<boolean> {
   return new Promise(resolve => {
@@ -30,7 +31,6 @@ async function findAvailablePort(startPort: number = 3000): Promise<number> {
 }
 
 // ─── Tâche d'import quotidien ─────────────────────────────────────────────────
-
 let cronJob: ReturnType<typeof cron.schedule> | null = null;
 
 async function scheduleDailyImport() {
@@ -38,12 +38,10 @@ async function scheduleDailyImport() {
   const importTime = (await getSetting("import_time")) || "06:00";
   const [hour, minute] = importTime.split(":").map(Number);
   const cronExpr = `0 ${minute ?? 0} ${hour ?? 6} * * *`;
-
   if (cronJob) {
     cronJob.stop();
     cronJob = null;
   }
-
   cronJob = cron.schedule(cronExpr, async () => {
     console.log(`[Cron] Import automatique démarré à ${new Date().toISOString()}`);
     try {
@@ -53,10 +51,81 @@ async function scheduleDailyImport() {
       console.error("[Cron] Erreur lors de l'import:", e);
     }
   });
-
   console.log(`[Cron] Import quotidien planifié à ${importTime} (${cronExpr})`);
 }
 
+// ─── Planificateur RSS natif ──────────────────────────────────────────────────
+let rssCronJob: ReturnType<typeof cron.schedule> | null = null;
+
+/**
+ * Démarre (ou redémarre) le planificateur RSS en lisant la configuration
+ * depuis la table rss_settings. Peut être appelé au démarrage et à chaque
+ * modification des paramètres RSS via l'interface d'administration.
+ */
+export async function scheduleRssFetch() {
+  // Arrêter le cron existant s'il y en a un
+  if (rssCronJob) {
+    rssCronJob.stop();
+    rssCronJob = null;
+    console.log("[RSS Cron] Planificateur RSS arrêté.");
+  }
+
+  let settings = null;
+  try {
+    settings = await getRssSettings();
+  } catch (e) {
+    // En cas d'erreur (ex: colonne manquante lors d'une migration),
+    // utiliser les valeurs par défaut et planifier un retry dans 2 minutes
+    console.warn("[RSS Cron] Impossible de lire les paramètres RSS, utilisation des valeurs par défaut:", (e as Error).message);
+    setTimeout(() => scheduleRssFetch(), 2 * 60 * 1000);
+    // Démarrer quand même avec les valeurs par défaut
+    settings = { autoFetchEnabled: true, fetchMode: "interval" as const, fetchIntervalMinutes: 60, scheduledTime: "06:00" };
+  }
+
+  if (!settings || !settings.autoFetchEnabled) {
+    console.log("[RSS Cron] Lecture automatique des flux RSS désactivée.");
+    return;
+  }
+
+  let cronExpr: string;
+
+  if (settings.fetchMode === "interval") {
+    // Mode intervalle : toutes les N minutes
+    const intervalMin = Math.max(5, settings.fetchIntervalMinutes ?? 60);
+    if (intervalMin < 60) {
+      cronExpr = `*/${intervalMin} * * * *`;
+    } else if (intervalMin % 60 === 0) {
+      const hours = intervalMin / 60;
+      cronExpr = `0 */${hours} * * *`;
+    } else {
+      cronExpr = `*/${intervalMin} * * * *`;
+    }
+    console.log(`[RSS Cron] Mode intervalle — toutes les ${intervalMin} minutes (${cronExpr})`);
+  } else {
+    // Mode planifié : heure fixe quotidienne
+    const scheduledTime = settings.scheduledTime ?? "06:00";
+    const [hour, minute] = scheduledTime.split(":").map(Number);
+    cronExpr = `0 ${minute ?? 0} ${hour ?? 6} * * *`;
+    console.log(`[RSS Cron] Mode planifié — tous les jours à ${scheduledTime} (${cronExpr})`);
+  }
+
+  rssCronJob = cron.schedule(cronExpr, async () => {
+    console.log(`[RSS Cron] Lecture des flux RSS démarrée à ${new Date().toISOString()}`);
+    try {
+      const summary = await runRssFetch();
+      console.log(
+        `[RSS Cron] Lecture terminée — ${summary.totalFeeds} flux, ` +
+        `+${summary.totalNewItems} nouveaux articles, ` +
+        `${summary.errorFeeds} erreur(s)`
+      );
+    } catch (e) {
+      console.error("[RSS Cron] Erreur lors de la lecture des flux:", e);
+    }
+  });
+
+  console.log("[RSS Cron] Planificateur RSS démarré.");
+}
+
 async function startServer() {
   const app = express();
   const server = createServer(app);
@@ -67,7 +136,6 @@ async function startServer() {
   registerOAuthRoutes(app);
   app.use(uploadRoutes);
   app.use(scheduledRoutes);
-
   app.use(
     "/api/trpc",
     createExpressMiddleware({ router: appRouter, createContext })
@@ -81,18 +149,17 @@ async function startServer() {
 
   const preferredPort = parseInt(process.env.PORT || "3000");
   const port = await findAvailablePort(preferredPort);
-
   if (port !== preferredPort) {
     console.log(`Port ${preferredPort} is busy, using port ${port} instead`);
   }
 
   server.listen(port, async () => {
     console.log(`Server running on http://localhost:${port}/`);
-
     // Initialisation post-démarrage
     try {
       await ensureAdminExists();
       await scheduleDailyImport();
+      await scheduleRssFetch();
     } catch (e) {
       console.error("[Init] Erreur d'initialisation:", e);
     }

server/db.ts

@@ -1,5 +1,6 @@
 import { eq, desc, and, like, gte, lte, or, sql } from "drizzle-orm";
 import { drizzle } from "drizzle-orm/mysql2";
+import mysql from "mysql2/promise";
 import {
   InsertUser,
   users,
@@ -15,17 +16,25 @@ import {
   rssSettings,
   type InsertRssFeed,
   type InsertRssSettings,
+  type ImportLog,
   type RssFeed,
   type RssSettings,
 } from "../drizzle/schema";
 import { ENV } from "./_core/env";
 
-let _db: ReturnType<typeof drizzle> | null = null;
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+let _db: any = null;
 
 export async function getDb() {
   if (!_db && process.env.DATABASE_URL) {
     try {
-      _db = drizzle(process.env.DATABASE_URL);
+      const pool = mysql.createPool({
+        uri: process.env.DATABASE_URL,
+        waitForConnections: true,
+        connectionLimit: 10,
+        enableKeepAlive: true,
+      });
+      _db = drizzle(pool);
     } catch (error) {
       console.warn("[Database] Failed to connect:", error);
       _db = null;
@@ -174,9 +183,9 @@ export async function getVeilleDistinctValues() {
   ]);
 
   return {
-    categories: cats.map((r) => r.value!).filter(Boolean).sort(),
-    niveaux: niveaux.map((r) => r.value!).filter(Boolean).sort(),
-    territoires: territoires.map((r) => r.value!).filter(Boolean).sort(),
+    categories: cats.map((r: { value: string | null }) => r.value!).filter(Boolean).sort(),
+    niveaux: niveaux.map((r: { value: string | null }) => r.value!).filter(Boolean).sort(),
+    territoires: territoires.map((r: { value: string | null }) => r.value!).filter(Boolean).sort(),
   };
 }
 
@@ -241,8 +250,8 @@ export async function getAapDistinctValues() {
   ]);
 
   return {
-    regions: regions.map((r) => r.value!).filter(Boolean).sort(),
-    departements: departements.map((r) => r.value!).filter(Boolean).sort(),
+    regions: regions.map((r: { value: string | null }) => r.value!).filter(Boolean).sort(),
+    departements: departements.map((r: { value: string | null }) => r.value!).filter(Boolean).sort(),
   };
 }
 
@@ -310,9 +319,9 @@ export async function getImportStats() {
   ]);
 
   const total = allLogs.length;
-  const success = allLogs.filter(l => l.status === 'success').length;
-  const errors = allLogs.filter(l => l.status === 'error').length;
-  const totalNewRows = allLogs.reduce((sum, l) => sum + (l.newRows ?? 0), 0);
+  const success = allLogs.filter((l: ImportLog) => l.status === 'success').length;
+  const errors = allLogs.filter((l: ImportLog) => l.status === 'error').length;
+  const totalNewRows = allLogs.reduce((sum: number, l: ImportLog) => sum + (l.newRows ?? 0), 0);
 
   return {
     totalVeille: Number(veilleCount[0]?.count ?? 0),

server/routers.ts

@@ -33,6 +33,7 @@ import {
   saveRssSettings,
 } from "./db";
 import { importVeille, importAAP, runFullImport, getImportConfig } from "./importer";
+import { scheduleRssFetch } from "./_core/index";
 import { loginLocalUser, hashPassword, ensureAdminExists } from "./localAuth";
 
 // ─── Middleware admin ─────────────────────────────────────────────────────────
@@ -408,6 +409,8 @@ export const appRouter = router({
       }))
       .mutation(async ({ input }) => {
         await saveRssSettings(input);
+        // Recharger le planificateur RSS avec les nouveaux paramètres
+        await scheduleRssFetch();
         return { success: true };
       }),
   }),

server/rssEngine.ts

@@ -364,9 +364,7 @@ async function processFeed(feed: RssFeed): Promise<FetchResult> {
             categorie,
             niveau,
             territoire,
-            territoires: territoire !== "France" && territoire !== "Auvergne-Rhône-Alpes"
-              ? [territoire]
-              : [],
+            territoires: JSON.stringify(territoire !== "France" && territoire !== "Auvergne-Rhône-Alpes" ? [territoire] : []),
             resume: description || null,
             source: feed.name,
             lien: link || null,
@@ -375,7 +373,7 @@ async function processFeed(feed: RssFeed): Promise<FetchResult> {
           });
           result.newItems++;
         } catch (e: any) {
-          if (e?.code === "ER_DUP_ENTRY" || e?.message?.includes("Duplicate entry")) {
+          if (e?.code === "ER_DUP_ENTRY" || e?.cause?.code === "ER_DUP_ENTRY" || e?.message?.includes("Duplicate entry") || e?.cause?.message?.includes("Duplicate entry")) {
             // Article existant → ajouter le territoire à la liste si c'est un nouveau département
             if (territoire !== "France" && territoire !== "Auvergne-Rhône-Alpes") {
               await db.execute(
@@ -393,6 +391,7 @@ async function processFeed(feed: RssFeed): Promise<FetchResult> {
               result.skippedItems++;
             }
           } else {
+            console.error(`[RSS DEBUG veille] code=${e?.code} errno=${e?.errno} sqlMsg=${e?.sqlMessage} msg=${String(e?.message).substring(0,200)}`);
             throw e;
           }
         }
@@ -410,13 +409,13 @@ async function processFeed(feed: RssFeed): Promise<FetchResult> {
             categorie,
             region,
             departement,
-            departements: departement ? [departement] : [],
+            departements: JSON.stringify(departement ? [departement] : []),
             lien: link || null,
             datePublication: pubDate,
           });
           result.newItems++;
         } catch (e: any) {
-          if (e?.code === "ER_DUP_ENTRY" || e?.message?.includes("Duplicate entry")) {
+          if (e?.code === "ER_DUP_ENTRY" || e?.cause?.code === "ER_DUP_ENTRY" || e?.message?.includes("Duplicate entry") || e?.cause?.message?.includes("Duplicate entry")) {
             // Article existant → ajouter le département à la liste
             if (departement) {
               await db.execute(
@@ -504,18 +503,27 @@ export async function migrateExistingItems(): Promise<MigrationSummary> {
       const normalizedTitle = buildMergeKey(row.titre || "");
       const newDedupKey = dedupHash(normalizedTitle + "|veille");
 
+      try {
         await db.update(veilleItems)
         .set({
           categorie: newCategorie,
           niveau: newNiveau,
           territoire: newTerritoire,
-          territoires: newTerritoire !== "France" && newTerritoire !== "Auvergne-Rhône-Alpes"
-            ? [newTerritoire] : [],
+          territoires: JSON.stringify(newTerritoire !== "France" && newTerritoire !== "Auvergne-Rhône-Alpes" ? [newTerritoire] : []),
           titre: row.titre,
           dedupKey: newDedupKey,
         })
         .where(eq(veilleItems.id, row.id));
       veilleUpdated++;
+      } catch (e: any) {
+        // Si le nouveau dedupKey existe déjà → cet article est un doublon, le supprimer
+        if (e?.code === "ER_DUP_ENTRY" || e?.cause?.code === "ER_DUP_ENTRY" || e?.cause?.message?.includes("Duplicate entry")) {
+          await db.delete(veilleItems).where(eq(veilleItems.id, row.id));
+          veilleMerged++;
+        } else {
+          throw e;
+        }
+      }
     } else {
       // Groupe : fusionner en gardant le premier, supprimer les autres
       const sorted = group.sort((a: (typeof veilleRows)[number], b: (typeof veilleRows)[number]) => a.id - b.id);
@@ -538,12 +546,13 @@ export async function migrateExistingItems(): Promise<MigrationSummary> {
       const newDedupKey = dedupHash(normalizedTitle + "|veille");
 
       // Mettre à jour le principal
+      try {
         await db.update(veilleItems)
           .set({
             categorie: newCategorie,
             niveau: allTerritoires.length > 1 ? "departemental" : "regional",
             territoire: allTerritoires.length > 0 ? allTerritoires[0] : "Auvergne-Rhône-Alpes",
-          territoires: allTerritoires,
+            territoires: JSON.stringify(allTerritoires),
             titre: primary.titre,
             dedupKey: newDedupKey,
           })
@@ -555,6 +564,17 @@ export async function migrateExistingItems(): Promise<MigrationSummary> {
           veilleMerged++;
         }
         veilleUpdated++;
+      } catch (e: any) {
+        // Si le newDedupKey existe déjà → supprimer tout le groupe
+        if (e?.code === "ER_DUP_ENTRY" || e?.cause?.code === "ER_DUP_ENTRY" || e?.cause?.message?.includes("Duplicate entry")) {
+          for (const row of sorted) {
+            await db.delete(veilleItems).where(eq(veilleItems.id, row.id));
+            veilleMerged++;
+          }
+        } else {
+          throw e;
+        }
+      }
     }
   }
 
@@ -577,16 +597,26 @@ export async function migrateExistingItems(): Promise<MigrationSummary> {
       const normalizedTitle = buildMergeKey(row.titre || "");
       const newDedupKey = dedupHash(normalizedTitle + "|aap");
 
+      try {
         await db.update(aapItems)
         .set({
           region: newRegion,
           departement: newDept,
-          departements: newDept ? [newDept] : [],
+          departements: JSON.stringify(newDept ? [newDept] : []),
           titre: row.titre,
           dedupKey: newDedupKey,
         })
         .where(eq(aapItems.id, row.id));
       aapUpdated++;
+      } catch (e: any) {
+        // Si le nouveau dedupKey existe déjà → cet article est un doublon, le supprimer
+        if (e?.code === "ER_DUP_ENTRY" || e?.cause?.code === "ER_DUP_ENTRY" || e?.cause?.message?.includes("Duplicate entry")) {
+          await db.delete(aapItems).where(eq(aapItems.id, row.id));
+          aapMerged++;
+        } else {
+          throw e;
+        }
+      }
     } else {
       // Fusionner
       const sorted = group.sort((a: (typeof aapRows)[number], b: (typeof aapRows)[number]) => a.id - b.id);
@@ -604,11 +634,12 @@ export async function migrateExistingItems(): Promise<MigrationSummary> {
       const normalizedTitle = buildMergeKey(primary.titre || "");
       const newDedupKey = dedupHash(normalizedTitle + "|aap");
 
+      try {
         await db.update(aapItems)
           .set({
             region: "Auvergne-Rhône-Alpes",
             departement: allDepts.length > 0 ? allDepts[0] : null,
-          departements: allDepts,
+            departements: JSON.stringify(allDepts),
             titre: primary.titre,
             dedupKey: newDedupKey,
           })
@@ -619,6 +650,17 @@ export async function migrateExistingItems(): Promise<MigrationSummary> {
           aapMerged++;
         }
         aapUpdated++;
+      } catch (e: any) {
+        // Si le newDedupKey existe déjà → supprimer tout le groupe
+        if (e?.code === "ER_DUP_ENTRY" || e?.cause?.code === "ER_DUP_ENTRY" || e?.cause?.message?.includes("Duplicate entry")) {
+          for (const row of sorted) {
+            await db.delete(aapItems).where(eq(aapItems.id, row.id));
+            aapMerged++;
+          }
+        } else {
+          throw e;
+        }
+      }
     }
   }