const { routeIdsSql } = require('./alertTargets'); const CLAIM_DUE_ALERTS_SQL = ` WITH due AS ( SELECT a.id FROM alerts a WHERE a.is_active = true AND (a.start_time > a.created_at OR a.notification_pending = true) AND a.start_time <= CURRENT_TIMESTAMP AND (a.end_time IS NULL OR a.end_time > CURRENT_TIMESTAMP) AND a.notification_sent_at IS NULL AND a.notification_attempts < 5 AND ( a.notification_claimed_at IS NULL OR a.notification_claimed_at < CURRENT_TIMESTAMP - INTERVAL '5 minutes' ) ORDER BY a.start_time ASC, a.id ASC LIMIT $1 FOR UPDATE SKIP LOCKED ) UPDATE alerts a SET notification_claimed_at = CURRENT_TIMESTAMP, notification_attempts = a.notification_attempts + 1 FROM due WHERE a.id = due.id RETURNING a.*, ${routeIdsSql()} AS route_ids, CASE WHEN cardinality(a.route_numbers)=1 THEN a.route_numbers[1] END AS notification_route_number `; class AlertSchedulerService { constructor({ pool, notificationService, logger, batchSize = 50 } = {}) { this.pool = pool || require('../config/database'); this.notificationService = notificationService || require('./alertNotificationService').getAlertNotificationService(); this.logger = logger || require('../config/logger'); this.batchSize = batchSize; this.running = false; } async dispatchDueAlerts() { if (this.running || !this.notificationService.enabled) { return { dispatched: 0, skipped: this.running ? 'running' : 'disabled' }; } this.running = true; try { const claimed = await this.pool.query(CLAIM_DUE_ALERTS_SQL, [this.batchSize]); let dispatched = 0; for (const row of claimed.rows) { const { notification_route_number: routeNumber, ...alert } = row; try { const result = await this.notificationService.sendAlertCreated({ ...alert, route_number: routeNumber || '' }); if (result.sent) { await this.pool.query( `UPDATE alerts SET notification_sent_at = CURRENT_TIMESTAMP, notification_claimed_at = NULL WHERE id = $1`, [alert.id] ); dispatched += 1; } else { await this.releaseClaim(alert.id); } } catch (error) { await this.releaseClaim(alert.id); this.logger.error('Failed to dispatch scheduled alert notification', { alertId: String(alert.id), reason: error.message || 'Unknown notification error', firebaseCode: error.code || 'unknown' }); } } return { dispatched, claimed: claimed.rows.length }; } finally { this.running = false; } } async releaseClaim(alertId) { await this.pool.query( `UPDATE alerts SET notification_claimed_at = NULL WHERE id = $1 AND notification_sent_at IS NULL`, [alertId] ); } } function startAlertScheduler({ intervalMs = 30_000, service } = {}) { const scheduler = service || new AlertSchedulerService(); const run = () => scheduler.dispatchDueAlerts().catch((error) => { scheduler.logger.error('Scheduled alert delivery cycle failed', { reason: error.message || 'Unknown scheduler error' }); }); run(); const timer = setInterval(run, intervalMs); timer.unref?.(); return () => clearInterval(timer); } module.exports = { AlertSchedulerService, CLAIM_DUE_ALERTS_SQL, startAlertScheduler };