418 lines
13 KiB
TypeScript
418 lines
13 KiB
TypeScript
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<boolean> {
|
|
const parsedTenantId = parseUlid(tenantId);
|
|
const parsedActorId = parseUlid(this.systemActorId);
|
|
const parsedEmployeeId = parseUlid(employeeId);
|
|
|
|
return this.database.tenantTransaction<boolean>(
|
|
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<DispatchReport> {
|
|
const parsedTenantId = parseUlid(tenantId);
|
|
const parsedActorId = parseUlid(this.systemActorId);
|
|
const parsedOutboxId = parseUlid(outboxId);
|
|
|
|
const outboxRow = await this.database.tenantTransaction<OutboxRow | undefined>(
|
|
parsedTenantId,
|
|
parsedActorId,
|
|
async (client) => {
|
|
const result = await client.query<OutboxRow>(
|
|
`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<SubscriptionRow[]>(
|
|
parsedTenantId,
|
|
parsedActorId,
|
|
async (client) => {
|
|
const result = await client.query<SubscriptionRow>(
|
|
`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<DeliveryRow | undefined>(
|
|
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<DeliveryRow>(
|
|
`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<void>(
|
|
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<void>(
|
|
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<DispatchReport[]> {
|
|
const parsedTenantId = parseUlid(tenantId);
|
|
const parsedActorId = parseUlid(this.systemActorId);
|
|
|
|
const pendingRows = await this.database.tenantTransaction<Array<{ id: string }>>(
|
|
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;
|
|
}
|
|
}
|