Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x | const axios = require('axios');
const mongoose = require('mongoose');
const { Context } = require('@temporalio/activity');
const Schedule = require('../models/Schedule');
const config = require('../config');
const { getRedisClient } = require('../redis/client');
/**
* Registre des cibles planifiables : `type` -> constructeur d'appel (URL + body).
* Ajouter un nouveau type planifiable = ajouter une entrée ici.
* Le déclenchement passe toujours par le point d'entrée métier (API gateway),
* pas directement par Temporal (auth, config, pool, suivi `jobs`).
*/
const TARGETS = {
crawler: ({ gatewayUrl, targetId, scheduleId }) => ({
url: `${gatewayUrl}/api/v1/crawlers/${targetId}/start`,
data: { scheduleId: String(scheduleId) },
}),
workflow: ({ gatewayUrl, organizationId, targetId, payload }) => ({
url: `${gatewayUrl}/api/v1/${organizationId}/workflow/${targetId}/extract`,
data: { ...(payload || {}) },
}),
};
// TTL de la clé de dédup : couvre très largement la fenêtre de rejeu d'une
// exécution planifiée (3 tentatives, initialInterval 10 s, startToCloseTimeout 5 min).
const DEDUP_TTL_SECONDS = 3600;
/**
* Identifiant de l'exécution planifiée courante, **stable entre les tentatives**
* d'une même activité (retry Temporal) et **unique d'une exécution à l'autre**.
*
* On lit le `runId` du workflow porteur (`SchedulerTW`) depuis le contexte
* d'activité. `workflowInfo().scheduledTime` n'existe pas dans le SDK Temporal TS
* 1.23, et faire transiter l'heure de tir comme argument de l'activité
* modifierait le contrat workflow→activité (non-déterminisme au rejeu des
* exécutions en vol). Le `runId` évite ces deux écueils : aucune modification du
* workflow n'est nécessaire.
*/
function resolveExecutionToken() {
try {
const info = Context.current().info;
return (
info?.workflowExecution?.runId ||
(info?.scheduledTimestampMs != null ? String(info.scheduledTimestampMs) : null)
);
} catch (e) {
return null;
}
}
function buildDedupKey(scheduleId, token) {
return `scheduler:dedup:${scheduleId}:${token}`;
}
async function getScheduleApiKey(apiKeyId) {
const keyDoc = await mongoose.connection.db
.collection('apikeys')
.findOne({ _id: new mongoose.Types.ObjectId(apiKeyId) });
return keyDoc?.keyValue || null;
}
/**
* Activité générique déclenchée par `SchedulerTW` : résout la cible depuis la
* planification et appelle son point d'entrée (via l'API gateway).
*
* Idempotente au rejeu : une clé Redis `scheduler:dedup:<scheduleId>:<runId>`
* mémorise le `jobId` créé. Un retry Temporal (même exécution) relit cette clé
* et retourne le job existant au lieu d'en créer un second.
*/
async function executeScheduledTrigger(scheduleId) {
const schedule = await Schedule.findOne({ temporalScheduleId: scheduleId });
if (!schedule) throw new Error(`Schedule ${scheduleId} not found`);
if (!schedule.isActive) {
console.log(`[SchedulerActivity] Schedule ${scheduleId} paused, skipping`);
return { skipped: 'paused' };
}
const build = TARGETS[schedule.type];
if (!build) {
console.error(`[SchedulerActivity] Unknown schedule type: ${schedule.type}`);
schedule.lastRunStatus = 'failed';
await schedule.save();
return { skipped: 'unknown_type' };
}
// Dédup au rejeu : l'exécution planifiée est identifiée par son `runId`.
const token = resolveExecutionToken();
const dedupKey = token ? buildDedupKey(scheduleId, token) : null;
if (dedupKey) {
try {
const existingJobId = await getRedisClient().get(dedupKey);
if (existingJobId) {
console.log(
`[SchedulerActivity] Schedule ${scheduleId} déjà déclenché (run ${token}) → job ${existingJobId}, rejeu ignoré`,
);
return { deduped: true, jobId: existingJobId };
}
} catch (e) {
// La dédup est un filet : si Redis est indisponible, on déclenche quand même.
console.warn(`[SchedulerActivity] Dédup indisponible (lecture) : ${e.message}`);
}
}
const apiKey = await getScheduleApiKey(schedule.apiKeyId);
if (!apiKey) {
console.error(`[SchedulerActivity] Schedule ${scheduleId}: API key not found`);
schedule.lastRunStatus = 'failed';
await schedule.save();
return { skipped: 'no_api_key' };
}
const { url, data } = build({
gatewayUrl: config.gatewayUrl,
organizationId: schedule.organizationId,
targetId: schedule.targetId,
payload: schedule.payload,
scheduleId: schedule._id,
});
schedule.lastRunAt = new Date();
schedule.runCount = (schedule.runCount || 0) + 1;
try {
const res = await axios.post(url, data, {
headers: { 'x-api-key': apiKey },
timeout: 120000,
});
schedule.lastJobId = res.data?.data?.jobId || null;
schedule.lastRunStatus = 'success';
// Persiste la clé de dédup dès la création du job (avant la sauvegarde Mongo)
// pour fermer au plus tôt la fenêtre de rejeu.
if (dedupKey && schedule.lastJobId) {
try {
await getRedisClient().set(dedupKey, schedule.lastJobId, 'EX', DEDUP_TTL_SECONDS);
} catch (e) {
console.warn(`[SchedulerActivity] Dédup indisponible (écriture) : ${e.message}`);
}
}
console.log(
`[SchedulerActivity] ${schedule.type} triggered for schedule ${scheduleId} → job ${schedule.lastJobId || '?'}`,
);
} catch (e) {
console.error(`[SchedulerActivity] ${schedule.type} HTTP failed: ${e.message}`);
schedule.lastRunStatus = 'failed';
}
await schedule.save();
return { jobId: schedule.lastJobId, status: schedule.lastRunStatus };
}
module.exports = { executeScheduledTrigger, TARGETS, buildDedupKey, resolveExecutionToken };
|