diff --git a/config.template.jsonc b/config.template.jsonc index 50943b407..cbc0f45f7 100644 --- a/config.template.jsonc +++ b/config.template.jsonc @@ -350,6 +350,10 @@ // not, a row past this is swept. Off by default; any value above 0 arms // the sweep. // "notificationRetentionDays": 14, + // + // How long a live notification waits for the recipient's open tabs to + // acknowledge it, in ms, before its delivery is reported as unreceived. + // "notificationAckTimeoutMs": 5000, // ── Alarms / alerting ─────────────────────────────────────────────── // Where system alarms go. Severity is the routing decision — each diff --git a/src/backend/clients/event/types.ts b/src/backend/clients/event/types.ts index f4351817d..267ffb975 100644 --- a/src/backend/clients/event/types.ts +++ b/src/backend/clients/event/types.ts @@ -578,6 +578,18 @@ export type EventMap = { wire: 'notif.message' | 'notif.unreads' | 'notif.ack'; response: unknown; }; + /** + * How one live `notif.message` went, on the node that sent it. `received` + * once a socket acknowledged it within the window, `unreceived` when none + * did; `ackCapable` false means no socket there acknowledges at all. + */ + 'notif.delivery.settled': { + uid: string; + userId: number; + outcome: NotifDeliveryOutcome; + ackCapable: boolean; + responders: NotifDeliveryResponder[]; + }; // ---- Metering ---- // Recurring charges are pure mechanism here: the metering service knows @@ -997,8 +1009,22 @@ export type MatchingEvents

= P extends `${infer Prefix}.*` ? Extract : P & EventKey; +/** + * Where a live `notif.message` ended up: acknowledged, sent but not + * acknowledged, no socket anywhere, or sockets only in a peer region. + */ +export type NotifDeliveryOutcome = + 'received' | 'unreceived' | 'no-socket' | 'remote-only'; + +/** One socket's receipt for a `notif.message`. */ +export interface NotifDeliveryResponder { + visibilityState: 'visible' | 'hidden' | 'unknown'; +} + export type EventMetadata = { from_outside?: boolean; + /** The emitter sent it to this region's sockets; peers still deliver it. */ + region_delivered?: boolean; /** Set by `emitAndWait` when a listener threw, so a verdict hook can refuse. */ listener_failed?: boolean; }; diff --git a/src/backend/services/events/notifications.integration.test.ts b/src/backend/services/events/notifications.integration.test.ts index 875c3cb31..8775737ce 100644 --- a/src/backend/services/events/notifications.integration.test.ts +++ b/src/backend/services/events/notifications.integration.test.ts @@ -103,6 +103,14 @@ const withServer = (config: IConfig) => { state.sent.push({ room: spec.room ?? spec.socket, key, data }); }, ); + vi.spyOn( + state.server.services.socket, + 'sendWithAck', + ).mockImplementation(async (specifier, key, data) => { + const spec = Array.isArray(specifier) ? specifier[0] : specifier; + state.sent.push({ room: spec.room ?? spec.socket, key, data }); + return { responses: [], timedOut: false, sockets: 0, acking: 0 }; + }); state.server.services.events.onDelivered = (envelope) => { state.delivered.push(envelope); }; @@ -332,6 +340,35 @@ describe('notifications with the fold-in on', () => { ); }); + it('settles the delivery once, on the region that sent it', async () => { + const user = await makeUser(state.server); + const settled: Array<{ uid: string; outcome: string }> = []; + const handler = (_k: string, data: unknown) => { + const d = data as { uid: string; userId: number; outcome: string }; + if (d.userId === user.id) settled.push(d); + }; + state.server.clients.event.on('notif.delivery.settled', handler); + try { + const uid = await state.server.services.notification.notify( + [user.id], + { title: 'receipt' }, + { type: 'share.received' }, + ); + await waitFor(() => settled.length > 0); + expect(settled).toEqual([ + expect.objectContaining({ uid, outcome: 'no-socket' }), + ]); + + await waitFor(() => forwardedFor(user.id).length > 0); + await applyAsPeer(state.server, forwardedFor(user.id)[0]); + await settle(); + expect(notifSends(state.sent)).toHaveLength(2); + expect(settled).toHaveLength(1); + } finally { + state.server.clients.event.off?.('notif.delivery.settled', handler); + } + }); + it('delivers to a subscription on the mailbox slice, and to no other', async () => { const user = await makeUser(state.server); const appUid = await makeApp(state.server, user.id); @@ -591,6 +628,28 @@ describe('notifications with the fold-in off', () => { expect(dispatched).toEqual([]); expect(state.delivered).toEqual([]); state.server.clients.event.off?.(NOTIF_DELIVERY_EVENT, handler); + + // The peer region still gets the GUI event and sends its own copy. + await waitFor(() => + forwarded().some( + (event) => + event.key === 'outer.gui.notif.message' && + (event.data as { user_id_list?: number[] }) + .user_id_list?.[0] === user.id, + ), + ); + const forward = forwarded().find( + (event) => + event.key === 'outer.gui.notif.message' && + (event.data as { user_id_list?: number[] }) + .user_id_list?.[0] === user.id, + )!; + await applyAsPeer(state.server, forward); + await settle(); + expect(notifSends(state.sent)).toHaveLength(2); + expect(notifSends(state.sent)[1].data).toEqual( + notifSends(state.sent)[0].data, + ); }); }); diff --git a/src/backend/services/events/registry.ts b/src/backend/services/events/registry.ts index 3544c68e1..58aeccb3f 100644 --- a/src/backend/services/events/registry.ts +++ b/src/backend/services/events/registry.ts @@ -448,6 +448,10 @@ export const UNPUBLISHED_INTERNAL_EVENTS = [ event: 'notif.written', reason: 'A node-local hook after a notification is written, for extensions; the notification itself publishes through `notif.created`.', }, + { + event: 'notif.delivery.settled', + reason: 'A node-local report on how the live socket copy of a notification went; the notification itself publishes through `notif.created`.', + }, ] as const satisfies readonly UnpublishedInternalEvent[]; // Every `fs.*`/`kv.*`/`notif.*` bus key needs a decision. A new key that is in diff --git a/src/backend/services/notification/NotificationService.ts b/src/backend/services/notification/NotificationService.ts index a9835c005..702590899 100644 --- a/src/backend/services/notification/NotificationService.ts +++ b/src/backend/services/notification/NotificationService.ts @@ -24,6 +24,7 @@ import type { NotifEventContext } from '../events/registry.js'; import { PuterService } from '../types.js'; import { NotificationSocketAdapter, + notificationAckTimeoutMs, notificationsFoldInEnabled, } from './notificationSocket.js'; import { @@ -320,10 +321,23 @@ export class NotificationService extends PuterService { event: this.clients.event, socket: this.services.socket, foldIn: () => notificationsFoldInEnabled(this.config), + ackTimeoutMs: () => notificationAckTimeoutMs(this.config), + heldInPeerRegion: (userId) => this.#heldInPeerRegion(userId), }); return this.#adapter; } + /** A presence read that errors counts as no peer socket. */ + async #heldInPeerRegion(userId: number): Promise { + const eventForward = this.services.eventForward; + if (!eventForward?.active) return false; + try { + return (await eventForward.regionsFor(userId, null)).length > 0; + } catch { + return false; + } + } + /** * Publish one persisted notification. With the fold-in on, the events layer * owns the delivery and feeds the socket adapter; without it, the adapter diff --git a/src/backend/services/notification/notificationDelivery.test.ts b/src/backend/services/notification/notificationDelivery.test.ts new file mode 100644 index 000000000..7ddaf0827 --- /dev/null +++ b/src/backend/services/notification/notificationDelivery.test.ts @@ -0,0 +1,245 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import { io as ioClient, type Socket as ClientSocket } from 'socket.io-client'; +import { + afterAll, + afterEach, + beforeAll, + describe, + expect, + it, + vi, +} from 'vitest'; +import type { EventMap } from '../../clients/event/types.js'; +import type { PuterServer } from '../../server.js'; +import { + allocateEphemeralPort, + createTestUser, + setupTestServer, + type TestUserCredentials, +} from '../../testUtil.js'; + +type Settled = EventMap['notif.delivery.settled']; + +const ACK_TIMEOUT_MS = 300; + +describe('live notification delivery receipts', () => { + let server: PuterServer; + let origin: string; + let user: TestUserCredentials; + let userId: number; + const openSockets: ClientSocket[] = []; + const settled: Settled[] = []; + const onSettled = (_key: string, data: unknown) => { + settled.push(data as Settled); + }; + + beforeAll(async () => { + const port = await allocateEphemeralPort(); + server = await setupTestServer( + { port, notificationAckTimeoutMs: ACK_TIMEOUT_MS } as never, + { listen: true }, + ); + origin = `http://puter.localhost:${port}`; + user = await createTestUser(server, { + username: 'receipt-user', + password: 'receipt-user-password', + }); + userId = (await server.stores.user.getByUsername(user.username))!.id; + server.clients.event.on('notif.delivery.settled', onSettled); + }, 60_000); + + afterEach(async () => { + vi.restoreAllMocks(); + for (const socket of openSockets.splice(0)) socket.disconnect(); + await vi.waitFor(() => + expect(server.services.socket.has({ room: userId })).toBe(false), + ); + settled.length = 0; + }); + + afterAll(async () => { + server?.clients.event.off('notif.delivery.settled', onSettled); + await server?.shutdown(); + }, 60_000); + + const connect = (auth: Record): Promise => + new Promise((resolve, reject) => { + const socket = ioClient(origin, { + auth: { auth_token: user.token, ...auth }, + transports: ['websocket'], + reconnection: false, + }); + openSockets.push(socket); + socket.on('connect', () => resolve(socket)); + socket.on('connect_error', reject); + }); + + const notify = async ( + type: 'share.received' | 'app.message' = 'share.received', + ): Promise => { + const uid = await server.services.notification.notify( + [userId], + { title: 'hello' }, + type === 'app.message' + ? { type, appUid: 'app-receipts' } + : { type }, + ); + return uid!; + }; + + const settledFor = async (uid: string): Promise => { + await vi.waitFor( + () => expect(settled.some((s) => s.uid === uid)).toBe(true), + { timeout: 3_000 }, + ); + return settled.find((s) => s.uid === uid)!; + }; + + it('settles as received once a tab acknowledges, with its visibility', async () => { + const tab = await connect({ acks: ['notif.message'] }); + const got: unknown[] = []; + tab.on('notif.message', (msg: unknown, ack: (r: unknown) => void) => { + got.push(msg); + ack({ visibilityState: 'hidden' }); + }); + + const uid = await notify(); + + expect(await settledFor(uid)).toEqual({ + uid, + userId, + outcome: 'received', + ackCapable: true, + responders: [{ visibilityState: 'hidden' }], + }); + expect(got).toEqual([ + { + uid, + notification: { title: 'hello', type: 'share.received' }, + }, + ]); + }); + + it('settles as unreceived when no tab acknowledges within the window', async () => { + const tab = await connect({ acks: ['notif.message'] }); + const got = new Promise((resolve) => + tab.once('notif.message', resolve), + ); + + const started = Date.now(); + const uid = await notify(); + const result = await settledFor(uid); + + expect(Date.now() - started).toBeGreaterThanOrEqual( + ACK_TIMEOUT_MS - 50, + ); + expect(result).toMatchObject({ + outcome: 'unreceived', + ackCapable: true, + responders: [], + }); + await expect(got).resolves.toMatchObject({ uid }); + }); + + it('reports a socket that never offered to acknowledge as not ack-capable', async () => { + const tab = await connect({}); + const got = new Promise((resolve) => + tab.once('notif.message', resolve), + ); + + const uid = await notify(); + + expect(await settledFor(uid)).toMatchObject({ + outcome: 'unreceived', + ackCapable: false, + responders: [], + }); + await expect(got).resolves.toMatchObject({ uid }); + }); + + it('settles as no-socket when the user has no connection anywhere', async () => { + const uid = await notify(); + expect(await settledFor(uid)).toEqual({ + uid, + userId, + outcome: 'no-socket', + ackCapable: false, + responders: [], + }); + }); + + it('settles as remote-only when only a peer region holds a socket', async () => { + const eventForward = server.services.eventForward; + vi.spyOn(eventForward, 'active', 'get').mockReturnValue(true); + vi.spyOn(eventForward, 'regionsFor').mockResolvedValue(['peer-a']); + + const uid = await notify(); + + expect(await settledFor(uid)).toMatchObject({ + outcome: 'remote-only', + ackCapable: false, + }); + expect(eventForward.regionsFor).toHaveBeenCalledWith(userId, null); + }); + + it('settles an app message on the desktop wire the same way', async () => { + const tab = await connect({ acks: ['notif.message'] }); + tab.on('notif.message', (_m: unknown, ack: (r: unknown) => void) => + ack({ visibilityState: 'visible' }), + ); + + const uid = await notify('app.message'); + + expect(await settledFor(uid)).toMatchObject({ + outcome: 'received', + responders: [{ visibilityState: 'visible' }], + }); + }); + + it('settles a rewritten notification as it is delivered again', async () => { + const tab = await connect({ acks: ['notif.message'] }); + tab.on('notif.message', (_m: unknown, ack: (r: unknown) => void) => + ack({ visibilityState: 'visible' }), + ); + const uid = await notify(); + await settledFor(uid); + settled.length = 0; + + await server.services.notification.notifyUpdate( + uid, + userId, + { title: 'hello again' }, + { type: 'share.received' }, + ); + + expect(await settledFor(uid)).toMatchObject({ outcome: 'received' }); + }); + + it('sends nothing and settles nothing for a silent notification', async () => { + await server.services.notification.notify( + [userId], + { title: 'quiet' }, + { type: 'share.received', silent: true }, + ); + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(settled).toEqual([]); + }); +}); diff --git a/src/backend/services/notification/notificationSocket.ts b/src/backend/services/notification/notificationSocket.ts index cb04893c7..70ce47cb4 100644 --- a/src/backend/services/notification/notificationSocket.ts +++ b/src/backend/services/notification/notificationSocket.ts @@ -18,7 +18,11 @@ */ import type { EventClient } from '../../clients/event/EventClient.js'; -import type { SocketService } from '../socket/SocketService.js'; +import type { + NotifDeliveryOutcome, + NotifDeliveryResponder, +} from '../../clients/event/types.js'; +import type { AckedSend, SocketService } from '../socket/SocketService.js'; import type { IConfig } from '../../types.js'; /** @@ -41,12 +45,24 @@ import type { IConfig } from '../../types.js'; * advertised (`gui_params.eventsNotifications`) and falls back to this wire * when that lapses, so the adapter is what keeps older clients and the fallback * working. It goes when the flag is on everywhere. + * + * Either way, the emitting region sends `notif.message` itself, asking sockets + * that acknowledge it for a receipt, and reports the result as the node-local + * `notif.delivery.settled`. */ export type NotifWire = 'notif.message' | 'notif.unreads' | 'notif.ack'; export const NOTIF_DELIVERY_EVENT = 'outer.notif.delivery'; +/** How long a delivered `notif.message` waits for receipts by default. */ +export const DEFAULT_NOTIFICATION_ACK_TIMEOUT_MS = 5_000; + +interface MessageResponse { + uid: string; + notification: unknown; +} + /** One unread as the replay payload carries it. */ export interface UnreadNotification { uid: string; @@ -63,10 +79,40 @@ export const notificationsFoldInEnabled = (config: IConfig): boolean => config.events?.enabled === true && config.events?.notificationsFoldIn === true; +/** Configured receipt window for a delivered `notif.message`. */ +export const notificationAckTimeoutMs = (config: IConfig): number => { + const configured = Number(config.notificationAckTimeoutMs); + return Number.isFinite(configured) && configured > 0 + ? configured + : DEFAULT_NOTIFICATION_ACK_TIMEOUT_MS; +}; + +const deliveryOutcome = async ( + sent: AckedSend, + heldInPeerRegion: () => Promise, +): Promise => { + if (sent.sockets === 0) { + return (await heldInPeerRegion()) ? 'remote-only' : 'no-socket'; + } + return sent.responses.length > 0 ? 'received' : 'unreceived'; +}; + +const responderOf = (response: unknown): NotifDeliveryResponder => { + const state = (response as { visibilityState?: unknown } | null) + ?.visibilityState; + return { + visibilityState: + state === 'visible' || state === 'hidden' ? state : 'unknown', + }; +}; + interface AdapterDeps { event: EventClient; socket: SocketService; foldIn: () => boolean; + ackTimeoutMs: () => number; + /** Whether a peer region holds a socket for the user. */ + heldInPeerRegion: (userId: number) => Promise; } export class NotificationSocketAdapter { @@ -78,12 +124,16 @@ export class NotificationSocketAdapter { /** * Start delivering what the events layer dispatches. Both the region that - * emitted and the peers that received it land here exactly once, so the - * handler makes no distinction between them. + * emitted and the peers that received it land here exactly once; only the + * emitting region waits on receipts for a `notif.message`. */ attach(): void { - this.#deps.event.on(NOTIF_DELIVERY_EVENT, (_key, data) => { + this.#deps.event.on(NOTIF_DELIVERY_EVENT, (_key, data, meta) => { const { userId, wire, response } = data; + if (wire === 'notif.message' && !meta?.from_outside) { + this.#deliverMessage(userId, response as MessageResponse); + return; + } void this.#deps.socket .send({ room: userId }, wire, response) .catch((err: unknown) => { @@ -96,10 +146,11 @@ export class NotificationSocketAdapter { message(userId: number, uid: string, notification: unknown): void { const response = { uid, notification }; if (this.#dispatched(userId, 'notif.message', response)) return; + this.#deliverMessage(userId, response); this.#deps.event.emit( 'outer.gui.notif.message', { user_id_list: [userId], response }, - {}, + { region_delivered: true }, ); } @@ -130,6 +181,38 @@ export class NotificationSocketAdapter { return this.#deps.socket.has({ room: userId }); } + #deliverMessage(userId: number, response: MessageResponse): void { + void this.#sendWithReceipt(userId, response).catch((err: unknown) => { + console.warn('[notification] socket send failed', err); + }); + } + + async #sendWithReceipt( + userId: number, + response: MessageResponse, + ): Promise { + const sent = await this.#deps.socket.sendWithAck( + { room: userId }, + 'notif.message', + response, + { timeoutMs: this.#deps.ackTimeoutMs() }, + ); + const outcome = await deliveryOutcome(sent, () => + this.#deps.heldInPeerRegion(userId), + ); + this.#deps.event.emit( + 'notif.delivery.settled', + { + uid: response.uid, + userId, + outcome, + ackCapable: sent.acking > 0, + responders: sent.responses.map(responderOf), + }, + {}, + ); + } + /** Take the events-layer route, or report that the legacy one applies. */ #dispatched(userId: number, wire: NotifWire, response: unknown): boolean { if (!this.#deps.foldIn()) return false; diff --git a/src/backend/services/socket/SocketService.test.ts b/src/backend/services/socket/SocketService.test.ts index 041a8b4bd..ae9cabc86 100644 --- a/src/backend/services/socket/SocketService.test.ts +++ b/src/backend/services/socket/SocketService.test.ts @@ -45,6 +45,7 @@ import { accountSocketRoom, buildSocketReauthError, decideSocketAuth, + handshakeAcks, socketRoomsFor, SocketService, type SocketReauthError, @@ -285,6 +286,32 @@ describe('socketRoomsFor', () => { // -- Live socket.io integration -------------------------------------- +describe('handshakeAcks', () => { + it('keeps the event names a client lists', () => { + expect(handshakeAcks(['notif.message', 'other'])).toEqual([ + 'notif.message', + 'other', + ]); + }); + + it('drops anything that is not a short non-empty string', () => { + expect( + handshakeAcks(['ok', '', 7, null, { a: 1 }, 'x'.repeat(65)]), + ).toEqual(['ok']); + }); + + it('reads nothing from a value that is not a list', () => { + expect(handshakeAcks(undefined)).toEqual([]); + expect(handshakeAcks('notif.message')).toEqual([]); + expect(handshakeAcks({ 0: 'notif.message' })).toEqual([]); + }); + + it('takes at most 16 names', () => { + const many = Array.from({ length: 20 }, (_, i) => `e${i}`); + expect(handshakeAcks(many)).toEqual(many.slice(0, 16)); + }); +}); + describe('SocketService (live socket.io)', () => { let server: PuterServer; let socketService: SocketService; @@ -494,6 +521,131 @@ describe('SocketService (live socket.io)', () => { tabB.disconnect(); }); + describe('sendWithAck', () => { + const userIdOf = async () => + (await server.stores.user.getByUsername(user.username))!.id; + + const closeAll = async (userId: number, sockets: ClientSocket[]) => { + for (const socket of sockets) socket.disconnect(); + await vi.waitFor(() => + expect(socketService.has({ room: userId })).toBe(false), + ); + }; + + it('collects the acknowledgement of each socket that declared it', async () => { + const userId = await userIdOf(); + const acking = await connect({ + auth_token: user.token, + acks: ['demo.ack'], + }); + const plain = await connect({ auth_token: user.token }); + acking.on('demo.ack', (data: unknown, ack: (r: unknown) => void) => + ack({ got: data }), + ); + const plainGot = new Promise((resolve) => + plain.once('demo.ack', (...args: unknown[]) => resolve(args)), + ); + + const sent = await socketService.sendWithAck( + { room: userId }, + 'demo.ack', + { n: 1 }, + { timeoutMs: 2_000 }, + ); + + expect(sent).toEqual({ + responses: [{ got: { n: 1 } }], + timedOut: false, + sockets: 2, + acking: 1, + }); + // A socket that never declared it gets the event with no callback. + expect(await plainGot).toEqual([{ n: 1 }]); + await closeAll(userId, [acking, plain]); + }); + + it('times out with the acknowledgements that did arrive', async () => { + const userId = await userIdOf(); + const answers = await connect({ + auth_token: user.token, + acks: ['demo.ack'], + }); + const silent = await connect({ + auth_token: user.token, + acks: ['demo.ack'], + }); + answers.on('demo.ack', (_d: unknown, ack: (r: unknown) => void) => + ack('here'), + ); + const silentGot = new Promise((resolve) => + silent.once('demo.ack', resolve), + ); + + const started = Date.now(); + const sent = await socketService.sendWithAck( + { room: userId }, + 'demo.ack', + { n: 2 }, + { timeoutMs: 300 }, + ); + + expect(Date.now() - started).toBeGreaterThanOrEqual(250); + expect(sent).toEqual({ + responses: ['here'], + timedOut: true, + sockets: 2, + acking: 2, + }); + expect(await silentGot).toEqual({ n: 2 }); + await closeAll(userId, [answers, silent]); + }); + + it('returns at once when no socket acknowledges the event', async () => { + const userId = await userIdOf(); + const other = await connect({ + auth_token: user.token, + acks: ['some.other.event'], + }); + const got = new Promise((resolve) => + other.once('demo.ack', resolve), + ); + + const started = Date.now(); + const sent = await socketService.sendWithAck( + { room: userId }, + 'demo.ack', + { n: 3 }, + { timeoutMs: 10_000 }, + ); + + expect(Date.now() - started).toBeLessThan(2_000); + expect(sent).toEqual({ + responses: [], + timedOut: false, + sockets: 1, + acking: 0, + }); + expect(await got).toEqual({ n: 3 }); + await closeAll(userId, [other]); + }); + + it('reports no sockets for a room nobody is in', async () => { + await expect( + socketService.sendWithAck( + { room: 'nobody-here' }, + 'demo.ack', + {}, + { timeoutMs: 10_000 }, + ), + ).resolves.toEqual({ + responses: [], + timedOut: false, + sockets: 0, + acking: 0, + }); + }); + }); + it('emits web.socket.user-connected when the GUI signals it is really open', async () => { const row = await server.stores.user.getByUsername(appUser.username); const userId = row!.id; @@ -895,6 +1047,29 @@ describe('SocketService — outer.gui fan-out', () => { spy.mockRestore(); }); + it('leaves this region to the emitter that already delivered to it', async () => { + const spy = captureSends(); + await server.clients.event.emitAndWait( + 'outer.gui.notif.message', + { user_id_list: [4344], response: { uid: 'n-2' } }, + { region_delivered: true }, + ); + expect(sends).toHaveLength(0); + + // A peer region still delivers the copy it was handed. + await server.clients.event.emitAndWait( + 'outer.gui.notif.message', + { user_id_list: [4344], response: { uid: 'n-2' } }, + { region_delivered: true, from_outside: true }, + ); + await vi.waitFor(() => expect(sends).toHaveLength(1)); + expect(sends[0]).toMatchObject({ + spec: { room: 4344 }, + key: 'notif.message', + }); + spy.mockRestore(); + }); + it('ignores an event with no recipients', async () => { const spy = captureSends(); await server.clients.event.emitAndWait( diff --git a/src/backend/services/socket/SocketService.ts b/src/backend/services/socket/SocketService.ts index bfe69db21..4cf398edb 100644 --- a/src/backend/services/socket/SocketService.ts +++ b/src/backend/services/socket/SocketService.ts @@ -208,6 +208,41 @@ interface OuterGuiPayload { response: unknown; } +/** What one {@link SocketService.sendWithAck} reached and heard back. */ +export interface AckedSend { + /** One per socket that acknowledged, as that socket sent it. */ + responses: unknown[]; + /** A socket that acknowledges this event did not answer in time. */ + timedOut: boolean; + /** Sockets the event went to, across this cluster's nodes. */ + sockets: number; + /** Of those, the ones that acknowledge this event. */ + acking: number; +} + +/** A listing that waits on another node longer than this counts this node only. */ +const SOCKET_LIST_TIMEOUT_MS = 1_000; +const MAX_HANDSHAKE_ACKS = 16; +const MAX_ACK_EVENT_LENGTH = 64; + +/** The events a client says it acknowledges, from `handshake.auth.acks`. */ +export const handshakeAcks = (raw: unknown): string[] => + Array.isArray(raw) + ? raw + .filter( + (key): key is string => + typeof key === 'string' && + key.length > 0 && + key.length <= MAX_ACK_EVENT_LENGTH, + ) + .slice(0, MAX_HANDSHAKE_ACKS) + : []; + +const acksEvent = (data: unknown, key: string): boolean => { + const acks = (data as { acks?: unknown } | undefined)?.acks; + return Array.isArray(acks) && acks.includes(key); +}; + interface UploadProgressPayload { upload_tracker: UploadProgressTrackerLike; meta?: Record; @@ -342,6 +377,118 @@ export class SocketService extends PuterService { } } + /** + * Send an event and wait up to `timeoutMs` for the sockets that declared + * they acknowledge it (`handshake.auth.acks`); the rest get a plain emit. + * Covers every node in the cluster, or this node alone when the adapter + * can't list the others. + */ + async sendWithAck( + specifiers: SocketSpecifier | SocketSpecifier[], + key: string, + data: unknown, + { timeoutMs }: { timeoutMs: number }, + ): Promise { + const none: AckedSend = { + responses: [], + timedOut: false, + sockets: 0, + acking: 0, + }; + const io = this.#io; + if (!io) return none; + const targets = ( + Array.isArray(specifiers) ? specifiers : [specifiers] + ).flatMap((spec) => + spec.room !== undefined + ? [String(spec.room)] + : spec.socket + ? [spec.socket] + : [], + ); + if (targets.length === 0) return none; + + const { sockets, clustered } = await this.#listSockets(targets); + const acking = sockets + .filter((socket) => acksEvent(socket.data, key)) + .map((socket) => socket.id); + io.to(targets).except(acking).emit(key, data); + if (acking.length === 0) return { ...none, sockets: sockets.length }; + + const heard = clustered + ? await this.#broadcastWithAck(acking, key, data, timeoutMs) + : await this.#emitWithAckHere(acking, key, data, timeoutMs); + return { ...heard, sockets: sockets.length, acking: acking.length }; + } + + /** `fetchSockets` needs the adapter's server count, which a mock lacks. */ + async #listSockets(targets: string[]): Promise<{ + sockets: { id: string; data: unknown }[]; + clustered: boolean; + }> { + const io = this.#io!; + try { + const sockets = await io + .in(targets) + .timeout(SOCKET_LIST_TIMEOUT_MS) + .fetchSockets(); + return { sockets, clustered: true }; + } catch { + const ids = new Set(); + for (const target of targets) { + io.sockets.adapter.rooms.get(target)?.forEach((id) => { + ids.add(id); + }); + } + const sockets = [...ids].flatMap((id) => { + const socket = io.sockets.sockets.get(id); + return socket ? [{ id, data: socket.data as unknown }] : []; + }); + return { sockets, clustered: false }; + } + } + + #broadcastWithAck( + ids: string[], + key: string, + data: unknown, + timeoutMs: number, + ): Promise> { + return new Promise((resolve) => { + this.#io!.to(ids) + .timeout(timeoutMs) + .emit(key, data, (err: Error | null, responses: unknown[]) => { + resolve({ responses: responses ?? [], timedOut: !!err }); + }); + }); + } + + async #emitWithAckHere( + ids: string[], + key: string, + data: unknown, + timeoutMs: number, + ): Promise> { + const heard = await Promise.all( + ids.map(async (id) => { + const socket = this.#io!.sockets.sockets.get(id); + if (!socket) return { ok: false as const }; + try { + const response: unknown = await socket + .timeout(timeoutMs) + .emitWithAck(key, data); + return { ok: true as const, response }; + } catch { + return { ok: false as const }; + } + }), + ); + return { + responses: heard.flatMap((h) => (h.ok ? [h.response] : [])), + timedOut: heard.some((h) => !h.ok), + }; + } + /** * Check whether the specifier currently resolves to at least one live * socket on _this_ node. Note: doesn't check other cluster nodes — intended @@ -419,8 +566,7 @@ export class SocketService extends PuterService { // `{ auth: { ... } }`, not the query string. puter-js uses // `io(url, { auth: { auth_token } })`. const handshakeAuth = socket.handshake.auth as - | Record - | undefined; + Record | undefined; const tokenRaw = typeof handshakeAuth?.auth_token === 'string' ? handshakeAuth.auth_token @@ -471,6 +617,7 @@ export class SocketService extends PuterService { socket.actor = decision.accept; socket.authToken = token; + socket.data.acks = handshakeAcks(handshakeAuth?.acks); // user.id is numeric in the DB; stringify for room name // so adapter lookups key on a stable type. socket.join(socketRoomsFor(decision.accept)); @@ -851,7 +998,9 @@ export class SocketService extends PuterService { // notification (item.added/updated/removed/moved/pending, // cache.updated, submission.done, …). EventClient walks the // dot-prefix tree at emit time so we get them all. - this.clients.event.on('outer.gui.*', (key: string, data: unknown) => { + this.clients.event.on('outer.gui.*', (key, data, meta) => { + // The emitter already reached this region's sockets itself. + if (meta?.region_delivered && !meta.from_outside) return; this.#handleOuterGui(key, data as OuterGuiPayload).catch( (err: unknown) => { console.error('[socket] outer.gui handler error', err); @@ -929,8 +1078,7 @@ export class SocketService extends PuterService { // their next poll of /cache/last-change-timestamp. const originalSocketId = ( data.response as - | { original_client_socket_id?: string } - | undefined + { original_client_socket_id?: string } | undefined )?.original_client_socket_id; await this.send({ room: userId }, 'cache.updated', { timestamp, @@ -944,9 +1092,7 @@ export class SocketService extends PuterService { #handleUploadProgress(data: UploadProgressPayload): void { const meta = data.meta ?? {}; const userId = (meta.user_id ?? meta.userId) as - | number - | string - | undefined; + number | string | undefined; if (!userId) { console.warn('[socket] upload-progress missing user_id', { meta }); return; diff --git a/src/backend/types.ts b/src/backend/types.ts index e3abcc96f..645ab1dee 100644 --- a/src/backend/types.ts +++ b/src/backend/types.ts @@ -971,6 +971,12 @@ interface IConfigOptional { */ notificationRetentionDays?: number; + /** + * How long a live notification waits for the recipient's sockets to + * acknowledge it before its delivery settles as unreceived. Default 5000. + */ + notificationAckTimeoutMs?: number; + /** * How many `app.message` notifications one app may create for one user. * Omit a field for the built-in default (30 per minute, 300 per day); a diff --git a/src/gui/src/UI/Dashboard/UIDashboard.js b/src/gui/src/UI/Dashboard/UIDashboard.js index b2bbb85c1..2c8daec24 100644 --- a/src/gui/src/UI/Dashboard/UIDashboard.js +++ b/src/gui/src/UI/Dashboard/UIDashboard.js @@ -28,6 +28,7 @@ import UIWindowFeedback from '../UIWindowFeedback.js'; import apply_item_added_to_containers from '../../helpers/applyItemAddedToContainers.js'; import { clear_shared_param } from '../../helpers/parseSharedPath.js'; import UIDashboardNotifications from './UIDashboardNotifications.js'; +import { NOTIFICATION_SOCKET_ACKS } from '../../helpers/notificationFeed.js'; /** * Creates and displays the Dashboard window. * @@ -276,6 +277,7 @@ async function UIDashboard (options) { window.socket = io(`${window.gui_origin}/`, { auth: { auth_token: window.auth_token, + acks: [...NOTIFICATION_SOCKET_ACKS], }, transports: ['websocket', 'polling'], withCredentials: true, diff --git a/src/gui/src/UI/Dashboard/UIDashboardNotifications.ack.test.js b/src/gui/src/UI/Dashboard/UIDashboardNotifications.ack.test.js new file mode 100644 index 000000000..d4b06a008 --- /dev/null +++ b/src/gui/src/UI/Dashboard/UIDashboardNotifications.ack.test.js @@ -0,0 +1,83 @@ +// @vitest-environment jsdom +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +import { describe, expect, it, vi } from 'vitest'; +import jQuery from '../../lib/jquery-3.6.1/jquery-3.6.1.min.js'; + +globalThis.$ = jQuery; +globalThis.jQuery = jQuery; +globalThis.html_encode = (value) => String(value ?? ''); +window.html_encode = globalThis.html_encode; +globalThis.i18n = (key) => key; +window.matchMedia = () => ({ matches: false, addEventListener () {}, removeEventListener () {} }); +window.api_origin = 'https://api.puter.test'; +window.auth_token = 'tok-1'; +window.user = { uuid: 'u-1' }; +globalThis.fetch = vi.fn(async () => new Response('{}', { status: 404 })); + +vi.mock('../UINotification.js', () => ({ default: vi.fn(() => ({})) })); +vi.mock('../UIWindow.js', () => ({ reveal_dashboard: vi.fn() })); +vi.mock('../../helpers/getAppByUid.js', () => ({ get_app_by_uid: vi.fn(async () => null) })); +vi.mock('../../helpers/launchApp.js', () => ({ default: vi.fn() })); +vi.mock('../../helpers/notificationApi.js', async (importOriginal) => ({ + ...(await importOriginal()), + getNotification: vi.fn(async () => null), + listNotifications: vi.fn(async () => ({ items: [] })), + markNotificationAcknowledged: vi.fn(async () => {}), +})); + +const { default: UIDashboardNotifications } = await import('./UIDashboardNotifications.js'); + +/** Records listeners the way socket.io-client registers them. */ +const fakeSocket = () => { + const handlers = {}; + return { + handlers, + on (event, fn) { + (handlers[event] ??= []).push(fn); + }, + emit () {}, + }; +}; + +describe('UIDashboardNotifications notif.message', () => { + it('acknowledges the delivery with the page visibility', () => { + document.body.innerHTML = '

'; + Object.defineProperty(document, 'visibilityState', { value: 'hidden', configurable: true }); + const socket = fakeSocket(); + UIDashboardNotifications({ $el_window: $('.win'), socket }); + + const ack = vi.fn(); + for ( const handler of socket.handlers['notif.message'] ) { + handler({ uid: 'n-1', notification: { title: 'hi' } }, ack); + } + + expect(ack).toHaveBeenCalledTimes(1); + expect(ack).toHaveBeenCalledWith({ visibilityState: 'hidden' }); + }); + + it('takes a message sent without a callback', () => { + document.body.innerHTML = '
'; + const socket = fakeSocket(); + UIDashboardNotifications({ $el_window: $('.win'), socket }); + + expect(() => socket.handlers['notif.message'][0]({ uid: 'n-2', notification: { title: 'hi' } })) + .not.toThrow(); + }); +}); diff --git a/src/gui/src/UI/Dashboard/UIDashboardNotifications.js b/src/gui/src/UI/Dashboard/UIDashboardNotifications.js index 561812630..d50de98bd 100644 --- a/src/gui/src/UI/Dashboard/UIDashboardNotifications.js +++ b/src/gui/src/UI/Dashboard/UIDashboardNotifications.js @@ -20,7 +20,11 @@ import UINotification from '../UINotification.js'; import { reveal_dashboard } from '../UIWindow.js'; import { getNotification, listNotifications, markNotificationAcknowledged } from '../../helpers/notificationApi.js'; -import { applyToastMark, createNotificationFeed } from '../../helpers/notificationFeed.js'; +import { + acknowledgeDelivery, + applyToastMark, + createNotificationFeed, +} from '../../helpers/notificationFeed.js'; import { formatAbsoluteTime, formatRelativeTime } from '../../i18n/formatTime.js'; import { handleNotificationTap, @@ -769,7 +773,8 @@ export default function UIDashboardNotifications ({ $el_window, socket }) { }); void feed.start(); - socket.on('notif.message', ({ uid, notification }) => { + socket.on('notif.message', ({ uid, notification }, ack) => { + acknowledgeDelivery(ack); if ( feed.isActive() ) return; receive([{ uid, notification }]); }); diff --git a/src/gui/src/UI/UIDesktop.js b/src/gui/src/UI/UIDesktop.js index 5da25d6d0..612029693 100644 --- a/src/gui/src/UI/UIDesktop.js +++ b/src/gui/src/UI/UIDesktop.js @@ -44,7 +44,12 @@ import { urlFileLaunchOptions } from '../helpers/confirmUrlFileAccess.js'; import item_icon from '../helpers/itemIcon.js'; import { SHARED_PATH_PARAM, clear_shared_param } from '../helpers/parseSharedPath.js'; import resolve_shared_item from '../helpers/resolveSharedItem.js'; -import { applyToastMark, createNotificationFeed } from '../helpers/notificationFeed.js'; +import { + NOTIFICATION_SOCKET_ACKS, + acknowledgeDelivery, + applyToastMark, + createNotificationFeed, +} from '../helpers/notificationFeed.js'; import { getNotification, markNotificationAcknowledged } from '../helpers/notificationApi.js'; import { elementsWithData, @@ -140,6 +145,7 @@ async function UIDesktop (options) { window.socket = io(`${window.gui_origin }/`, { auth: { auth_token: window.auth_token, + acks: [...NOTIFICATION_SOCKET_ACKS], }, transports: ['websocket', 'polling'], withCredentials: true, @@ -324,7 +330,8 @@ async function UIDesktop (options) { * This event is triggered if a user receives a notification during * an active session. */ - window.socket.on('notif.message', async ({ uid, notification }) => { + window.socket.on('notif.message', async ({ uid, notification }, ack) => { + acknowledgeDelivery(ack); if ( notification_feed.isActive() ) return; show_notification({ uid, notification }); }); diff --git a/src/gui/src/helpers/notificationFeed.js b/src/gui/src/helpers/notificationFeed.js index e460e7f25..aaaa986f2 100644 --- a/src/gui/src/helpers/notificationFeed.js +++ b/src/gui/src/helpers/notificationFeed.js @@ -35,6 +35,24 @@ import { markNotificationShown, } from './notificationApi.js'; +/** Socket events this page acknowledges, listed in the handshake's `acks`. */ +export const NOTIFICATION_SOCKET_ACKS = Object.freeze(['notif.message']); + +const acknowledged = new WeakSet(); + +/** + * Answer a live `notif.message`'s acknowledgement callback with whether the + * page is in view. Every listener gets the same callback, so this answers once. + * + * @param {unknown} ack - The callback socket.io passes, when the server asked for one + * @param {{ visibilityState?: string }} [doc] + */ +export const acknowledgeDelivery = (ack, doc = globalThis.document) => { + if ( typeof ack !== 'function' || acknowledged.has(ack) ) return; + acknowledged.add(ack); + ack({ visibilityState: doc?.visibilityState ?? 'unknown' }); +}; + /** A session's own mailbox slices; each covers its audience's rows for every app. */ export const NOTIFICATION_SUBJECTS = Object.freeze([ 'notif:account', diff --git a/src/gui/src/helpers/notificationFeed.test.js b/src/gui/src/helpers/notificationFeed.test.js index cc72a1a3e..ec703679b 100644 --- a/src/gui/src/helpers/notificationFeed.test.js +++ b/src/gui/src/helpers/notificationFeed.test.js @@ -19,7 +19,9 @@ import { afterEach, describe, it, expect, vi } from 'vitest'; import { + NOTIFICATION_SOCKET_ACKS, NOTIFICATION_SUBJECTS, + acknowledgeDelivery, applyToastMark, createNotificationFeed, eventsNotificationsAvailable, @@ -374,3 +376,32 @@ describe('createNotificationFeed', () => { for ( const sub of sdk.subs ) expect(sub.off).toHaveBeenCalled(); }); }); + +describe('acknowledgeDelivery', () => { + it('answers with whether the page is in view', () => { + const ack = vi.fn(); + acknowledgeDelivery(ack, { visibilityState: 'hidden' }); + expect(ack).toHaveBeenCalledWith({ visibilityState: 'hidden' }); + }); + + it('answers once when two listeners get the same message', () => { + const ack = vi.fn(); + acknowledgeDelivery(ack, { visibilityState: 'visible' }); + acknowledgeDelivery(ack, { visibilityState: 'visible' }); + expect(ack).toHaveBeenCalledTimes(1); + }); + + it('does nothing for a message sent without a callback', () => { + expect(() => acknowledgeDelivery(undefined)).not.toThrow(); + }); + + it('says unknown where there is no document to ask', () => { + const ack = vi.fn(); + acknowledgeDelivery(ack, undefined); + expect(ack).toHaveBeenCalledWith({ visibilityState: 'unknown' }); + }); + + it('lists notif.message as acknowledged in the handshake', () => { + expect(NOTIFICATION_SOCKET_ACKS).toEqual(['notif.message']); + }); +});