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 | const { Client, Connection } = require('@temporalio/client');
const config = require('../config');
let temporalClient = null;
async function getTemporalClient() {
if (!temporalClient) {
const connection = await Connection.connect({ address: config.temporalAddress });
temporalClient = new Client({ connection });
}
return temporalClient;
}
async function createSchedule(scheduleId, cronExpression, workflowArgs) {
const client = await getTemporalClient();
await client.schedule.create({
scheduleId,
spec: {
cronExpressions: [cronExpression],
},
action: {
type: 'startWorkflow',
workflowType: 'SchedulerTW',
args: [workflowArgs],
taskQueue: 'scheduler-activities',
workflowExecutionTimeout: '1 hour',
},
});
}
async function pauseSchedule(scheduleId) {
const client = await getTemporalClient();
const handle = client.schedule.getHandle(scheduleId);
await handle.pause();
}
async function unpauseSchedule(scheduleId) {
const client = await getTemporalClient();
const handle = client.schedule.getHandle(scheduleId);
await handle.unpause();
}
async function deleteSchedule(scheduleId) {
const client = await getTemporalClient();
const handle = client.schedule.getHandle(scheduleId);
await handle.delete();
}
async function triggerSchedule(scheduleId) {
const client = await getTemporalClient();
const handle = client.schedule.getHandle(scheduleId);
await handle.trigger();
}
/**
* Réconciliation au démarrage : si un schedule Temporal existant porte encore un
* ancien type de workflow (ex. `CrawlerScheduleWorkflow` avant le suffixe `TW`),
* on le recrée avec le type courant. Idempotent et non bloquant.
*/
async function reconcileSchedules(schedules, expectedType = 'SchedulerTW') {
const client = await getTemporalClient();
let recreated = 0;
for (const s of schedules) {
if (!s.temporalScheduleId || !s.cronExpression) continue;
let currentType = null;
try {
const desc = await client.schedule.getHandle(s.temporalScheduleId).describe();
// `action` est à la racine de la description (ScheduleDescription.action).
currentType = desc.action?.workflowType || null;
} catch {
currentType = null; // schedule introuvable → à (re)créer
}
if (currentType === expectedType) continue;
try {
if (currentType) {
await client.schedule.getHandle(s.temporalScheduleId).delete();
}
await createSchedule(s.temporalScheduleId, s.cronExpression, {
scheduleId: s.temporalScheduleId,
organizationId: s.organizationId,
});
recreated++;
console.log(`[Scheduler] schedule ${s.temporalScheduleId} recréé (${currentType || 'absent'} → ${expectedType})`);
} catch (e) {
console.error(`[Scheduler] reconcile schedule ${s.temporalScheduleId} : ${e.message}`);
}
}
return recreated;
}
module.exports = {
getTemporalClient,
createSchedule,
pauseSchedule,
unpauseSchedule,
deleteSchedule,
triggerSchedule,
reconcileSchedules,
};
|