mirror of
https://github.com/HeyPuter/puter.git
synced 2026-11-07 06:23:05 +00:00
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.
This commit is contained in:
1 parent
53bf7e9303
commit
4fd9fae926
12 files changed
+1596
-11
No files matched your search
@@ -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<string, unknown>;
|
||||
/** 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 string> =
|
||||
K extends `${infer Head}.${infer Tail}`
|
||||
?
|
||||
| `${Head}.*`
|
||||
| (Tail extends `${string}.${string}`
|
||||
? `${Head}.${WildcardPrefixes<Tail>}`
|
||||
: never)
|
||||
? | `${Head}.*`
|
||||
| (Tail extends `${string}.${string}`
|
||||
? `${Head}.${WildcardPrefixes<Tail>}`
|
||||
: never)
|
||||
: never;
|
||||
|
||||
export type ListenKey = EventKey | WildcardPrefixes<EventKey>;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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 <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
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<PushResult[]> {
|
||||
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<string, unknown>, 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<TestUser> =>
|
||||
(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<string, unknown>,
|
||||
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('<token>');
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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 <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
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<PushResult[]> {
|
||||
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<Person> => {
|
||||
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<ClientSocket> =>
|
||||
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<unknown>): Promise<number> => {
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -17,11 +17,41 @@
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
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<typeof setInterval> | null = null;
|
||||
readonly #providers = new Map<PushPlatform, PushProvider>();
|
||||
#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<void> {
|
||||
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<string, unknown>,
|
||||
): Promise<string | null> {
|
||||
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<boolean> {
|
||||
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<void> {
|
||||
const byPlatform = new Map<PushPlatform, PushDeliveryTarget[]>();
|
||||
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<string>();
|
||||
settled.forEach((outcome, index) => {
|
||||
let results: Array<PushResult | undefined> = [];
|
||||
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('<token>')
|
||||
.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<number> {
|
||||
const days = this.#retentionDays();
|
||||
@@ -117,3 +326,19 @@ export class PushService extends PuterService {
|
||||
this.#staleSweep = sweep;
|
||||
}
|
||||
}
|
||||
|
||||
const withTimeout = <T>(promise: Promise<T>): Promise<T> => {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
const timeout = new Promise<never>((_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';
|
||||
@@ -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 <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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 <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
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<string, unknown>;
|
||||
}
|
||||
|
||||
const str = (value: unknown): string | undefined =>
|
||||
typeof value === 'string' && value.trim() !== '' ? value : undefined;
|
||||
|
||||
const record = (value: unknown): Record<string, unknown> =>
|
||||
value && typeof value === 'object' && !Array.isArray(value)
|
||||
? (value as Record<string, unknown>)
|
||||
: {};
|
||||
|
||||
/** "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;
|
||||
};
|
||||
@@ -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 <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
import {
|
||||
forwardTarget,
|
||||
type EventForwardService,
|
||||
} from '../events/EventForwardService.js';
|
||||
import type { SocketService } from '../socket/SocketService.js';
|
||||
|
||||
export interface PushPresenceDeps {
|
||||
socket: Pick<SocketService, 'has'>;
|
||||
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<boolean> => {
|
||||
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;
|
||||
}
|
||||
};
|
||||
@@ -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 <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
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<PushResult[]>;
|
||||
}
|
||||
@@ -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`.
|
||||
|
||||
Reference in new issue
Block a user