From 4fd9fae926879392adf50dc0aa79e2a241a95dcf Mon Sep 17 00:00:00 2001 From: Juan Castro Date: Fri, 23 Oct 2026 15:00:55 -0400 Subject: [PATCH] feat(push): send pushes for new notifications through a registered provider Every non-silent notification write and in-place update emits a node-local `notif.written` event. PushService picks it up when push is enabled and a provider is registered, and sends it to the user's devices unless the user is connected where the notification already shows. - `app.message` needs the user's push consent for that app and reaches Puter's devices plus that app's; every other type follows the user's Puter preference and reaches Puter's devices only. - Presence counts this node's sockets, and other nodes and regions where presence is kept. - Providers register per platform with `registerProvider`; core ships none. Batches of 100 tokens, bounded concurrency, no retries. Delivered devices are touched, unregistered ones removed, others get a failure counted. - Messages are built per type and fit to a 3,800 byte budget; `data` is never cut. --- src/backend/clients/event/types.ts | 20 +- src/backend/services/events/registry.ts | 4 + src/backend/services/index.ts | 2 +- .../notification/NotificationService.ts | 25 +- .../push/PushService.delivery.test.ts | 524 ++++++++++++++++++ .../push/PushService.presence.test.ts | 230 ++++++++ src/backend/services/push/PushService.ts | 233 +++++++- src/backend/services/push/pushMessage.test.ts | 242 ++++++++ src/backend/services/push/pushMessage.ts | 202 +++++++ src/backend/services/push/pushPresence.ts | 49 ++ src/backend/services/push/types.ts | 65 +++ src/docs/src/rate-limits-and-quotas.md | 11 + 12 files changed, 1596 insertions(+), 11 deletions(-) create mode 100644 src/backend/services/push/PushService.delivery.test.ts create mode 100644 src/backend/services/push/PushService.presence.test.ts create mode 100644 src/backend/services/push/pushMessage.test.ts create mode 100644 src/backend/services/push/pushMessage.ts create mode 100644 src/backend/services/push/pushPresence.ts create mode 100644 src/backend/services/push/types.ts diff --git a/src/backend/clients/event/types.ts b/src/backend/clients/event/types.ts index 294c9e274..fa9d20585 100644 --- a/src/backend/clients/event/types.ts +++ b/src/backend/clients/event/types.ts @@ -516,6 +516,17 @@ export type EventMap = { /** Events handler runs behind whatever created it. */ handlerDepth?: number; }; + /** Every non-silent write or update, only on the node that wrote it. */ + 'notif.written': { + userId: number; + uid: string; + type: string; + /** The row's app. Absent on an update, which keeps the row's own. */ + appUid?: string | null; + value: Record; + /** An earlier notification rewritten in place. */ + updated?: true; + }; /** * One notification, addressed at whatever sockets each region holds for the * recipient. `outer.*` reaches peer regions and never sibling nodes, so @@ -927,11 +938,10 @@ export type EventKey = keyof EventMap & string; // Generates a wildcard for every non-final dot-separated prefix of K. export type WildcardPrefixes = K extends `${infer Head}.${infer Tail}` - ? - | `${Head}.*` - | (Tail extends `${string}.${string}` - ? `${Head}.${WildcardPrefixes}` - : never) + ? | `${Head}.*` + | (Tail extends `${string}.${string}` + ? `${Head}.${WildcardPrefixes}` + : never) : never; export type ListenKey = EventKey | WildcardPrefixes; diff --git a/src/backend/services/events/registry.ts b/src/backend/services/events/registry.ts index cd2553e5f..69a56b1b1 100644 --- a/src/backend/services/events/registry.ts +++ b/src/backend/services/events/registry.ts @@ -429,6 +429,10 @@ export const UNPUBLISHED_INTERNAL_EVENTS = [ event: 'kv.flushed', reason: 'No subject addresses "the whole namespace went": the grammar anchors on a key or a prefix, and fanning a flush out per key would report keys a truncated enumeration never saw.', }, + { + event: 'notif.written', + reason: 'The node-local trigger for pushes to devices; 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/index.ts b/src/backend/services/index.ts index d74ba77a5..5c1e5b25c 100644 --- a/src/backend/services/index.ts +++ b/src/backend/services/index.ts @@ -140,7 +140,7 @@ export const puterServices = { // both; `fs` reaches back for dispatch at call time only. events: EventsService, notification: NotificationService, - // Device lifecycle only: listens for session revokes and sweeps stale rows. + // Reaches `socket`, `eventForward` and `permission` at call time only. push: PushService, // Declared after `auth` (origin → app uid resolution happens through // AuthService.appUidFromOrigin). diff --git a/src/backend/services/notification/NotificationService.ts b/src/backend/services/notification/NotificationService.ts index 494c669b5..d81cc1609 100644 --- a/src/backend/services/notification/NotificationService.ts +++ b/src/backend/services/notification/NotificationService.ts @@ -167,12 +167,24 @@ export class NotificationService extends PuterService { return null; } - if (!opts.silent) + if (!opts.silent) { + this.clients.event.emit( + 'notif.written', + { + userId, + uid, + type: registered.type, + appUid, + value: payload, + }, + {}, + ); await this.#push(userId, uid, payload, { type: registered.type, audience: registered.audience, appUid, }); + } return uid; }), ); @@ -215,6 +227,17 @@ export class NotificationService extends PuterService { // to anything deduplicating on `event.id`. this.#socket().message(userId, uid, payload); this.#markDelivered(uid, userId); + this.clients.event.emit( + 'notif.written', + { + userId, + uid, + type: registered.type, + value: payload, + updated: true, + }, + {}, + ); } return true; } diff --git a/src/backend/services/push/PushService.delivery.test.ts b/src/backend/services/push/PushService.delivery.test.ts new file mode 100644 index 000000000..5c17920a2 --- /dev/null +++ b/src/backend/services/push/PushService.delivery.test.ts @@ -0,0 +1,524 @@ +/* + * 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 { v4 as uuidv4 } from 'uuid'; +import { + afterAll, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from 'vitest'; +import { makeActor } from '../../core/actor.js'; +import type { PuterServer } from '../../server.js'; +import type { PushPlatform } from '../../stores/push/PushDeviceStore.js'; +import { setupTestServer } from '../../testUtil.js'; +import type { IConfig } from '../../types.js'; +import type { PermissionService } from '../permission/PermissionService.js'; +import { NOTIFICATIONS_PUSH_PERMISSION } from './pushConsent.js'; +import { PUTER_PUSH_PREFERENCE_KEY } from './PushService.js'; +import type { PushBatch, PushProvider, PushResult } from './types.js'; + +interface TestUser { + id: number; + uuid: string; + username: string; +} + +/** Records every call; a token's text decides its scripted result. */ +class FakeProvider implements PushProvider { + readonly calls: PushBatch[] = []; + constructor(readonly platforms: PushPlatform[] = ['web']) {} + + async send(batch: PushBatch): Promise { + this.calls.push(batch); + return batch.tokens.map((token) => { + if (token.includes('gone')) return { status: 'unregistered' }; + if (token.includes('bad')) { + return { status: 'failed', reason: `rejected ${token}` }; + } + return { status: 'delivered' }; + }); + } +} + +/** Long enough for a delivery that was going to start to have finished. */ +const settle = () => new Promise((resolve) => setTimeout(resolve, 300)); + +const bootWith = (config: Record, provider?: FakeProvider) => { + const state = { server: null as unknown as PuterServer }; + beforeAll(async () => { + state.server = await setupTestServer(config as IConfig); + if (provider) state.server.services.push.registerProvider(provider); + }); + afterAll(async () => { + await state.server?.shutdown(); + }); + return state; +}; + +const makeUser = async (server: PuterServer): Promise => + (await server.stores.user.create({ + username: `pd${Math.random().toString(36).slice(2, 10)}`, + uuid: uuidv4(), + password: null, + email: null, + requires_email_confirmation: false, + })) as TestUser; + +const makeApp = async ( + server: PuterServer, + owner: TestUser, +): Promise<{ uid: string; name: string }> => { + const create = server.stores.app.create.bind( + server.stores.app, + ) as unknown as ( + fields: Record, + opts: { ownerUserId: number }, + ) => Promise<{ uid: string; name: string }>; + return create( + { + name: `push-${uuidv4()}`, + title: 'Push app', + index_url: `https://push-${uuidv4()}.test/`, + }, + { ownerUserId: owner.id }, + ); +}; + +const registerDevice = ( + server: PuterServer, + userId: number, + token: string, + opts: { appUid?: string; platform?: PushPlatform } = {}, +) => + server.stores.pushDevice.register({ + userId, + sessionUid: `sess-${uuidv4()}`, + platform: opts.platform ?? 'web', + token, + appUid: opts.appUid ?? null, + }); + +const grantPush = (server: PuterServer, user: TestUser, appUid: string) => + ( + server.services.permission as unknown as PermissionService + ).grantUserAppPermission( + makeActor({ user: user as never }), + appUid, + NOTIFICATIONS_PUSH_PERMISSION, + ); + +const shareReceived = (server: PuterServer, userId: number) => + server.services.notification.notify( + [userId], + { + title: 'alice shared an item with you', + fields: { + username: 'alice', + count: 1, + senders: [{ username: 'alice', count: 1 }], + target: { + path: '/alice/Shared/report.txt', + name: 'report.txt', + }, + }, + }, + { type: 'share.received' }, + ); + +const tokensOf = (calls: PushBatch[]) => + calls.flatMap((call) => call.tokens).sort(); + +describe('PushService delivery', () => { + const provider = new FakeProvider(); + const state = bootWith({ push: { enabled: true } }, provider); + const server = () => state.server; + + beforeEach(() => { + provider.calls.length = 0; + }); + + it('sends a share to the user’s Puter devices, naming the item', async () => { + const user = await makeUser(server()); + const token = `tok-${uuidv4()}`; + await registerDevice(server(), user.id, token); + + const uid = await shareReceived(server(), user.id); + + await vi.waitFor(() => expect(provider.calls).toHaveLength(1)); + const [call] = provider.calls; + expect(call.platform).toBe('web'); + expect(call.tokens).toEqual([token]); + expect(call.message).toMatchObject({ + uid, + title: 'alice shared report.txt', + group: uid, + target: { path: '/alice/Shared/report.txt' }, + }); + }); + + it('uses the stored title for a share of several items', async () => { + const user = await makeUser(server()); + await registerDevice(server(), user.id, `tok-${uuidv4()}`); + + await server().services.notification.notify( + [user.id], + { + title: 'alice shared 2 items with you', + fields: { senders: [{ username: 'alice', count: 2 }] }, + }, + { type: 'share.received' }, + ); + + await vi.waitFor(() => expect(provider.calls).toHaveLength(1)); + expect(provider.calls[0].message.title).toBe( + 'alice shared 2 items with you', + ); + }); + + it('sends nothing to a user with no devices', async () => { + const user = await makeUser(server()); + const listed = vi.spyOn(server().stores.pushDevice, 'listForDelivery'); + await shareReceived(server(), user.id); + await vi.waitFor(() => expect(listed).toHaveBeenCalled()); + await settle(); + listed.mockRestore(); + expect(provider.calls).toHaveLength(0); + }); + + it('sends an app message once, to Puter’s and that app’s devices only', async () => { + const user = await makeUser(server()); + const other = await makeUser(server()); + const app = await makeApp(server(), user); + const otherApp = await makeApp(server(), user); + await grantPush(server(), user, app.uid); + + await registerDevice(server(), user.id, 'tok-puter'); + await registerDevice(server(), user.id, 'tok-app', { appUid: app.uid }); + await registerDevice(server(), user.id, 'tok-other-app', { + appUid: otherApp.uid, + }); + await registerDevice(server(), other.id, 'tok-other-user'); + + const uid = await server().services.notification.notify( + [user.id], + { title: 'New reply', text: 'see you at 5', group: 'room-7' }, + { type: 'app.message', appUid: app.uid }, + ); + + await vi.waitFor(() => expect(provider.calls).toHaveLength(1)); + await settle(); + expect(provider.calls).toHaveLength(1); + expect(tokensOf(provider.calls)).toEqual(['tok-app', 'tok-puter']); + expect(provider.calls[0].message).toMatchObject({ + uid, + title: 'New reply', + body: 'see you at 5', + group: 'room-7', + target: { app: app.name }, + }); + }); + + it('sends a sensitive app message as a generic one under the app’s name', async () => { + const user = await makeUser(server()); + const app = await makeApp(server(), user); + await grantPush(server(), user, app.uid); + await registerDevice(server(), user.id, `tok-${uuidv4()}`); + + await server().services.notification.notify( + [user.id], + { + title: 'Code qx-marker', + text: 'your sign-in code qx-marker', + data: { code: 'qx-marker' }, + sensitive: true, + ttl: 90, + app: { uid: app.uid, name: 'stamped-name' }, + }, + { type: 'app.message', appUid: app.uid }, + ); + + await vi.waitFor(() => expect(provider.calls).toHaveLength(1)); + const message = provider.calls[0].message; + expect(message).toMatchObject({ + title: 'stamped-name', + body: 'New notification from stamped-name', + target: { app: 'stamped-name' }, + ttl: 90, + sensitive: true, + }); + expect(JSON.stringify(message)).not.toContain('qx-marker'); + }); + + it('sends no app message without the user’s consent, or after it is revoked', async () => { + const user = await makeUser(server()); + const app = await makeApp(server(), user); + await registerDevice(server(), user.id, `tok-${uuidv4()}`); + const send = () => + server().services.notification.notify( + [user.id], + { title: 'hello' }, + { type: 'app.message', appUid: app.uid }, + ); + + await send(); + await settle(); + expect(provider.calls).toHaveLength(0); + + await grantPush(server(), user, app.uid); + await send(); + await vi.waitFor(() => expect(provider.calls).toHaveLength(1)); + + await ( + server().services.permission as unknown as PermissionService + ).revokeUserAppPermission( + makeActor({ user: user as never }), + app.uid, + NOTIFICATIONS_PUSH_PERMISSION, + ); + await send(); + await settle(); + expect(provider.calls).toHaveLength(1); + }); + + it('sends Puter’s own notification about an app to Puter’s devices, with no grant', async () => { + const user = await makeUser(server()); + const app = await makeApp(server(), user); + await registerDevice(server(), user.id, 'tok-puter-dev'); + await registerDevice(server(), user.id, 'tok-app-dev', { + appUid: app.uid, + }); + + await server().services.notification.notify( + [user.id], + { title: 'Successfully deployed https://w.puter.work' }, + { type: 'app.worker.deployed', appUid: app.uid }, + ); + + await vi.waitFor(() => expect(provider.calls).toHaveLength(1)); + expect(tokensOf(provider.calls)).toEqual(['tok-puter-dev']); + expect(provider.calls[0].message.title).toBe( + 'Successfully deployed https://w.puter.work', + ); + }); + + it('follows the user’s Puter preference for Puter’s own notifications', async () => { + const user = await makeUser(server()); + await registerDevice(server(), user.id, `tok-${uuidv4()}`); + await server().stores.user.updateMetadata(user.id, { + [PUTER_PUSH_PREFERENCE_KEY]: false, + }); + + await shareReceived(server(), user.id); + await settle(); + expect(provider.calls).toHaveLength(0); + }); + + it('never pushes a silent notification', async () => { + const user = await makeUser(server()); + await registerDevice(server(), user.id, `tok-${uuidv4()}`); + + const uid = await server().services.notification.notify( + [user.id], + { title: 'quiet' }, + { type: 'share.received', silent: true }, + ); + await server().services.notification.notifyUpdate( + uid!, + user.id, + { title: 'still quiet' }, + { type: 'share.received', silent: true }, + ); + await settle(); + expect(provider.calls).toHaveLength(0); + }); + + it('pushes an update again under the same group', async () => { + const user = await makeUser(server()); + await registerDevice(server(), user.id, `tok-${uuidv4()}`); + + const uid = await shareReceived(server(), user.id); + await vi.waitFor(() => expect(provider.calls).toHaveLength(1)); + + const updated = await server().services.notification.notifyUpdate( + uid!, + user.id, + { + title: 'alice and bob shared 2 items with you', + fields: { + senders: [ + { username: 'alice', count: 1 }, + { username: 'bob', count: 1 }, + ], + }, + }, + { type: 'share.received' }, + ); + expect(updated).toBe(true); + + await vi.waitFor(() => expect(provider.calls).toHaveLength(2)); + expect(provider.calls[1].message).toMatchObject({ + uid, + group: provider.calls[0].message.group, + title: 'alice and bob shared 2 items with you', + }); + }); + + it('removes unregistered devices, counts failures once, and bumps the rest', async () => { + const user = await makeUser(server()); + const gone = await registerDevice( + server(), + user.id, + `gone-${uuidv4()}`, + ); + const bad = await registerDevice(server(), user.id, `bad-${uuidv4()}`); + const good = await registerDevice(server(), user.id, `ok-${uuidv4()}`); + await server().clients.db.write( + 'UPDATE `push_device` SET `last_seen_at` = ? WHERE `user_id` = ?', + [1000, user.id], + ); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + + await shareReceived(server(), user.id); + + const devices = server().stores.pushDevice; + await vi.waitFor(async () => { + expect(await devices.getByUid(gone.uid)).toBeNull(); + expect( + (await devices.getByUid(good.uid))!.lastSeenAt, + ).toBeGreaterThan(1000); + }); + await settle(); + const failed = await devices.getByUid(bad.uid); + expect(failed!.failureCount).toBe(1); + expect(failed!.lastSeenAt).toBe(1000); + expect(provider.calls).toHaveLength(1); + + const logged = JSON.stringify(warn.mock.calls); + warn.mockRestore(); + expect(logged).toContain(''); + expect(logged).not.toMatch(/bad-[0-9a-f]{8}-/); + }); + + it('sends 250 devices on one platform in three calls', async () => { + const user = await makeUser(server()); + for (let i = 0; i < 250; i++) { + await registerDevice(server(), user.id, `bulk-${i}-${uuidv4()}`); + } + + await shareReceived(server(), user.id); + + await vi.waitFor(() => expect(provider.calls).toHaveLength(3)); + await settle(); + expect(provider.calls).toHaveLength(3); + expect(provider.calls.map((call) => call.tokens.length).sort()).toEqual( + [100, 100, 50].sort(), + ); + expect(new Set(tokensOf(provider.calls)).size).toBe(250); + }); + + it('trims a long title to the size budget', async () => { + const user = await makeUser(server()); + await registerDevice(server(), user.id, `tok-${uuidv4()}`); + + await server().services.notification.notify( + [user.id], + { title: `Failed to deploy w! ${'x'.repeat(6000)}` }, + { type: 'app.worker.deployFailed' }, + ); + + await vi.waitFor(() => expect(provider.calls).toHaveLength(1)); + const message = provider.calls[0].message; + expect( + Buffer.byteLength(JSON.stringify(message), 'utf8'), + ).toBeLessThanOrEqual(3800); + expect(message.title.startsWith('Failed to deploy w!')).toBe(true); + }); + + it('skips devices on a platform no provider carries', async () => { + const user = await makeUser(server()); + await registerDevice(server(), user.id, 'tok-ios', { platform: 'ios' }); + await shareReceived(server(), user.id); + await settle(); + expect(provider.calls).toHaveLength(0); + }); +}); + +describe('PushService with push off', () => { + const provider = new FakeProvider(); + const state = bootWith({}, provider); + + it('sends nothing', async () => { + const user = await makeUser(state.server); + await registerDevice(state.server, user.id, `tok-${uuidv4()}`); + const listed = vi.spyOn( + state.server.stores.pushDevice, + 'listForDelivery', + ); + await shareReceived(state.server, user.id); + await settle(); + expect(listed).not.toHaveBeenCalled(); + expect(provider.calls).toHaveLength(0); + }); +}); + +describe('PushService with no provider', () => { + const state = bootWith({ push: { enabled: true } }); + + it('reads nothing and sends nothing', async () => { + const user = await makeUser(state.server); + await registerDevice(state.server, user.id, `tok-${uuidv4()}`); + const listed = vi.spyOn( + state.server.stores.pushDevice, + 'listForDelivery', + ); + await shareReceived(state.server, user.id); + await settle(); + expect(listed).not.toHaveBeenCalled(); + }); +}); + +describe('PushService with the notification fold-in on', () => { + const provider = new FakeProvider(); + const state = bootWith( + { + push: { enabled: true }, + events: { enabled: true, notificationsFoldIn: true }, + }, + provider, + ); + + it('sends one push per write though both notification events fire', async () => { + const user = await makeUser(state.server); + await registerDevice(state.server, user.id, `tok-${uuidv4()}`); + const created = vi.fn(); + state.server.clients.event.on('notif.created', created); + + await shareReceived(state.server, user.id); + + await vi.waitFor(() => expect(provider.calls).toHaveLength(1)); + await settle(); + expect(created).toHaveBeenCalledTimes(1); + expect(provider.calls).toHaveLength(1); + state.server.clients.event.off?.('notif.created', created); + }); +}); diff --git a/src/backend/services/push/PushService.presence.test.ts b/src/backend/services/push/PushService.presence.test.ts new file mode 100644 index 000000000..229a303bc --- /dev/null +++ b/src/backend/services/push/PushService.presence.test.ts @@ -0,0 +1,230 @@ +/* + * 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 { v4 as uuidv4 } from 'uuid'; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from 'vitest'; +import { makeActor } from '../../core/actor.js'; +import { PRESENCE_NO_APP } from '../../stores/events/PresenceStore.js'; +import { + createTestUser, + setupPuterTestEnv, + type PuterTestEnv, +} from '../../testUtil.js'; +import type { IConfig } from '../../types.js'; +import type { PermissionService } from '../permission/PermissionService.js'; +import { NOTIFICATIONS_PUSH_PERMISSION } from './pushConsent.js'; +import type { PushBatch, PushProvider, PushResult } from './types.js'; + +const { axiosRequestMock } = vi.hoisted(() => ({ + axiosRequestMock: vi.fn(), +})); + +vi.mock('axios', () => ({ + default: { request: axiosRequestMock }, + request: axiosRequestMock, +})); + +const BOOT_TIMEOUT_MS = 120_000; +const PEER_ID = 'peer-a'; + +class FakeProvider implements PushProvider { + readonly platforms = ['web'] as const; + readonly calls: PushBatch[] = []; + async send(batch: PushBatch): Promise { + this.calls.push(batch); + return batch.tokens.map(() => ({ status: 'delivered' })); + } +} + +const settle = () => new Promise((resolve) => setTimeout(resolve, 300)); + +let env: PuterTestEnv; +const provider = new FakeProvider(); +const openSockets: ClientSocket[] = []; + +interface Person { + id: number; + uuid: string; + token: string; + app: { uid: string; token: string }; +} + +const makePerson = async (): Promise => { + const username = `pp${Math.random().toString(36).slice(2, 10)}`; + const { token } = await createTestUser(env.server, { + username, + password: 'pw-test-1234', + }); + const user = (await env.server.stores.user.getByUsername(username))!; + const appUid = `app-${uuidv4()}`; + await env.server.clients.db.write( + 'INSERT INTO `apps` (`uid`, `name`, `title`, `index_url`, `owner_user_id`) VALUES (?, ?, ?, ?, ?)', + [appUid, appUid, appUid, `https://${appUid}.example/`, user.id], + ); + const appToken = await env.server.services.auth.getUserAppToken( + { user: user as never, effectiveApp: null }, + appUid, + ); + await ( + env.server.services.permission as unknown as PermissionService + ).grantUserAppPermission( + makeActor({ user }), + appUid, + NOTIFICATIONS_PUSH_PERMISSION, + ); + await env.server.stores.pushDevice.register({ + userId: user.id, + sessionUid: `sess-${uuidv4()}`, + platform: 'web', + token: `tok-${uuidv4()}`, + }); + return { + id: user.id, + uuid: user.uuid, + token, + app: { uid: appUid, token: appToken }, + }; +}; + +const connect = (token: string): Promise => + new Promise((resolve, reject) => { + const socket = ioClient(env.origin, { + auth: { auth_token: token }, + transports: ['websocket'], + reconnection: false, + }); + openSockets.push(socket); + socket.on('connect', () => resolve(socket)); + socket.on('connect_error', (err: Error) => reject(err)); + }); + +const puterNotification = (person: Person) => + env.server.services.notification.notify( + [person.id], + { title: 'A subscription ended' }, + { type: 'app.events.ended' }, + ); + +const appMessage = (person: Person) => + env.server.services.notification.notify( + [person.id], + { title: 'New reply' }, + { type: 'app.message', appUid: person.app.uid }, + ); + +/** How many pushes went out after `send`, once it has had time to. */ +const pushesAfter = async (send: () => Promise): Promise => { + const before = provider.calls.length; + await send(); + await settle(); + return provider.calls.length - before; +}; + +beforeAll(async () => { + env = await setupPuterTestEnv({ + events: { enabled: true }, + push: { enabled: true }, + broadcast: { + webhook: { peerId: 'self-node', secret: 'self-secret' }, + peers: [ + { + peerId: PEER_ID, + webhook: true, + webhook_url: 'http://push-peer.invalid/broadcast/webhook', + webhook_secret: 'peer-secret', + }, + ], + }, + } as unknown as IConfig); + env.server.services.push.registerProvider(provider); +}, BOOT_TIMEOUT_MS); + +beforeEach(() => { + axiosRequestMock.mockResolvedValue({ + status: 200, + statusText: 'OK', + data: 'ok', + }); +}); + +afterEach(() => { + while (openSockets.length) openSockets.pop()?.disconnect(); +}); + +afterAll(async () => { + await env?.shutdown(); +}); + +describe('PushService presence', () => { + it('pushes when nothing is connected', async () => { + const person = await makePerson(); + expect(await pushesAfter(() => puterNotification(person))).toBe(1); + expect(await pushesAfter(() => appMessage(person))).toBe(1); + }); + + it('does not push while the desktop is connected', async () => { + const person = await makePerson(); + await connect(person.token); + + expect(await pushesAfter(() => puterNotification(person))).toBe(0); + expect(await pushesAfter(() => appMessage(person))).toBe(0); + }); + + it('does not push an app’s message while that app is connected', async () => { + const person = await makePerson(); + await connect(person.app.token); + + expect(await pushesAfter(() => appMessage(person))).toBe(0); + // The app's socket does not show Puter's own notifications. + expect(await pushesAfter(() => puterNotification(person))).toBe(1); + }); + + it('does not push while another node in this region holds the desktop', async () => { + const person = await makePerson(); + await env.server.stores.presence.addConnection( + person.id, + PRESENCE_NO_APP, + `sibling-${uuidv4()}`, + ); + + expect(await pushesAfter(() => puterNotification(person))).toBe(0); + }); + + it('does not push while another region holds the app', async () => { + const person = await makePerson(); + await env.server.stores.presence.join( + person.uuid, + person.app.uid, + PEER_ID, + ); + + expect(await pushesAfter(() => appMessage(person))).toBe(0); + expect(await pushesAfter(() => puterNotification(person))).toBe(1); + }); +}); diff --git a/src/backend/services/push/PushService.ts b/src/backend/services/push/PushService.ts index e93d9ee76..fab652b42 100644 --- a/src/backend/services/push/PushService.ts +++ b/src/backend/services/push/PushService.ts @@ -17,11 +17,41 @@ * along with this program. If not, see . */ -import type { - PushDevice, - PushDeviceRegistration, +import type { EventMap } from '../../clients/event/types.js'; +import { + hashPushToken, + type PushDeliveryTarget, + type PushDevice, + type PushDeviceRegistration, } from '../../stores/push/PushDeviceStore.js'; +import { runWithConcurrencyLimitSettled } from '../../util/concurrency.js'; +import { APP_MESSAGE_TYPE } from '../notification/appMessage.js'; import { PuterService } from '../types.js'; +import { hasPushConsent } from './pushConsent.js'; +import { buildPushMessage, fitPushMessage } from './pushMessage.js'; +import { isUserConnected } from './pushPresence.js'; +import type { + PushMessage, + PushPlatform, + PushProvider, + PushResult, +} from './types.js'; + +/** Most tokens one provider call takes. */ +export const PUSH_BATCH_SIZE = 100; +/** Most devices one notification goes to. */ +const MAX_DEVICES_PER_PUSH = 1000; +/** Provider calls one notification has open at once. */ +const BATCH_CONCURRENCY = 4; +/** Notifications this node delivers at once; past it, a push is skipped. */ +const MAX_IN_FLIGHT = 64; +/** A provider call still open after this counts as failed. */ +const SEND_TIMEOUT_MS = 30_000; + +/** User metadata key for pushes of Puter's own notifications; absent is on. */ +export const PUTER_PUSH_PREFERENCE_KEY = 'pushPuter'; + +type WrittenNotification = EventMap['notif.written']; const DEFAULT_DEVICE_RETENTION_DAYS = 60; const STALE_SWEEP_INTERVAL_MS = 60 * 60 * 1000; @@ -31,14 +61,33 @@ const STALE_BATCH_SIZE = 500; const STALE_MAX_BATCHES = 50; const MAX_DEVICES_PER_USER = 20; -/** Keeps push devices in step with their sessions; runs even with push off. */ +/** Pushes new notifications to devices; device cleanup runs with push off. */ export class PushService extends PuterService { #staleSweep: ReturnType | null = null; + readonly #providers = new Map(); + #inFlight = 0; #stopping = false; override onServerStart(): void { this.#armStaleSweep(); + this.clients.event.on('notif.written', (_key, data) => { + if (this.config.push?.enabled !== true) return; + if (this.#providers.size === 0) return; + if (this.#inFlight >= MAX_IN_FLIGHT) { + console.warn('[push] too many pushes in flight, skipped one'); + return; + } + this.#inFlight++; + void this.#deliver(data) + .catch((err: unknown) => { + console.warn('[push] delivery failed', err); + }) + .finally(() => { + this.#inFlight--; + }); + }); + this.clients.event.on( 'outer.auth.sessions.revoked', (_key: string, data: unknown, meta: unknown) => { @@ -79,6 +128,166 @@ export class PushService extends PuterService { return device; } + // -- Extension hooks ---------------------------------------------- + + /** The latest provider registered for a platform carries its pushes. */ + registerProvider(provider: PushProvider): void { + for (const platform of provider.platforms) { + this.#providers.set(platform, provider); + } + } + + // -- Delivery ----------------------------------------------------- + + /** Only `app.message` is an app's; every other type goes to Puter devices. */ + async #deliver(written: WrittenNotification): Promise { + const { userId, uid, type, value } = written; + const fromApp = type === APP_MESSAGE_TYPE; + const appUid = fromApp ? (written.appUid ?? null) : null; + if (fromApp && !appUid) return; + + const devices = ( + await this.stores.pushDevice.listForDelivery(userId, appUid, { + limit: MAX_DEVICES_PER_PUSH, + }) + ).filter((device) => this.#providers.has(device.platform)); + if (devices.length === 0) return; + + if (!(await this.#allowed(userId, appUid))) return; + + const presence = { + socket: this.services.socket, + eventForward: this.services.eventForward, + }; + if (await isUserConnected(presence, userId, null)) return; + if (appUid && (await isUserConnected(presence, userId, appUid))) return; + + const appName = appUid ? await this.#appName(appUid, value) : null; + if (appUid && !appName) return; + const built = buildPushMessage( + { uid, type, value }, + appName ? { name: appName } : null, + ); + if (!built) return; + const message = fitPushMessage(built); + if (!message) { + console.warn('[push] message over the size budget, not sent', { + uid, + type, + }); + return; + } + + await this.#send(devices, message); + } + + /** The name the server stamped on the message, else the app row's. */ + async #appName( + appUid: string, + value: Record, + ): Promise { + const stamped = (value.app as { name?: unknown } | undefined)?.name; + if (typeof stamped === 'string' && stamped) return stamped; + const app = await this.stores.app.getByUid(appUid); + return typeof app?.name === 'string' && app.name ? app.name : null; + } + + /** The user's consent for the app, or their Puter preference without one. */ + async #allowed(userId: number, appUid: string | null): Promise { + if (appUid) { + return hasPushConsent( + { stores: this.stores, services: this.services }, + userId, + appUid, + ); + } + const user = await this.stores.user.getById(userId); + if (!user) return false; + return user.metadata?.[PUTER_PUSH_PREFERENCE_KEY] !== false; + } + + /** One call per platform batch; each device's result is recorded once. */ + async #send( + devices: PushDeliveryTarget[], + message: PushMessage, + ): Promise { + const byPlatform = new Map(); + for (const device of devices) { + const list = byPlatform.get(device.platform) ?? []; + list.push(device); + byPlatform.set(device.platform, list); + } + const batches: Array<{ + platform: PushPlatform; + devices: PushDeliveryTarget[]; + }> = []; + for (const [platform, list] of byPlatform) { + for (let i = 0; i < list.length; i += PUSH_BATCH_SIZE) { + batches.push({ + platform, + devices: list.slice(i, i + PUSH_BATCH_SIZE), + }); + } + } + + const settled = await runWithConcurrencyLimitSettled( + batches, + BATCH_CONCURRENCY, + (batch) => + withTimeout( + this.#providers.get(batch.platform)!.send({ + platform: batch.platform, + tokens: batch.devices.map((device) => device.token), + message, + }), + ), + ); + + const delivered: string[] = []; + const failed: string[] = []; + const unregistered: string[] = []; + const reasons = new Set(); + settled.forEach((outcome, index) => { + let results: Array = []; + if (outcome.status === 'rejected') { + reasons.add(errorName(outcome.reason)); + } else if (Array.isArray(outcome.value)) { + results = outcome.value; + } + batches[index].devices.forEach((device, i) => { + const result = results[i]; + if (result?.status === 'delivered') { + delivered.push(device.uid); + } else if (result?.status === 'unregistered') { + unregistered.push(hashPushToken(device.token)); + } else { + failed.push(device.uid); + if (result?.status === 'failed' && result.reason) { + reasons.add( + String(result.reason) + .split(device.token) + .join('') + .slice(0, 120), + ); + } + } + }); + }); + + if (failed.length > 0) { + console.warn('[push] not delivered to some devices', { + uid: message.uid, + devices: failed.length, + reasons: [...reasons].slice(0, 5), + }); + } + await Promise.all([ + this.stores.pushDevice.touch(delivered), + this.stores.pushDevice.recordFailure(failed), + this.stores.pushDevice.removeByTokenHashes(unregistered), + ]); + } + /** Every node sweeps; an overlap costs only empty indexed batches. */ async sweepStaleDevices(): Promise { const days = this.#retentionDays(); @@ -117,3 +326,19 @@ export class PushService extends PuterService { this.#staleSweep = sweep; } } + +const withTimeout = (promise: Promise): Promise => { + let timer: ReturnType | undefined; + const timeout = new Promise((_resolve, reject) => { + timer = setTimeout( + () => reject(new Error('push provider timed out')), + SEND_TIMEOUT_MS, + ); + timer.unref?.(); + }); + return Promise.race([promise, timeout]).finally(() => clearTimeout(timer)); +}; + +/** Only the error's name: a provider's message may carry tokens. */ +const errorName = (err: unknown): string => + err instanceof Error ? err.name || 'Error' : 'error'; diff --git a/src/backend/services/push/pushMessage.test.ts b/src/backend/services/push/pushMessage.test.ts new file mode 100644 index 000000000..edd271040 --- /dev/null +++ b/src/backend/services/push/pushMessage.test.ts @@ -0,0 +1,242 @@ +/* + * 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 } from 'vitest'; +import { + buildPushMessage, + fitPushMessage, + PUSH_MESSAGE_MAX_BYTES, +} from './pushMessage.js'; +import type { PushMessage } from './types.js'; + +const app = { name: 'chat-app' }; +const bytes = (message: PushMessage): number => + Buffer.byteLength(JSON.stringify(message), 'utf8'); + +describe('buildPushMessage', () => { + it('names the sender and the item for one named share', () => { + const message = buildPushMessage( + { + uid: 'n1', + type: 'share.received', + value: { + title: 'alice shared an item with you', + fields: { + username: 'alice', + count: 1, + senders: [{ username: 'alice', count: 1 }], + target: { + path: '/alice/Shared/report.txt', + name: 'report.txt', + }, + }, + }, + }, + null, + ); + expect(message).toEqual({ + uid: 'n1', + title: 'alice shared report.txt', + group: 'n1', + target: { path: '/alice/Shared/report.txt' }, + ttl: 3600, + priority: 'normal', + }); + }); + + it('uses the stored title for several items or senders', () => { + const message = buildPushMessage( + { + uid: 'n2', + type: 'share.received', + value: { + title: 'alice and bob shared 3 items with you', + fields: { + senders: [ + { username: 'alice', count: 2 }, + { username: 'bob', count: 1 }, + ], + }, + }, + }, + null, + ); + expect(message?.title).toBe('alice and bob shared 3 items with you'); + expect(message?.target).toBeUndefined(); + }); + + it('uses the stored title for share.claimed and developer types', () => { + for (const type of [ + 'share.claimed', + 'app.worker.deployed', + 'app.worker.deployFailed', + 'app.events.ended', + 'app.events.suspended', + ]) { + const message = buildPushMessage( + { uid: 'n3', type, value: { title: `title of ${type}` } }, + null, + ); + expect(message?.title).toBe(`title of ${type}`); + expect(message?.group).toBe('n3'); + expect(message?.target).toBeUndefined(); + } + }); + + it('carries an app message with its group, target and data', () => { + const message = buildPushMessage( + { + uid: 'n4', + type: 'app.message', + value: { + title: 'New reply', + text: 'see you at 5', + icon: 'https://chat.example/icon.png', + target: '/rooms/7', + data: { room: 7 }, + group: 'room-7', + priority: 'normal', + ttl: 120, + }, + }, + app, + ); + expect(message).toEqual({ + uid: 'n4', + title: 'New reply', + body: 'see you at 5', + icon: 'https://chat.example/icon.png', + group: 'room-7', + target: { app: 'chat-app', path: '/rooms/7' }, + data: { room: 7 }, + ttl: 120, + priority: 'normal', + }); + }); + + it('groups an app message by its uid without a group', () => { + const message = buildPushMessage( + { uid: 'n5', type: 'app.message', value: { title: 'hi' } }, + app, + ); + expect(message?.group).toBe('n5'); + expect(message?.target).toEqual({ app: 'chat-app' }); + }); + + it('replaces a sensitive app message with a generic one', () => { + const message = buildPushMessage( + { + uid: 'n6', + type: 'app.message', + value: { + title: 'Code 123456', + text: 'your login code', + icon: 'https://chat.example/icon.png', + target: '/codes/1', + data: { code: 123456 }, + sensitive: true, + ttl: 60, + }, + }, + app, + ); + expect(message).toEqual({ + uid: 'n6', + title: 'chat-app', + body: 'New notification from chat-app', + group: 'n6', + target: { app: 'chat-app' }, + ttl: 60, + priority: 'normal', + sensitive: true, + }); + }); + + it('has nothing for an unknown type or a missing title', () => { + expect( + buildPushMessage( + { uid: 'x', type: 'other', value: { title: 't' } }, + null, + ), + ).toBeNull(); + expect( + buildPushMessage( + { uid: 'x', type: 'share.claimed', value: {} }, + null, + ), + ).toBeNull(); + expect( + buildPushMessage( + { uid: 'x', type: 'app.message', value: { title: 't' } }, + null, + ), + ).toBeNull(); + }); +}); + +describe('fitPushMessage', () => { + const base: PushMessage = { + uid: 'u', + title: 'title', + group: 'u', + ttl: 60, + priority: 'normal', + }; + + it('leaves a message under the budget alone', () => { + const message = { ...base, body: 'short' }; + expect(fitPushMessage(message)).toBe(message); + }); + + it('shortens a long body to fit, by whole characters', () => { + const fitted = fitPushMessage({ ...base, body: '😀'.repeat(2000) }); + expect(fitted).not.toBeNull(); + expect(bytes(fitted!)).toBeLessThanOrEqual(PUSH_MESSAGE_MAX_BYTES); + expect(fitted!.title).toBe('title'); + expect(fitted!.body!.endsWith('…')).toBe(true); + expect( + Array.from(fitted!.body!.slice(0, -1)).every((c) => c === '😀'), + ).toBe(true); + }); + + it('shortens the title once the body is gone, then drops the icon', () => { + const titled = fitPushMessage({ + ...base, + title: 't'.repeat(5000), + body: 'b'.repeat(100), + }); + expect(titled!.body).toBeUndefined(); + expect(bytes(titled!)).toBeLessThanOrEqual(PUSH_MESSAGE_MAX_BYTES); + + const iconless = fitPushMessage({ + ...base, + icon: `https://x.example/${'i'.repeat(3000)}`, + data: { blob: 'd'.repeat(1500) }, + }); + expect(iconless).not.toBeNull(); + expect(iconless!.icon).toBeUndefined(); + expect(iconless!.title).toBe('title'); + }); + + it('never cuts data: too much of it and there is no push', () => { + expect( + fitPushMessage({ ...base, data: { blob: 'd'.repeat(4000) } }), + ).toBeNull(); + }); +}); diff --git a/src/backend/services/push/pushMessage.ts b/src/backend/services/push/pushMessage.ts new file mode 100644 index 000000000..9db295202 --- /dev/null +++ b/src/backend/services/push/pushMessage.ts @@ -0,0 +1,202 @@ +/* + * 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 { + APP_MESSAGE_TTL_DEFAULT_SECONDS, + APP_MESSAGE_TYPE, +} from '../notification/appMessage.js'; +import { + shareNotifyCount, + shareSendersFromFields, +} from '../share/shareNotifyTitle.js'; +import type { PushMessage, PushTarget } from './types.js'; + +/** Under the 4 KB web push payload, with room for a provider to wrap it. */ +export const PUSH_MESSAGE_MAX_BYTES = 3800; + +/** How long a provider may hold one of Puter's own pushes. */ +export const PUSH_DEFAULT_TTL_SECONDS = APP_MESSAGE_TTL_DEFAULT_SECONDS; + +/** What a push is built from: one written notification. */ +export interface PushSource { + uid: string; + type: string; + value: Record; +} + +const str = (value: unknown): string | undefined => + typeof value === 'string' && value.trim() !== '' ? value : undefined; + +const record = (value: unknown): Record => + value && typeof value === 'object' && !Array.isArray(value) + ? (value as Record) + : {}; + +/** "alice shared report.txt" for one named item, otherwise the stored title. */ +const shareReceived = (source: PushSource): PushMessage | null => { + const fields = record(source.value.fields); + const senders = shareSendersFromFields(fields); + const target = record(fields.target); + const name = str(target.name); + const path = str(target.path); + const single = senders.length === 1 && shareNotifyCount(senders) === 1; + const title = + single && name + ? `${senders[0].username} shared ${name}` + : str(source.value.title); + if (!title) return null; + return { + uid: source.uid, + title, + group: source.uid, + ...(single && path ? { target: { path } } : {}), + ttl: PUSH_DEFAULT_TTL_SECONDS, + priority: 'normal', + }; +}; + +const storedTitle = (source: PushSource): PushMessage | null => { + const title = str(source.value.title); + if (!title) return null; + return { + uid: source.uid, + title, + group: source.uid, + ttl: PUSH_DEFAULT_TTL_SECONDS, + priority: 'normal', + }; +}; + +const appMessage = ( + source: PushSource, + app: { name: string }, +): PushMessage | null => { + const value = source.value; + const title = str(value.title); + if (!title) return null; + const ttl = Number(value.ttl); + const base = { + uid: source.uid, + group: str(value.group) ?? source.uid, + ttl: + Number.isInteger(ttl) && ttl > 0 + ? ttl + : APP_MESSAGE_TTL_DEFAULT_SECONDS, + priority: 'normal' as const, + }; + if (value.sensitive === true) { + return { + ...base, + title: app.name, + body: `New notification from ${app.name}`, + target: { app: app.name }, + sensitive: true, + }; + } + const target: PushTarget = { app: app.name }; + const path = str(value.target); + if (path) target.path = path; + const body = str(value.text); + const icon = str(value.icon); + return { + ...base, + title, + ...(body ? { body } : {}), + ...(icon ? { icon } : {}), + target, + ...(value.data !== undefined ? { data: value.data } : {}), + }; +}; + +/** Null for a type with no push wording, or a value missing its title. */ +export const buildPushMessage = ( + source: PushSource, + app: { name: string } | null, +): PushMessage | null => { + switch (source.type) { + case 'share.received': + return shareReceived(source); + case 'share.claimed': + case 'app.worker.deployed': + case 'app.worker.deployFailed': + case 'app.events.ended': + case 'app.events.suspended': + return storedTitle(source); + case APP_MESSAGE_TYPE: + return app ? appMessage(source, app) : null; + default: + return null; + } +}; + +const fits = (message: PushMessage): boolean => + Buffer.byteLength(JSON.stringify(message), 'utf8') <= + PUSH_MESSAGE_MAX_BYTES; + +/** Whole characters, so a surrogate pair is never split. */ +const cut = (chars: string[], n: number): string => { + if (n >= chars.length) return chars.join(''); + return n <= 0 ? '' : `${chars.slice(0, n - 1).join('')}…`; +}; + +/** Largest n in [0, max] whose build fits, or -1. */ +const longestFitting = ( + max: number, + build: (n: number) => PushMessage, +): number => { + if (fits(build(max))) return max; + if (max === 0 || !fits(build(0))) return -1; + let lo = 0; + let hi = max - 1; + while (lo < hi) { + const mid = Math.ceil((lo + hi) / 2); + if (fits(build(mid))) lo = mid; + else hi = mid - 1; + } + return lo; +}; + +/** Cuts body, then title, then icon; null when `data` alone leaves no room. */ +export const fitPushMessage = (message: PushMessage): PushMessage | null => { + if (fits(message)) return message; + const titleChars = Array.from(message.title); + const bodyChars = Array.from(message.body ?? ''); + const { body: _body, icon, ...rest } = message; + for (const withIcon of icon === undefined ? [false] : [true, false]) { + const build = (titleLen: number, bodyLen: number): PushMessage => { + const text = cut(bodyChars, bodyLen); + return { + ...rest, + title: cut(titleChars, titleLen), + ...(text ? { body: text } : {}), + ...(withIcon ? { icon } : {}), + }; + }; + const bodyLen = longestFitting(bodyChars.length, (n) => + build(titleChars.length, n), + ); + if (bodyLen >= 0) return build(titleChars.length, bodyLen); + // A title cut to nothing is no push; keep at least one character. + const titleLen = longestFitting(titleChars.length, (n) => + build(Math.max(1, n), 0), + ); + if (titleLen >= 0) return build(Math.max(1, titleLen), 0); + } + return null; +}; diff --git a/src/backend/services/push/pushPresence.ts b/src/backend/services/push/pushPresence.ts new file mode 100644 index 000000000..70376e0a6 --- /dev/null +++ b/src/backend/services/push/pushPresence.ts @@ -0,0 +1,49 @@ +/* + * 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 { + forwardTarget, + type EventForwardService, +} from '../events/EventForwardService.js'; +import type { SocketService } from '../socket/SocketService.js'; + +export interface PushPresenceDeps { + socket: Pick; + eventForward: Pick< + EventForwardService, + 'active' | 'heldInRegion' | 'regionsFor' + >; +} + +/** Other nodes and regions count only where presence is kept (peers set). */ +export const isUserConnected = async ( + deps: PushPresenceDeps, + userId: number, + appUid: string | null, +): Promise => { + if (deps.socket.has(forwardTarget(userId, appUid))) return true; + if (!deps.eventForward.active) return false; + try { + if (await deps.eventForward.heldInRegion(userId, appUid)) return true; + return (await deps.eventForward.regionsFor(userId, appUid)).length > 0; + } catch (err) { + console.warn('[push] presence read failed', err); + return false; + } +}; diff --git a/src/backend/services/push/types.ts b/src/backend/services/push/types.ts new file mode 100644 index 000000000..b21612faf --- /dev/null +++ b/src/backend/services/push/types.ts @@ -0,0 +1,65 @@ +/* + * 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 type { PushPlatform } from '../../stores/push/PushDeviceStore.js'; + +export type { PushPlatform }; + +/** Where a tap leads, read by Puter's own client. No app means the desktop. */ +export interface PushTarget { + /** App name. */ + app?: string; + path?: string; +} + +/** One push, independent of the provider that carries it. */ +export interface PushMessage { + /** The notification's uid. */ + uid: string; + title: string; + body?: string; + icon?: string; + /** A later push with the same group replaces this one on the device. */ + group: string; + target?: PushTarget; + data?: unknown; + /** Seconds a provider may hold it for an offline device. */ + ttl: number; + priority: 'normal'; + /** Content was withheld; the provider should not add any of its own. */ + sensitive?: boolean; +} + +export type PushResult = + | { status: 'delivered' } + /** The provider says the token no longer names a device. */ + | { status: 'unregistered' } + | { status: 'failed'; reason?: string }; + +export interface PushBatch { + platform: PushPlatform; + tokens: string[]; + message: PushMessage; +} + +/** `send` resolves one result per token, in the same order. */ +export interface PushProvider { + platforms: readonly PushPlatform[]; + send(batch: PushBatch): Promise; +} diff --git a/src/docs/src/rate-limits-and-quotas.md b/src/docs/src/rate-limits-and-quotas.md index f86cd9e5e..9ae511188 100644 --- a/src/docs/src/rate-limits-and-quotas.md +++ b/src/docs/src/rate-limits-and-quotas.md @@ -405,6 +405,17 @@ An app, its workers and the tokens it issues can send the user they act for an ` Over the rate limit the call fails with `429 too_many_requests`, and a refused call counts against neither window. A value over any size limit, or with a field the format doesn't have, fails with `400 bad_request`. A deployment can change the rate limits with `notificationAppMessageLimits: { perMinute, perDay }`. +Where the deployment sends push notifications, these apply to each push: + +| Limit | Value | Scope | +| --------------------------------- | -------------------- | ---------------- | +| Push size | 3,800 bytes, as JSON | Per notification | +| Devices one push reaches | 1,000 | Per notification | +| Pushes being sent at once | 64 | Per server | +| Time a push service has to answer | 30 seconds | Per call | + +A push over the size limit is shortened. The `text` is cut first, then the `title`, and then the icon is left out. If the `data` alone doesn't fit, the notification isn't pushed. Past the number of pushes being sent at once, a notification isn't pushed. A notification that isn't pushed is still in the user's notification list. + ### Teams Available only where the deployment has teams turned on; elsewhere `puter.teams` rejects with `not_found`.