feat: Implement alert creation service with FCM notifications
- Add AlertCreationService to handle alert creation and notifications. - Implement idempotency key support to prevent duplicate alerts. - Create alert notification service for sending FCM messages. - Add tests for alert creation and notification services. - Document FCM notification flow and alert creation process.
This commit is contained in:
267
backend/src/services/alertCreationService.js
Normal file
267
backend/src/services/alertCreationService.js
Normal file
@@ -0,0 +1,267 @@
|
||||
const crypto = require('node:crypto');
|
||||
|
||||
const INSERT_ALERT_SQL = `
|
||||
WITH inserted AS (
|
||||
INSERT INTO alerts (route_id, alert_type, title, message, start_time, end_time, is_active)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7)
|
||||
RETURNING *
|
||||
)
|
||||
SELECT inserted.*, routes.route_number AS notification_route_number
|
||||
FROM inserted
|
||||
LEFT JOIN routes ON routes.id = inserted.route_id
|
||||
`;
|
||||
|
||||
function createHttpError(status, message) {
|
||||
const error = new Error(message);
|
||||
error.status = status;
|
||||
return error;
|
||||
}
|
||||
|
||||
function normalizeIdempotencyKey(value) {
|
||||
if (value === undefined || value === null) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const key = String(value).trim();
|
||||
|
||||
if (!key) {
|
||||
throw createHttpError(400, 'Idempotency-Key must not be empty');
|
||||
}
|
||||
|
||||
if (key.length > 255) {
|
||||
throw createHttpError(400, 'Idempotency-Key must not exceed 255 characters');
|
||||
}
|
||||
|
||||
return key;
|
||||
}
|
||||
|
||||
function normalizeAlertInput(payload) {
|
||||
return {
|
||||
route_id: payload.route_id || null,
|
||||
alert_type: payload.alert_type,
|
||||
title: payload.title,
|
||||
message: payload.message,
|
||||
start_time: payload.start_time,
|
||||
end_time: payload.end_time || null,
|
||||
is_active: payload.is_active !== false
|
||||
};
|
||||
}
|
||||
|
||||
function normalizeHashValue(value) {
|
||||
if (value instanceof Date) {
|
||||
return value.toISOString();
|
||||
}
|
||||
|
||||
return value;
|
||||
}
|
||||
|
||||
function createAlertRequestHash(alert) {
|
||||
const canonicalPayload = {
|
||||
route_id: normalizeHashValue(alert.route_id),
|
||||
alert_type: normalizeHashValue(alert.alert_type),
|
||||
title: normalizeHashValue(alert.title),
|
||||
message: normalizeHashValue(alert.message),
|
||||
start_time: normalizeHashValue(alert.start_time),
|
||||
end_time: normalizeHashValue(alert.end_time),
|
||||
is_active: normalizeHashValue(alert.is_active)
|
||||
};
|
||||
|
||||
return crypto
|
||||
.createHash('sha256')
|
||||
.update(JSON.stringify(canonicalPayload))
|
||||
.digest('hex');
|
||||
}
|
||||
|
||||
function splitNotificationRow(row) {
|
||||
const { notification_route_number: routeNumber, ...alert } = row;
|
||||
|
||||
return {
|
||||
alert,
|
||||
notificationAlert: {
|
||||
...alert,
|
||||
route_number: routeNumber ?? ''
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function parseAlertSnapshot(value) {
|
||||
if (!value) {
|
||||
return null;
|
||||
}
|
||||
|
||||
if (typeof value === 'string') {
|
||||
return JSON.parse(value);
|
||||
}
|
||||
|
||||
return value;
|
||||
}
|
||||
|
||||
class AlertCreationService {
|
||||
constructor({ pool, notificationService, logger }) {
|
||||
this.pool = pool;
|
||||
this.notificationService = notificationService;
|
||||
this.logger = logger;
|
||||
}
|
||||
|
||||
async createAlert(payload, { idempotencyKey: rawIdempotencyKey } = {}) {
|
||||
const alertInput = normalizeAlertInput(payload);
|
||||
const idempotencyKey = normalizeIdempotencyKey(rawIdempotencyKey);
|
||||
const result = idempotencyKey
|
||||
? await this.createIdempotentAlert(alertInput, idempotencyKey)
|
||||
: await this.insertAlert(this.pool, alertInput);
|
||||
|
||||
if (result.created && result.alert.is_active !== false) {
|
||||
await this.sendNotification(result.notificationAlert);
|
||||
}
|
||||
|
||||
return {
|
||||
alert: result.alert,
|
||||
created: result.created
|
||||
};
|
||||
}
|
||||
|
||||
async insertAlert(queryable, alertInput) {
|
||||
const result = await queryable.query(INSERT_ALERT_SQL, [
|
||||
alertInput.route_id,
|
||||
alertInput.alert_type,
|
||||
alertInput.title,
|
||||
alertInput.message,
|
||||
alertInput.start_time,
|
||||
alertInput.end_time,
|
||||
alertInput.is_active
|
||||
]);
|
||||
|
||||
const row = result.rows[0];
|
||||
|
||||
if (!row) {
|
||||
throw new Error('Alert insert did not return the created row');
|
||||
}
|
||||
|
||||
return {
|
||||
...splitNotificationRow(row),
|
||||
created: true
|
||||
};
|
||||
}
|
||||
|
||||
async createIdempotentAlert(alertInput, idempotencyKey) {
|
||||
const requestHash = createAlertRequestHash(alertInput);
|
||||
const client = await this.pool.connect();
|
||||
|
||||
try {
|
||||
await client.query('BEGIN');
|
||||
|
||||
const claimResult = await client.query(
|
||||
`INSERT INTO alert_creation_requests (idempotency_key, request_hash)
|
||||
VALUES ($1, $2)
|
||||
ON CONFLICT (idempotency_key) DO NOTHING
|
||||
RETURNING idempotency_key`,
|
||||
[idempotencyKey, requestHash]
|
||||
);
|
||||
|
||||
if (claimResult.rows.length === 0) {
|
||||
const existingRequestResult = await client.query(
|
||||
`SELECT request_hash, alert_id, response_body
|
||||
FROM alert_creation_requests
|
||||
WHERE idempotency_key = $1
|
||||
FOR UPDATE`,
|
||||
[idempotencyKey]
|
||||
);
|
||||
const existingRequest = existingRequestResult.rows[0];
|
||||
|
||||
if (!existingRequest) {
|
||||
throw new Error('Idempotency record disappeared during alert creation');
|
||||
}
|
||||
|
||||
if (existingRequest.request_hash !== requestHash) {
|
||||
throw createHttpError(
|
||||
409,
|
||||
'Idempotency-Key was already used with a different request payload'
|
||||
);
|
||||
}
|
||||
|
||||
const existingAlert = parseAlertSnapshot(existingRequest.response_body);
|
||||
|
||||
if (!existingAlert) {
|
||||
throw new Error('Idempotency record is missing its alert response snapshot');
|
||||
}
|
||||
|
||||
await client.query('COMMIT');
|
||||
|
||||
return {
|
||||
alert: existingAlert,
|
||||
notificationAlert: null,
|
||||
created: false
|
||||
};
|
||||
}
|
||||
|
||||
const created = await this.insertAlert(client, alertInput);
|
||||
|
||||
await client.query(
|
||||
`UPDATE alert_creation_requests
|
||||
SET alert_id = $2,
|
||||
response_body = $3::jsonb
|
||||
WHERE idempotency_key = $1`,
|
||||
[idempotencyKey, created.alert.id, JSON.stringify(created.alert)]
|
||||
);
|
||||
await client.query('COMMIT');
|
||||
|
||||
return created;
|
||||
} catch (error) {
|
||||
try {
|
||||
await client.query('ROLLBACK');
|
||||
} catch (rollbackError) {
|
||||
this.logger.error('Failed to roll back alert creation transaction', {
|
||||
reason: rollbackError.message
|
||||
});
|
||||
}
|
||||
|
||||
throw error;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
|
||||
async sendNotification(alert) {
|
||||
try {
|
||||
await this.notificationService.sendAlertCreated(alert);
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to dispatch alert-created notification', {
|
||||
alertId: String(alert.id),
|
||||
reason: error.message || 'Unknown Firebase error',
|
||||
firebaseCode: error.code || 'unknown'
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let singleton = null;
|
||||
|
||||
function getAlertCreationService() {
|
||||
if (!singleton) {
|
||||
const pool = require('../config/database');
|
||||
const logger = require('../config/logger');
|
||||
const { getAlertNotificationService } = require('./alertNotificationService');
|
||||
|
||||
singleton = new AlertCreationService({
|
||||
pool,
|
||||
logger,
|
||||
notificationService: getAlertNotificationService()
|
||||
});
|
||||
}
|
||||
|
||||
return singleton;
|
||||
}
|
||||
|
||||
function setAlertCreationServiceForTesting(service) {
|
||||
singleton = service;
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
AlertCreationService,
|
||||
createAlertRequestHash,
|
||||
getAlertCreationService,
|
||||
normalizeAlertInput,
|
||||
normalizeIdempotencyKey,
|
||||
parseAlertSnapshot,
|
||||
setAlertCreationServiceForTesting
|
||||
};
|
||||
Reference in New Issue
Block a user