import type { Database } from '../../infrastructure/database.js'; import { newUlid, parseUlid } from '../../core/ids.js'; import { type PushSubscriptionTarget, type VapidKeys, type WebPushOptions, sendWebPush, } from './web-push.js'; export interface PushSubscriptionInput { endpoint: string; p256dh: string; auth: string; user_agent?: string | undefined; } import { isPrivateOrLoopbackHost } from './web-push.js'; export class PushValidationError extends Error { readonly statusCode = 400; } export class PushConflictError extends Error { readonly statusCode = 409; } export interface PushDispatcherOptions { database: Database; vapidKeys: VapidKeys; systemActorId: string; fetchFn?: typeof fetch | undefined; } export interface DispatchDeliverySummary { endpoint: string; status: 'delivered' | 'failed' | 'skipped'; error?: string | undefined; } export interface DispatchReport { outboxId: string; subscriptionsFound: number; deliveries: DispatchDeliverySummary[]; published: boolean; } interface OutboxRow { id: string; tenant_id: string; event_type: string; subject_id: string; payload: unknown; } interface SubscriptionRow { id: string; endpoint: string; p256dh: string; auth: string; user_agent: string; } interface DeliveryRow { id: string; status: 'pending' | 'delivered' | 'failed'; attempts: number; } export class PushDispatcher { private readonly database: Database; private readonly vapidKeys: VapidKeys; private readonly systemActorId: string; private readonly fetchFn?: typeof fetch | undefined; constructor(options: PushDispatcherOptions) { this.database = options.database; this.vapidKeys = options.vapidKeys; this.systemActorId = options.systemActorId; this.fetchFn = options.fetchFn; } /** * Save or update an employee's WebPush subscription. */ async saveSubscription( tenantId: string, employeeId: string, input: PushSubscriptionInput, ): Promise<{ id: string; endpoint: string }> { const cleanEndpoint = input.endpoint.trim(); if (!cleanEndpoint.startsWith('https://')) { throw new PushValidationError('Subscription endpoint must use HTTPS'); } let parsedUrl: URL; try { parsedUrl = new URL(cleanEndpoint); } catch { throw new PushValidationError('Invalid endpoint URL'); } if (parsedUrl.protocol !== 'https:' || isPrivateOrLoopbackHost(parsedUrl.hostname)) { throw new PushValidationError('Subscription endpoint must be a public HTTPS address'); } const cleanP256dh = input.p256dh.trim(); const cleanAuth = input.auth.trim(); const userAgent = (input.user_agent ?? '').trim(); if (!cleanEndpoint || !cleanP256dh || !cleanAuth) { throw new PushValidationError('Endpoint, p256dh, and auth are required for WebPush subscription'); } const subscriptionId = newUlid(); const parsedTenantId = parseUlid(tenantId); const parsedActorId = parseUlid(this.systemActorId); const parsedEmployeeId = parseUlid(employeeId); return this.database.tenantTransaction<{ id: string; endpoint: string }>( parsedTenantId, parsedActorId, async (client) => { const existing = await client.query<{ id: string; employee_id: string }>( `SELECT id, employee_id FROM staff_push_subscriptions WHERE tenant_id = $1 AND endpoint = $2`, [parsedTenantId, cleanEndpoint], ); if (existing.rowCount && existing.rows[0]) { if (existing.rows[0].employee_id !== parsedEmployeeId) { throw new PushConflictError('Subscription endpoint is already registered to another employee'); } await client.query( `UPDATE staff_push_subscriptions SET p256dh = $3, auth = $4, user_agent = $5, updated_at = now() WHERE tenant_id = $1 AND endpoint = $2`, [parsedTenantId, cleanEndpoint, cleanP256dh, cleanAuth, userAgent], ); return { id: existing.rows[0].id, endpoint: cleanEndpoint }; } const result = await client.query<{ id: string; endpoint: string }>( `INSERT INTO staff_push_subscriptions (id, tenant_id, employee_id, endpoint, p256dh, auth, user_agent, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7, now()) RETURNING id, endpoint`, [ subscriptionId, parsedTenantId, parsedEmployeeId, cleanEndpoint, cleanP256dh, cleanAuth, userAgent, ], ); return result.rows[0]!; }, ); } /** * Remove an employee's WebPush subscription by endpoint. */ async deleteSubscription( tenantId: string, employeeId: string, endpoint: string, ): Promise { const parsedTenantId = parseUlid(tenantId); const parsedActorId = parseUlid(this.systemActorId); const parsedEmployeeId = parseUlid(employeeId); return this.database.tenantTransaction( parsedTenantId, parsedActorId, async (client) => { const result = await client.query( `DELETE FROM staff_push_subscriptions WHERE tenant_id = $1 AND employee_id = $2 AND endpoint = $3`, [parsedTenantId, parsedEmployeeId, endpoint], ); return (result.rowCount ?? 0) > 0; }, ); } /** * Dispatches push notifications for a specific transactional outbox record. * Ensures notification idempotency without changing the platform outbox publication state. */ async dispatchForOutbox(tenantId: string, outboxId: string): Promise { const parsedTenantId = parseUlid(tenantId); const parsedActorId = parseUlid(this.systemActorId); const parsedOutboxId = parseUlid(outboxId); const outboxRow = await this.database.tenantTransaction( parsedTenantId, parsedActorId, async (client) => { const result = await client.query( `SELECT id, tenant_id, event_type, subject_id, payload FROM staff_outbox WHERE tenant_id = $1 AND id = $2`, [parsedTenantId, parsedOutboxId], ); return result.rows[0]; }, ); if (!outboxRow) { return { outboxId, subscriptionsFound: 0, deliveries: [], published: false, }; } const employeeId = outboxRow.subject_id; const parsedEmployeeId = parseUlid(employeeId); // Fetch active subscriptions for this employee const subscriptions = await this.database.tenantTransaction( parsedTenantId, parsedActorId, async (client) => { const result = await client.query( `SELECT id, endpoint, p256dh, auth, user_agent FROM staff_push_subscriptions WHERE tenant_id = $1 AND employee_id = $2`, [parsedTenantId, parsedEmployeeId], ); return result.rows; }, ); // There is nothing to notify. The platform outbox remains available to its publisher. if (subscriptions.length === 0) { return { outboxId, subscriptionsFound: 0, deliveries: [], published: true, }; } // Neutral notification content according to Requirement 2: // No employee names, no salary or personal data in the push text. const notificationPayload = JSON.stringify({ title: 'Krona Staff', body: 'Отметка рабочего времени зарегистрирована', icon: '/staff/icon-192.svg', badge: '/staff/icon-192.svg', tag: 'attendance', data: { url: '/staff', event_type: outboxRow.event_type, }, }); const deliverySummaries: DispatchDeliverySummary[] = []; let deliveredCount = 0; let failedCount = 0; for (const sub of subscriptions) { // 1. Idempotency check in staff_notification_deliveries const deliveryRecord = await this.database.tenantTransaction( parsedTenantId, parsedActorId, async (client) => { const deliveryId = newUlid(); await client.query( `INSERT INTO staff_notification_deliveries (id, tenant_id, employee_id, outbox_id, endpoint, status, attempts) VALUES ($1, $2, $3, $4, $5, 'pending', 0) ON CONFLICT (tenant_id, outbox_id, endpoint) DO NOTHING`, [deliveryId, parsedTenantId, parsedEmployeeId, parsedOutboxId, sub.endpoint], ); const current = await client.query( `SELECT id, status, attempts FROM staff_notification_deliveries WHERE tenant_id = $1 AND outbox_id = $2 AND endpoint = $3`, [parsedTenantId, parsedOutboxId, sub.endpoint], ); return current.rows[0]; }, ); if (!deliveryRecord) continue; if (deliveryRecord.status === 'delivered') { deliverySummaries.push({ endpoint: sub.endpoint, status: 'skipped', }); deliveredCount++; continue; } if (deliveryRecord.attempts >= 5) { deliverySummaries.push({ endpoint: sub.endpoint, status: 'skipped', error: 'Max delivery attempts exceeded', }); failedCount++; continue; } // 2. Dispatch via WebPush const target: PushSubscriptionTarget = { endpoint: sub.endpoint, keys: { p256dh: sub.p256dh, auth: sub.auth }, }; const sendOptions: WebPushOptions = this.fetchFn ? { fetchFn: this.fetchFn } : {}; const result = await sendWebPush(target, notificationPayload, this.vapidKeys, sendOptions); if (result.successful) { await this.database.tenantTransaction( parsedTenantId, parsedActorId, async (client) => { await client.query( `UPDATE staff_notification_deliveries SET status = 'delivered', attempts = attempts + 1, delivered_at = now(), last_error = NULL, updated_at = now() WHERE tenant_id = $1 AND id = $2`, [parsedTenantId, deliveryRecord.id], ); }, ); deliverySummaries.push({ endpoint: sub.endpoint, status: 'delivered' }); deliveredCount++; } else { await this.database.tenantTransaction( parsedTenantId, parsedActorId, async (client) => { await client.query( `UPDATE staff_notification_deliveries SET status = 'failed', attempts = attempts + 1, last_error = $3, updated_at = now() WHERE tenant_id = $1 AND id = $2`, [parsedTenantId, deliveryRecord.id, result.error ?? 'Unknown delivery failure'], ); // Expired subscription cleanup on 404/410 if (result.shouldRemoveSubscription) { await client.query( `DELETE FROM staff_push_subscriptions WHERE tenant_id = $1 AND endpoint = $2`, [parsedTenantId, sub.endpoint], ); } }, ); deliverySummaries.push({ endpoint: sub.endpoint, status: 'failed', error: result.error, }); failedCount++; } } // The delivery table owns notification progress. staff_outbox.published_at is reserved // for the platform event publisher and must not be consumed by the push adapter. const shouldMarkPublished = deliveredCount > 0 || (deliveredCount === 0 && failedCount === 0); return { outboxId, subscriptionsFound: subscriptions.length, deliveries: deliverySummaries, published: shouldMarkPublished, }; } /** * Process all pending attendance outbox events. */ async dispatchPending(tenantId: string, limit = 20): Promise { const parsedTenantId = parseUlid(tenantId); const parsedActorId = parseUlid(this.systemActorId); const pendingRows = await this.database.tenantTransaction>( parsedTenantId, parsedActorId, async (client) => { const result = await client.query<{ id: string }>( `SELECT DISTINCT o.id FROM staff_outbox o JOIN staff_push_subscriptions s ON s.tenant_id = o.tenant_id AND s.employee_id = o.subject_id LEFT JOIN staff_notification_deliveries d ON d.tenant_id = o.tenant_id AND d.outbox_id = o.id AND d.endpoint = s.endpoint WHERE o.tenant_id = $1 AND o.event_type IN ('attendance.clock.in', 'attendance.clock.out') AND (d.id IS NULL OR (d.status = 'failed' AND d.attempts < 5)) ORDER BY o.id ASC LIMIT $2`, [parsedTenantId, limit], ); return result.rows; }, ); const reports: DispatchReport[] = []; for (const row of pendingRows) { const report = await this.dispatchForOutbox(tenantId, row.id); reports.push(report); } return reports; } }