From bf91b40ba6d668889674400fdc613b853f15a29f Mon Sep 17 00:00:00 2001 From: Juan Castro Date: Thu, 29 Oct 2026 13:37:22 -0400 Subject: [PATCH] feat(notifications): report how each live notif.message was received SocketService.sendWithAck sends an event to rooms or sockets and waits, up to a timeout, for an acknowledgement from each socket that listed the event in its handshake (`auth.acks`); every other socket gets a plain emit. It lists sockets across the cluster with fetchSockets and asks the listed ones through the adapter's broadcast acks, and counts this node alone when the adapter cannot list the others. The emitting region now sends notif.message through it, with the window from `notificationAckTimeoutMs` (default 5000), and emits the node-local `notif.delivery.settled` with the outcome: received, unreceived, no-socket or remote-only (sockets only in a peer region), whether any socket there acknowledges, and each responder's visibilityState. Peer regions keep sending their own copy as before; the GUI event carries `region_delivered` so this region does not send it twice. The desktop and dashboard list notif.message in their handshake and answer its callback once with document.visibilityState. Nothing in core acts on the settled event. --- config.template.jsonc | 4 + src/backend/clients/event/types.ts | 26 ++ .../events/notifications.integration.test.ts | 59 +++++ src/backend/services/events/registry.ts | 4 + .../notification/NotificationService.ts | 14 + .../notification/notificationDelivery.test.ts | 245 ++++++++++++++++++ .../notification/notificationSocket.ts | 93 ++++++- .../services/socket/SocketService.test.ts | 175 +++++++++++++ src/backend/services/socket/SocketService.ts | 162 +++++++++++- src/backend/types.ts | 6 + src/gui/src/UI/Dashboard/UIDashboard.js | 2 + .../UIDashboardNotifications.ack.test.js | 83 ++++++ .../UI/Dashboard/UIDashboardNotifications.js | 9 +- src/gui/src/UI/UIDesktop.js | 11 +- src/gui/src/helpers/notificationFeed.js | 18 ++ src/gui/src/helpers/notificationFeed.test.js | 31 +++ 16 files changed, 925 insertions(+), 17 deletions(-) create mode 100644 src/backend/services/notification/notificationDelivery.test.ts create mode 100644 src/gui/src/UI/Dashboard/UIDashboardNotifications.ack.test.js 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']); + }); +});