156 lines
4.6 KiB
JavaScript
156 lines
4.6 KiB
JavaScript
'use strict';
|
|
|
|
/**
|
|
* Task ID Generator Service
|
|
*
|
|
* Generates deterministic taskId and unique executionId for queue messages.
|
|
*
|
|
* Key Concepts:
|
|
* - taskId: Business identity (stable across retries) - enables deduplication
|
|
* - executionId: Execution identity (unique per attempt) - enables idempotency
|
|
*
|
|
* Simplified from 3 keys to 2 keys:
|
|
* - ❌ Old: taskId + idempotencyKey + correlationId
|
|
* - ✅ New: taskId + executionId (correlationId = taskId)
|
|
*/
|
|
|
|
const crypto = require('crypto');
|
|
const { v4: uuidv4 } = require('uuid');
|
|
const env = require('../helpers/env');
|
|
|
|
/**
|
|
* Generate deterministic taskId from message content
|
|
*
|
|
* @param {String} queueName - Queue name (e.g., 'dev_partner_tasks')
|
|
* @param {Object} message - Queue message payload
|
|
* @returns {String} taskId - Format: "{queueType}:{naturalKey}"
|
|
*
|
|
* Examples:
|
|
* - "partner_tasks:SATLOC:695d:02220710"
|
|
* - "jobs:12345:userId123:process"
|
|
* - "notifications:userId456:EMAIL:8a3f9c2e"
|
|
*/
|
|
function generateTaskId(queueName, message) {
|
|
// Normalize queue name (strip dev_ prefix for consistency)
|
|
const queueType = queueName.replace(/^dev_/, '');
|
|
|
|
switch (queueType) {
|
|
case env.QUEUE_NAME_PARTNER:
|
|
// Natural key: partnerCode + aircraftId + logId
|
|
if (!message.partnerCode || !message.aircraftId || !message.logId) {
|
|
throw new Error('Partner task missing required fields: partnerCode, aircraftId, logId');
|
|
}
|
|
return `${env.QUEUE_NAME_PARTNER}:${message.partnerCode}:${message.aircraftId}:${message.logId}`;
|
|
|
|
case env.QUEUE_NAME_JOBS:
|
|
// Natural key: appId + operation (each application import is unique)
|
|
// appId is the primary identifier for job imports
|
|
if (!message.appId) {
|
|
throw new Error('Job task missing required field: appId');
|
|
}
|
|
const operation = message.operation || message.updateOp || 'import';
|
|
return `${env.QUEUE_NAME_JOBS}:${message.appId}:${operation}`;
|
|
|
|
// case env.QUEUE_NAME_NOTIFICATIONS:
|
|
// // Natural key: userId + notificationType + content hash (first 8 chars)
|
|
// if (!message.userId || !message.type) {
|
|
// throw new Error('Notification task missing required fields: userId, type');
|
|
// }
|
|
// const contentHash = hashContent(message.content);
|
|
// return `notifications:${message.userId}:${message.type}:${contentHash}`;
|
|
|
|
default:
|
|
// Fallback: queue + timestamp + random (not deterministic, but functional)
|
|
return `${queueType}:${Date.now()}:${crypto.randomBytes(8).toString('hex')}`;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Generate unique executionId (UUID v4)
|
|
*
|
|
* @returns {String} executionId - UUID v4 format
|
|
*
|
|
* Used for:
|
|
* - Idempotency: Prevents duplicate processing if message redelivered
|
|
* - Retry tracking: Each retry gets new executionId but same taskId
|
|
*/
|
|
function generateExecutionId() {
|
|
return uuidv4();
|
|
}
|
|
|
|
/**
|
|
* Hash content for deterministic ID generation
|
|
*
|
|
* @param {*} content - Any JSON-serializable content
|
|
* @returns {String} 8-character hex hash
|
|
*/
|
|
function hashContent(content) {
|
|
const json = JSON.stringify(content || {});
|
|
return crypto.createHash('md5')
|
|
.update(json)
|
|
.digest('hex')
|
|
.substring(0, 8);
|
|
}
|
|
|
|
/**
|
|
* Extract queue type from queue name
|
|
*
|
|
* @param {String} queueName - Full queue name (e.g., 'dev_partner_tasks')
|
|
* @returns {String} Queue type without environment prefix
|
|
*/
|
|
function getQueueType(queueName) {
|
|
return queueName.replace(/^dev_/, '').replace(/^prod_/, '');
|
|
}
|
|
|
|
/**
|
|
* Parse taskId into components
|
|
*
|
|
* @param {String} taskId - Task ID to parse
|
|
* @returns {Object} Parsed components { queueType, parts }
|
|
*
|
|
* Example:
|
|
* parseTaskId("partner_tasks:SATLOC:695d:02220710")
|
|
* => { queueType: "partner_tasks", parts: ["SATLOC", "695d", "02220710"] }
|
|
*/
|
|
function parseTaskId(taskId) {
|
|
const [queueType, ...parts] = taskId.split(':');
|
|
return { queueType, parts };
|
|
}
|
|
|
|
/**
|
|
* Validate taskId format
|
|
*
|
|
* @param {String} taskId - Task ID to validate
|
|
* @returns {Boolean} true if valid format
|
|
*/
|
|
function isValidTaskId(taskId) {
|
|
if (!taskId || typeof taskId !== 'string') {
|
|
return false;
|
|
}
|
|
|
|
// Must have at least queue type and one component
|
|
const parts = taskId.split(':');
|
|
return parts.length >= 2;
|
|
}
|
|
|
|
/**
|
|
* Validate executionId format (UUID v4)
|
|
*
|
|
* @param {String} executionId - Execution ID to validate
|
|
* @returns {Boolean} true if valid UUID v4 format
|
|
*/
|
|
function isValidExecutionId(executionId) {
|
|
const uuidRegex = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
|
|
return uuidRegex.test(executionId);
|
|
}
|
|
|
|
module.exports = {
|
|
generateTaskId,
|
|
generateExecutionId,
|
|
hashContent,
|
|
getQueueType,
|
|
parseTaskId,
|
|
isValidTaskId,
|
|
isValidExecutionId
|
|
};
|