feat: add traceability

api/config/custom-environment-variables.cjs

@@ -11,8 +11,10 @@ module.exports = {
   },
   port: 'PORT',
   privateDirectoryUrl: 'PRIVATE_DIRECTORY_URL',
+  privateEventsUrl: 'PRIVATE_EVENTS_URL',
   secretKeys: {
-    limits: 'SECRET_LIMITS'
+    limits: 'SECRET_LIMITS',
+    events: 'SECRET_EVENTS'
   },
   observer: {
     active: 'OBSERVER_ACTIVE',

api/config/default.cjs

@@ -8,8 +8,10 @@ module.exports = {
   },
   pluginCategories: ['Essentiels', 'Mes plugins', 'Données de références', 'Tests'],
   privateDirectoryUrl: 'http://simple-directory:8080',
+  privateEventsUrl: null,
   secretKeys: {
-    limits: null
+    limits: null,
+    events: null
   },
   mongoUrl: 'mongodb://localhost:27017/data-fair-processings',
   port: 8080,

api/config/development.cjs

@@ -2,6 +2,10 @@ module.exports = {
   dataDir: '../data/development',
   port: 8082,
   privateDirectoryUrl: 'http://localhost:8080',
+  privateEventsUrl: 'http://localhost:8088',
+  secretKeys: {
+    events: 'secret-events'
+  },
   mongoUrl: 'mongodb://localhost:27017/data-fair-processings-development',
   observer: {
     port: 9092

api/config/type/schema.json

@@ -50,11 +50,17 @@
       "type": "string",
       "pattern": "^https?://"
     },
+    "privateEventsUrl": {
+      "type": "string"
+    },
     "secretKeys": {
       "type": "object",
       "properties": {
         "limits": {
           "type": ["string", "null"]
+        },
+        "events": {
+          "type": ["string", "null"]
         }
       }
     },

api/package.json

@@ -23,7 +23,7 @@
   "dependencies": {
     "@data-fair/processings-shared": "*",
     "@data-fair/lib-express": "^1.12.6",
-    "@data-fair/lib-node": "^2.2.3",
+    "@data-fair/lib-node": "^2.6.0",
     "@data-fair/lib-utils": "^1.3.0",
     "ajv": "^8.17.1",
     "ajv-formats": "^3.0.1",

api/src/routers/processings.ts

@@ -10,6 +10,7 @@ import path from 'path'
 import resolvePath from 'resolve-path'
 import { nanoid } from 'nanoid'
 
+import eventsQueue from '@data-fair/lib-node/events-queue.js'
 import { session } from '@data-fair/lib-express/index.js'
 import { httpError } from '@data-fair/lib-utils/http-errors.js'
 import { createNext } from '@data-fair/processings-shared/runs.ts'
@@ -30,6 +31,33 @@ const pluginsDir = path.join(config.dataDir, 'plugins')
 
 const sensitiveParts = ['permissions', 'webhookKey', 'config']
 
+/**
+ * Helper function to send events related to processings
+ * @param processing The processing object
+ * @param actionText The text describing the action (e.g. "a été créé")
+ * @param topicAction The action part of the topic key (e.g. "create", "delete")
+ * @param sessionState Optional session state for authentication
+ */
+const sendProcessingEvent = (
+  processing: Processing,
+  actionText: string,
+  topicAction: string,
+  sessionState?: SessionStateAuthenticated
+) => {
+  if (!config.privateEventsUrl && !config.secretKeys.events) return
+
+  eventsQueue.pushEvent({
+    title: `Le traitement ${processing.title} ${actionText}`,
+    topic: { key: `processings:processing-${topicAction}:${processing._id}` },
+    sender: processing.owner,
+    resource: {
+      type: 'processing',
+      id: processing._id,
+      title: processing.title,
+    }
+  }, sessionState)
+}
+
 /**
  * Check that a processing object is valid
  * Check if the plugin exists
@@ -250,6 +278,8 @@ router.post('', async (req, res) => {
 
   await validateFullProcessing(processing)
   await mongo.processings.insertOne(processing)
+
+  sendProcessingEvent(processing, 'a été créé', 'create', sessionState)
   res.status(200).json(cleanProcessing(processing, sessionState))
 })
 
@@ -300,6 +330,9 @@ router.patch('/:id', async (req, res) => {
   await mongo.processings.updateOne({ _id: req.params.id }, patch)
   await mongo.runs.updateMany({ 'processing._id': processing._id }, { $set: { permissions: patchedProcessing.permissions || [] } })
   await applyProcessing(mongo, patchedProcessing)
+  if (config.privateEventsUrl && config.secretKeys.events) {
+    sendProcessingEvent(patchedProcessing, 'a été modifié', 'patch', sessionState)
+  }
   res.status(200).json(cleanProcessing(patchedProcessing, sessionState))
 })
 
@@ -320,6 +353,7 @@ router.delete('/:id', async (req, res) => {
   if (permissions.getUserResourceProfile(processing.owner, processing.permissions, sessionState) !== 'admin') return res.status(403).send()
   await mongo.processings.deleteOne({ _id: req.params.id })
   if (processing) await deleteProcessing(mongo, processing)
+  sendProcessingEvent(processing, 'a été supprimé', 'delete', sessionState)
   res.sendStatus(204)
 })
 
@@ -347,11 +381,13 @@ router.delete('/:id/webhook-key', async (req, res) => {
 router.post('/:id/_trigger', async (req, res) => {
   const processing = await mongo.processings.findOne({ _id: req.params.id })
   if (!processing) return res.status(404).send()
-  if (req.query.key && req.query.key !== processing.webhookKey) {
-    return res.status(403).send('Mauvaise clé de déclenchement')
+  if (req.query.key) {
+    if (req.query.key !== processing.webhookKey) return res.status(403).send('Mauvaise clé de déclenchement')
+    sendProcessingEvent(processing, 'a été déclenché par webhook', 'trigger')
   } else {
     const sessionState = await session.reqAuthenticated(req)
     if (!['admin', 'exec'].includes(permissions.getUserResourceProfile(processing.owner, processing.permissions, sessionState) ?? '')) return res.status(403).send()
+    sendProcessingEvent(processing, 'a été déclenché manuellement', 'trigger', sessionState)
   }
   if (!processing.active) return res.status(409).send('Le traitement n\'est pas actif')
   res.send(await createNext(mongo.db, locks, processing, true, req.query.delay ? Number(req.query.delay) : 0))

api/src/server.ts

@@ -4,6 +4,7 @@ import { app } from './app.ts'
 import { session, assertAuthenticated } from '@data-fair/lib-express/index.js'
 import * as wsServer from '@data-fair/lib-express/ws-server.js'
 import * as wsEmitter from '@data-fair/lib-node/ws-emitter.js'
+import eventsQueue from '@data-fair/lib-node/events-queue.js'
 import { startObserver, stopObserver } from '@data-fair/lib-node/observer.js'
 import { createHttpTerminator } from 'http-terminator'
 import { exec as execCallback } from 'child_process'
@@ -32,6 +33,13 @@ export const start = async () => {
   session.init(config.privateDirectoryUrl)
   await mongo.init()
   await locks.start(mongo.db)
+  if (config.privateEventsUrl) {
+    if (!config.secretKeys.events) {
+      console.error('Missing secretKeys.events in config')
+    } else {
+      await eventsQueue.start({ eventsUrl: config.privateEventsUrl, eventsSecret: config.secretKeys.events })
+    }
+  }
 
   await wsServer.start(server, mongo.db, async (channel, sessionState) => {
     assertAuthenticated(sessionState)

shared/package.json

@@ -5,7 +5,7 @@
     "@data-fair/lib-express": "^1.13.1"
   },
   "dependencies": {
-    "@data-fair/lib-node": "^2.1.0",
+    "@data-fair/lib-node": "^2.6.0",
     "@data-fair/lib-utils": "^1.1.0",
     "cron": "^3.1.9",
     "dayjs": "^1.11.13",

ui/package.json

@@ -30,7 +30,7 @@
     "unplugin-fonts": "^1.1.1",
     "unplugin-vue-components": "^0.27.4",
     "unplugin-vue-router": "^0.10.8",
-    "vite": "^6.2.0",
+    "vite": "^6.2.4",
     "vite-plugin-vuetify": "^2.1.0",
     "vue": "^3.5.13",
     "vue-i18n": "^11.1.2",

worker/package.json

@@ -12,14 +12,14 @@
     "#types": "./src/types.ts"
   },
   "devDependencies": {
-    "@data-fair/lib-common-types": "^1.5.4",
+    "@data-fair/lib-common-types": "^1.8.0",
     "@data-fair/lib-express": "^1.13.1",
     "@types/child-process-promise": "^2.2.6",
     "@types/nodemailer": "^6.4.16",
     "@types/resolve-path": "^1.4.3"
   },
   "dependencies": {
-    "@data-fair/lib-node": "^2.2.3",
+    "@data-fair/lib-node": "^2.6.0",
     "@data-fair/processings-shared": "*",
     "axios": "^1.8.3",
     "axios-retry": "^4.5.0",