feat(push): one push a minute per user per app

src/backend/services/push/PushService.delivery.test.ts

@@ -643,3 +643,214 @@ describe('PushService with the notification fold-in on', () => {
         state.server.clients.event.off?.('notif.created', created);
     });
 });
+
+/** A user who granted push to a fresh app and has one device for it. */
+const appPushReady = async (server: PuterServer, user?: TestUser) => {
+    const owner = user ?? (await makeUser(server));
+    const app = await makeApp(server, owner);
+    await grantPush(server, owner, app.uid);
+    const token = `tok-${uuidv4()}`;
+    await registerDevice(server, owner.id, token, { appUid: app.uid });
+    return { user: owner, app, token };
+};
+
+const appMessage = (server: PuterServer, userId: number, appUid: string) =>
+    server.services.notification.notify(
+        [userId],
+        { title: 'New reply' },
+        { type: 'app.message', appUid },
+    );
+
+const limitWarnings = (warn: { mock: { calls: unknown[][] } }) =>
+    warn.mock.calls.filter((call) =>
+        String(call[0]).includes('app push limit'),
+    );
+
+describe('PushService app push limit', () => {
+    const provider = new FakeProvider();
+    const state = bootWith({ push: { enabled: true } }, provider);
+    const server = () => state.server;
+
+    beforeEach(() => {
+        provider.calls.length = 0;
+    });
+
+    it('pushes one app message a minute and still stores the next', async () => {
+        const { user, app } = await appPushReady(server());
+
+        await appMessage(server(), user.id, app.uid);
+        await vi.waitFor(() => expect(provider.calls).toHaveLength(1));
+        const second = await appMessage(server(), user.id, app.uid);
+        await settle();
+
+        expect(provider.calls).toHaveLength(1);
+        const stored = await server().stores.notification.getByUid(second!, {
+            userId: user.id,
+        });
+        expect(stored).toBeTruthy();
+    });
+
+    it('counts a message that would replace an earlier push', async () => {
+        const { user, app } = await appPushReady(server());
+        const inGroup = () =>
+            server().services.notification.notify(
+                [user.id],
+                { title: 'New reply', group: 'room-7' },
+                { type: 'app.message', appUid: app.uid },
+            );
+
+        await inGroup();
+        await vi.waitFor(() => expect(provider.calls).toHaveLength(1));
+        await inGroup();
+        await settle();
+
+        expect(provider.calls).toHaveLength(1);
+        expect(provider.calls[0].message.group).toBe(`${app.uid}:room-7`);
+    });
+
+    it('counts each user and each app on its own', async () => {
+        const first = await appPushReady(server());
+        const otherUser = await makeUser(server());
+        await grantPush(server(), otherUser, first.app.uid);
+        await registerDevice(server(), otherUser.id, `tok-${uuidv4()}`, {
+            appUid: first.app.uid,
+        });
+        const otherApp = await appPushReady(server(), first.user);
+
+        await appMessage(server(), first.user.id, first.app.uid);
+        await appMessage(server(), otherUser.id, first.app.uid);
+        await appMessage(server(), first.user.id, otherApp.app.uid);
+
+        await vi.waitFor(() => expect(provider.calls).toHaveLength(3));
+        await settle();
+        expect(provider.calls).toHaveLength(3);
+    });
+
+    it('does not count Puter’s own notifications', async () => {
+        const { user, app } = await appPushReady(server());
+        await registerDevice(server(), user.id, `tok-${uuidv4()}`);
+
+        await appMessage(server(), user.id, app.uid);
+        await shareReceived(server(), user.id);
+        await shareReceived(server(), user.id);
+
+        await vi.waitFor(() => expect(provider.calls).toHaveLength(3));
+        await settle();
+        expect(provider.calls).toHaveLength(3);
+    });
+
+    it('spends nothing on a message with no devices', async () => {
+        const user = await makeUser(server());
+        const app = await makeApp(server(), user);
+        await grantPush(server(), user, app.uid);
+
+        await appMessage(server(), user.id, app.uid);
+        await settle();
+        await registerDevice(server(), user.id, `tok-${uuidv4()}`, {
+            appUid: app.uid,
+        });
+        await appMessage(server(), user.id, app.uid);
+
+        await vi.waitFor(() => expect(provider.calls).toHaveLength(1));
+    });
+
+    it('spends nothing on a message without consent', async () => {
+        const user = await makeUser(server());
+        const app = await makeApp(server(), user);
+        await registerDevice(server(), user.id, `tok-${uuidv4()}`, {
+            appUid: app.uid,
+        });
+
+        await appMessage(server(), user.id, app.uid);
+        await settle();
+        expect(provider.calls).toHaveLength(0);
+        await grantPush(server(), user, app.uid);
+        await appMessage(server(), user.id, app.uid);
+
+        await vi.waitFor(() => expect(provider.calls).toHaveLength(1));
+    });
+
+    it('spends nothing while the user is connected', async () => {
+        const { user, app } = await appPushReady(server());
+        const has = vi
+            .spyOn(server().services.socket, 'has')
+            .mockReturnValue(true);
+        await appMessage(server(), user.id, app.uid);
+        await settle();
+        has.mockRestore();
+        expect(provider.calls).toHaveLength(0);
+
+        await appMessage(server(), user.id, app.uid);
+        await vi.waitFor(() => expect(provider.calls).toHaveLength(1));
+    });
+});
+
+describe('PushService with no app push limit', () => {
+    const provider = new FakeProvider();
+    const state = bootWith(
+        { push: { enabled: true, appPushesPerMinute: 0 } },
+        provider,
+    );
+
+    it('pushes every app message', async () => {
+        const { user, app } = await appPushReady(state.server);
+        for (let i = 0; i < 4; i++) {
+            await appMessage(state.server, user.id, app.uid);
+        }
+        await vi.waitFor(() => expect(provider.calls).toHaveLength(4));
+    });
+});
+
+describe('PushService with three app pushes a minute', () => {
+    const provider = new FakeProvider();
+    const state = bootWith(
+        { push: { enabled: true, appPushesPerMinute: 3 } },
+        provider,
+    );
+
+    it('pushes three app messages and skips the fourth', async () => {
+        const { user, app } = await appPushReady(state.server);
+        for (let i = 0; i < 4; i++) {
+            await appMessage(state.server, user.id, app.uid);
+            await settle();
+        }
+        expect(provider.calls).toHaveLength(3);
+    });
+});
+
+describe('PushService app push limit over time', () => {
+    const provider = new FakeProvider();
+    const state = bootWith({ push: { enabled: true } }, provider);
+
+    it('pushes again once the minute has passed, and warns once a minute', async () => {
+        const { user, app, token } = await appPushReady(state.server);
+        const warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
+        vi.useFakeTimers({ toFake: ['Date'] });
+        try {
+            for (let i = 0; i < 3; i++) {
+                await appMessage(state.server, user.id, app.uid);
+                await settle();
+            }
+            expect(provider.calls).toHaveLength(1);
+            expect(limitWarnings(warn)).toHaveLength(1);
+            expect(limitWarnings(warn)[0][1]).toEqual({ skipped: 1 });
+
+            vi.setSystemTime(Date.now() + 61_000);
+            await appMessage(state.server, user.id, app.uid);
+            await vi.waitFor(() => expect(provider.calls).toHaveLength(2));
+            await appMessage(state.server, user.id, app.uid);
+            await settle();
+            expect(provider.calls).toHaveLength(2);
+
+            const warnings = limitWarnings(warn);
+            expect(warnings).toHaveLength(2);
+            expect(warnings[1][1]).toEqual({ skipped: 2 });
+            const logged = JSON.stringify(warnings);
+            expect(logged).not.toContain(token);
+            expect(logged).not.toContain(app.uid);
+        } finally {
+            vi.useRealTimers();
+            warn.mockRestore();
+        }
+    });
+});

src/backend/services/push/PushService.ts

@@ -18,6 +18,7 @@
  */
 
 import type { EventMap } from '../../clients/event/types.js';
+import { checkRateLimit } from '../../core/http/middleware/rateLimit.js';
 import {
     hashPushToken,
     type PushDeliveryTarget,
@@ -50,10 +51,12 @@ const MAX_DEVICES_PER_PUSH = 1000;
 const BATCH_CONCURRENCY = 4;
 /** Notifications this node delivers at once; past it, a push is skipped. */
 const MAX_IN_FLIGHT = 64;
-/** At most one warning per window about pushes skipped at the cap. */
+/** At most one warning per window for each reason pushes are skipped. */
 const SKIP_WARN_INTERVAL_MS = 60_000;
 /** A provider call still open after this counts as failed. */
 const SEND_TIMEOUT_MS = 30_000;
+const DEFAULT_APP_PUSHES_PER_MINUTE = 1;
+const MINUTE_MS = 60_000;
 
 /** User metadata key for pushes of Puter's own notifications; absent is on. */
 export const PUTER_PUSH_PREFERENCE_KEY = 'pushPuter';
@@ -73,8 +76,12 @@ export class PushService extends PuterService {
     #staleSweep: ReturnType<typeof setInterval> | null = null;
     readonly #providers = new Map<PushPlatform, PushProvider>();
     #inFlight = 0;
-    #skipped = 0;
-    #lastSkipWarn = 0;
+    readonly #capSkips = new SkipLog(
+        '[push] at the in-flight cap, pushes skipped',
+    );
+    readonly #limitSkips = new SkipLog(
+        '[push] over the app push limit, pushes skipped',
+    );
     #stopping = false;
 
     override onServerStart(): void {
@@ -159,7 +166,7 @@ export class PushService extends PuterService {
         if (devices.length === 0 || this.#stopping) return;
 
         if (this.#inFlight >= MAX_IN_FLIGHT) {
-            this.#noteSkipped();
+            this.#capSkips.note();
             return;
         }
         this.#inFlight++;
@@ -201,19 +208,28 @@ export class PushService extends PuterService {
             });
             return;
         }
+        if (appUid && !(await this.#withinAppPushLimit(userId, appUid))) {
+            this.#limitSkips.note();
+            return;
+        }
 
         await this.#send(userId, devices, message);
     }
 
-    #noteSkipped(): void {
-        this.#skipped++;
-        const now = Date.now();
-        if (now - this.#lastSkipWarn < SKIP_WARN_INTERVAL_MS) return;
-        console.warn('[push] at the in-flight cap, pushes skipped', {
-            skipped: this.#skipped,
-        });
-        this.#skipped = 0;
-        this.#lastSkipWarn = now;
+    /** Spends one push from the app's budget for this user; fails open. */
+    async #withinAppPushLimit(
+        userId: number,
+        appUid: string,
+    ): Promise<boolean> {
+        const perMinute =
+            this.config.push?.appPushesPerMinute ??
+            DEFAULT_APP_PUSHES_PER_MINUTE;
+        if (!(perMinute > 0)) return true;
+        return checkRateLimit(
+            `push:app:${userId}:${appUid}`,
+            perMinute,
+            MINUTE_MS,
+        );
     }
 
     /** The name the server stamped on the message, else the app row's. */
@@ -364,6 +380,22 @@ export class PushService extends PuterService {
 
 class PushTimeoutError extends Error {}
 
+/** Counts skipped pushes; warns with the count at most once a window. */
+class SkipLog {
+    #skipped = 0;
+    #lastWarn = 0;
+    constructor(readonly message: string) {}
+
+    note(): void {
+        this.#skipped++;
+        const now = Date.now();
+        if (now - this.#lastWarn < SKIP_WARN_INTERVAL_MS) return;
+        console.warn(this.message, { skipped: this.#skipped });
+        this.#skipped = 0;
+        this.#lastWarn = now;
+    }
+}
+
 /** A provider's own code when it is a known one, never its text. */
 const failureCode = (
     result: PushResult | undefined,

src/backend/types.ts

@@ -987,6 +987,8 @@ interface IConfigOptional {
         enabled?: boolean;
         /** Days a device may go unseen before removal. Default 60; 0 keeps it. */
         deviceRetentionDays?: number;
+        /** Pushes a minute per user per app. Default 1; 0 or less is no limit. */
+        appPushesPerMinute?: number;
     };
 
     // -- Storage / S3 ------------------------------------------------

src/docs/src/rate-limits-and-quotas.md

@@ -407,15 +407,18 @@ Over the rate limit the call fails with `429 too_many_requests`, and a refused c
 
 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         |
+| 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          |
+| App pushes                        | 1/min                | Per user, per app |
 
 A push over the size limit is shortened. The `text` is cut first, then the icon is left out, and then the `title` is cut. If the `title` would keep fewer than 8 characters, or 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.
 
+The app push limit counts every push of an `app.message`, including one that replaces an earlier push in the same group. Over it, the push is skipped and the notification still shows in Puter. Puter's own notifications don't count, and neither does a notification that isn't pushed for another reason. If the limit can't be checked, the push is sent. A deployment can change the limit with `push: { appPushesPerMinute }`; 0 or less removes it.
+
 ### Teams
 
 Available only where the deployment has teams turned on; elsewhere `puter.teams` rejects with `not_found`.