feat(events): fetch can start at the newest notification, and the feed uses it (#4479)

* feat(events): fetch can start from the newest event

GET /events/fetch takes from=end: with no cursor it answers an empty page
whose cursor sits at the newest row of the slice, so later reads return
only rows created or replaced after it. puter.events.fetch() passes
`from` through. Adds a (user_id, audience, app_uid, id) index on
notification so the newest row of a slice is one seek.

* fix(puter-js): onPush starts its feed from the newest notification

A notifications listener started without catch-up asks for from: 'end'
and keeps the cursor it gets back instead of reading every stored
notification to mark them as seen. A server that ignores `from` answers
with the oldest page, and the listener reads through it as before.

* fix(events): read the newest notification id from the primary

* fix(events): mark the end cursor at the slice's latest replacement

The end cursor's replacement mark is the newest (replaced_at, id) in the
slice rather than the serving node's clock, since replaced_at is stamped
by whichever node wrote it.

* docs(events): describe the end cursor's replacement mark

doc/pagination.md

@@ -63,6 +63,9 @@ store:
   sealed cursor: the last `id` it returned, and the last `(replaced_at, id)`
   among rows up to that id. Each is its own indexed read, merged into one
   page, so a row replaced after a client read it comes back from that cursor.
+  `from=end` with no cursor answers an empty page whose cursor holds the
+  slice's newest `id` and the slice's latest `(replaced_at, id)` as the
+  replacement mark.
 
 Cursors that carry a sort also pin it: a request that passes a cursor plus a
 conflicting sort is rejected with 400.

src/backend/clients/database/SqliteDatabaseClient.test.ts

@@ -27,7 +27,7 @@ import { DatabaseClientFactory } from './index.js';
 import { SqliteDatabaseClient } from './SqliteDatabaseClient.js';
 
 /** Highest schema version the migration table can reach. */
-const CURRENT_SCHEMA_VERSION = 97;
+const CURRENT_SCHEMA_VERSION = 98;
 
 /**
  * These suites migrate real files on disk. Idle they finish in well under a

src/backend/clients/database/SqliteDatabaseClient.ts

@@ -136,6 +136,7 @@ const AVAILABLE_MIGRATIONS: [number, string[]][] = [
     [94, ['0099_kv-share-handle-notify.sql']],
     [95, ['0100_notification-send-key.sql']],
     [96, ['0101_user-scoped-handlers.sql']],
+    [97, ['0102_notification-scope-id-index.sql']],
 ];
 
 export class SqliteDatabaseClient extends AbstractDatabaseClient {

src/backend/clients/database/migrations/mysql/mysql_mig_56.sql

@@ -0,0 +1,41 @@
+-- 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/>.
+
+-- `notification` (`user_id`, `audience`, `app_uid`, `id`). See
+-- sqlite/0102_notification-scope-id-index.sql for the rationale. No per-file
+-- applied-state tracking, so the index is added only when missing and a
+-- replay does nothing.
+
+DROP PROCEDURE IF EXISTS _puter_add_notification_scope_id_index;
+DELIMITER //
+CREATE PROCEDURE _puter_add_notification_scope_id_index()
+BEGIN
+  IF NOT EXISTS (
+    SELECT 1 FROM INFORMATION_SCHEMA.STATISTICS
+    WHERE TABLE_SCHEMA = DATABASE()
+      AND TABLE_NAME = 'notification'
+      AND INDEX_NAME = 'idx_notification_scope_id'
+  ) THEN
+    ALTER TABLE `notification` ADD INDEX `idx_notification_scope_id`
+      (`user_id`, `audience`, `app_uid`, `id`), ALGORITHM=INPLACE, LOCK=NONE;
+  END IF;
+END//
+DELIMITER ;
+
+CALL _puter_add_notification_scope_id_index();
+
+DROP PROCEDURE IF EXISTS _puter_add_notification_scope_id_index;

src/backend/clients/database/migrations/postgres/postgres_mig_45.sql

@@ -0,0 +1,23 @@
+-- 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/>.
+
+-- notification (user_id, audience, app_uid, id). See
+-- sqlite/0102_notification-scope-id-index.sql for the rationale. Idempotent via
+-- IF NOT EXISTS.
+
+CREATE INDEX IF NOT EXISTS idx_notification_scope_id
+    ON notification (user_id, audience, app_uid, id);

src/backend/clients/database/migrations/sqlite/0102_notification-scope-id-index.sql

@@ -0,0 +1,22 @@
+-- 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/>.
+
+-- Index a mailbox slice by row id, so a fetch finds the newest row of a slice,
+-- and pages forward through one, with a seek rather than a scan of the slice.
+
+CREATE INDEX IF NOT EXISTS `idx_notification_scope_id`
+    ON `notification` (`user_id`, `audience`, `app_uid`, `id`);

src/backend/clients/event/types.ts

@@ -1041,11 +1041,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>;
@@ -1056,10 +1055,7 @@ export type MatchingEvents<P extends ListenKey> = P extends `${infer Prefix}.*`
 
 /** Acked, not acked, no socket anywhere, or sockets only in a peer region. */
 export type NotifDeliveryOutcome =
-    | 'received'
-    | 'unreceived'
-    | 'no-socket'
-    | 'remote-only';
+    'received' | 'unreceived' | 'no-socket' | 'remote-only';
 
 /** One socket's receipt for a live notification. */
 export interface NotifDeliveryResponder {

src/backend/controllers/events/EventsController.ts

@@ -134,7 +134,8 @@ export class EventsController extends PuterController {
      *
      * Stateless: no subscription, no stored position, nothing written. The
      * cursor comes back to the caller and goes out again as `cursor` (or
-     * `after`, the name it first shipped under).
+     * `after`, the name it first shipped under). `from=end` with no cursor
+     * starts at the newest row instead of the oldest.
      */
     @Get('/fetch', {
         subdomain: 'api',
@@ -145,9 +146,16 @@ export class EventsController extends PuterController {
     async fetch(req: Request, res: Response): Promise<void> {
         const actor = this.#requireActor(req);
         const query = (req.query ?? {}) as Record<string, unknown>;
+        const from = query.from as 'start' | 'end' | undefined;
+        if (from !== undefined && from !== 'start' && from !== 'end') {
+            throw new HttpError(400, '`from` must be `start` or `end`', {
+                legacyCode: 'bad_request',
+            });
+        }
 
         const page = await this.services.events.fetchMissed(actor, {
             subject: typeof query.subject === 'string' ? query.subject : '',
+            from,
             cursor: typeof query.cursor === 'string' ? query.cursor : undefined,
             after: typeof query.after === 'string' ? query.after : undefined,
             limit: normalizeLimit(query.limit, {

src/backend/services/events/EventsService.ts

@@ -206,6 +206,7 @@ import {
     notifFetchPosition,
     projectNotifRow,
     resolveNotifFetch,
+    type NotifFetchScope,
 } from './notifFetch.js';
 import {
     lookupFsSubject,
@@ -489,6 +490,8 @@ export interface FetchRequest {
     cursor?: string;
     /** The name `cursor` shipped under first; still read when it is absent. */
     after?: string;
+    /** With no cursor, `'end'` answers an empty page at the newest row. */
+    from?: 'start' | 'end';
     limit?: number;
 }
 
@@ -2217,6 +2220,13 @@ export class EventsService extends PuterService {
             if (!visible) return { items: [] };
         }
 
+        const opened = openCursor(
+            request.cursor ?? request.after,
+            this.config.jwt_secret_v2,
+        );
+        if (opened === undefined && request.from === 'end')
+            return this.#fetchEnd(user.id, scope);
+
         const asked = Math.floor(Number(request.limit));
         const limit = Math.min(
             Number.isFinite(asked) && asked > 0
@@ -2224,12 +2234,7 @@ export class EventsService extends PuterService {
                 : EVENTS_FETCH_LIMIT_DEFAULT,
             EVENTS_FETCH_LIMIT_CAP,
         );
-        const from = notifFetchPosition(
-            openCursor(
-                request.cursor ?? request.after,
-                this.config.jwt_secret_v2,
-            ),
-        );
+        const from = notifFetchPosition(opened);
 
         // One extra row answers "is there another page" without a count.
         // The generic slice is the desktop's view of the mailbox, so it leaves
@@ -2278,6 +2283,32 @@ export class EventsService extends PuterService {
         };
     }
 
+    /**
+     * An empty page whose cursor sits at the newest row of `scope` and at its
+     * latest replacement: reads from it return rows inserted or replaced
+     * later.
+     */
+    async #fetchEnd(
+        userId: number,
+        scope: NotifFetchScope,
+    ): Promise<PageResult<ProjectedNotifEvent>> {
+        const slice = { audience: scope.audience, appUid: scope.appUid };
+        const [latest, replaced] = await Promise.all([
+            this.stores.notification.latestScopedId(userId, slice),
+            this.stores.notification.latestScopedReplacement(userId, slice),
+        ]);
+        return {
+            items: [],
+            cursor: sealCursor(
+                notifFetchCursor({
+                    id: latest ?? 0,
+                    replaced: replaced ?? { replacedAt: 0, id: 0 },
+                }),
+                this.config.jwt_secret_v2,
+            ),
+        };
+    }
+
     /**
      * The audience predicate, applied to rows the query already scoped to the
      * caller's own mailbox. A row the actor may not see is dropped rather than

src/backend/services/events/fetch.integration.test.ts

@@ -616,6 +616,210 @@ describe('reading again from a saved cursor', () => {
     });
 });
 
+describe('starting from the end', () => {
+    const SUBJECT = 'notif:app-user';
+
+    const message = async (title: string, inApp = appUid): Promise<string> =>
+        (
+            await env.server.stores.notification.create({
+                userId,
+                value: { type: 'app.message', title },
+                type: 'app.message',
+                audience: 'app-user',
+                appUid: inApp,
+                group: `group-${uuidv4()}`,
+            })
+        ).uid as string;
+
+    const fromEnd = async (token = appToken): Promise<string> => {
+        const end = await fetchPage(token, { subject: SUBJECT, from: 'end' });
+        expect(end.status).toBe(200);
+        expect(end.body.items).toEqual([]);
+        expect(end.body.cursor).toBeTruthy();
+        return end.body.cursor!;
+    };
+
+    it('answers a cursor on an empty slice that returns what comes after', async () => {
+        await clearMailboxes();
+        const cursor = await fromEnd();
+
+        const later = await message('later');
+        const next = await fetchPage(appToken, { subject: SUBJECT, cursor });
+        expect(uidsOf(next)).toEqual([later]);
+    });
+
+    it('skips what is there and returns only newer rows', async () => {
+        await clearMailboxes();
+        for (let i = 0; i < 3; i++) await message(`before ${i}`);
+        const cursor = await fromEnd();
+
+        const quiet = await fetchPage(appToken, { subject: SUBJECT, cursor });
+        expect(quiet.body.items).toEqual([]);
+        expect(quiet.body.cursor).toBeUndefined();
+
+        const later = [await message('after 0'), await message('after 1')];
+        const next = await fetchPage(appToken, { subject: SUBJECT, cursor });
+        expect(uidsOf(next)).toEqual(later);
+    });
+
+    it('returns a row replaced after the cursor was taken', async () => {
+        await clearMailboxes();
+        const old = await message('old');
+        await message('older still unchanged');
+        const cursor = await fromEnd();
+
+        await env.server.services.notification.notifyUpdate(
+            old,
+            userId,
+            { title: 'replaced' },
+            { type: 'app.message', replace: true },
+        );
+
+        const next = await fetchPage(appToken, { subject: SUBJECT, cursor });
+        expect(uidsOf(next)).toEqual([old]);
+        expect(next.body.items![0]).toMatchObject({
+            replaced: true,
+            notification: { title: 'replaced' },
+        });
+    });
+
+    it('returns a replacement stamped by a clock behind the server`s', async () => {
+        await clearMailboxes();
+        const behind = Date.now() - 60_000;
+        const [early, late] = [await message('early'), await message('late')];
+        const replace = (uid: string, title: string, replacedAt: number) =>
+            env.server.stores.notification.updateValue(
+                uid,
+                userId,
+                { type: 'app.message', title },
+                { onlyUnseen: true, replacedAt },
+            );
+        await replace(early, 'replaced before', behind - 1_000);
+        const cursor = await fromEnd();
+
+        await replace(late, 'replaced after', behind);
+
+        const next = await fetchPage(appToken, { subject: SUBJECT, cursor });
+        expect(uidsOf(next)).toEqual([late]);
+        expect(next.body.items![0]).toMatchObject({
+            notification: { title: 'replaced after' },
+        });
+    });
+
+    it('keeps to the slice the subject names', async () => {
+        await clearMailboxes();
+        await message('own before');
+        const cursor = await fromEnd();
+
+        await message('other app', otherAppUid);
+        await seed(userId, { title: 'account row' });
+        const own = await message('own after');
+
+        const next = await fetchPage(appToken, { subject: SUBJECT, cursor });
+        expect(uidsOf(next)).toEqual([own]);
+    });
+
+    it('answers no cursor for a slice the caller cannot see', async () => {
+        await clearMailboxes();
+        await message('other app', otherAppUid);
+
+        for (const [token, subject] of [
+            [appToken, `notif:${otherAppUid}:app-user`],
+            [appToken, `notif:${userUuid}:account`],
+            [appReadUrlToken, SUBJECT],
+        ] as const) {
+            const page = await fetchPage(token, { subject, from: 'end' });
+            expect(page.status).toBe(200);
+            expect(page.body.items).toEqual([]);
+            expect(page.body.cursor).toBeUndefined();
+        }
+    });
+
+    it('reads from the cursor when one is given', async () => {
+        await clearMailboxes();
+        const uids = [await message('m0'), await message('m1')];
+        const first = await fetchPage(appToken, { subject: SUBJECT, limit: 1 });
+
+        const page = await fetchPage(appToken, {
+            subject: SUBJECT,
+            cursor: first.body.cursor,
+            from: 'end',
+        });
+        expect(uidsOf(page)).toEqual([uids[1]]);
+    });
+
+    it('rejects a `from` that is neither start nor end', async () => {
+        const page = await fetchPage(appToken, {
+            subject: SUBJECT,
+            from: 'middle',
+        });
+        expect(page.status).toBe(400);
+
+        const start = await fetchPage(appToken, {
+            subject: SUBJECT,
+            from: 'start',
+        });
+        expect(start.status).toBe(200);
+    });
+
+    it('starts the SDK feed without catch-up in one read', async () => {
+        await clearMailboxes();
+        for (let i = 0; i < 3; i++) await message(`before ${i}`);
+
+        const appActor = (await env.server.services.auth.authenticate(appToken))
+            .actor!;
+        const reads: Array<Record<string, unknown>> = [];
+        let receive: (delivery: { event: unknown }) => void = () => {};
+        const titles: string[] = [];
+        const feed = new NotificationFeed(
+            {
+                authToken: appToken,
+                events: {
+                    onLocal: async (
+                        _subject: string,
+                        handler: typeof receive,
+                    ) => {
+                        receive = handler;
+                        return { off: async () => {} };
+                    },
+                    fetch: async (options: Record<string, unknown>) => {
+                        reads.push(options);
+                        return env.server.services.events.fetchMissed(
+                            appActor,
+                            options,
+                        );
+                    },
+                },
+            } as never,
+            (notification: { title: string | null }) => {
+                titles.push(String(notification.title));
+            },
+            { catchUp: false },
+        );
+
+        try {
+            await feed.start();
+            expect(reads).toEqual([
+                { subject: SUBJECT, limit: 200, from: 'end' },
+            ]);
+
+            await message('after start');
+            receive({ event: { op: 'gap', reason: 'reconnect' } });
+            await vi.waitFor(() => expect(reads).toHaveLength(2));
+            await new Promise((resolve) => setTimeout(resolve, 0));
+        } finally {
+            await feed.off();
+        }
+
+        expect(reads[1]).toEqual({
+            subject: SUBJECT,
+            limit: 200,
+            cursor: expect.any(String),
+        });
+        expect(titles).toEqual(['after start']);
+    });
+});
+
 describe('cross-user isolation', () => {
     it('returns nothing of another account, whatever the subject names', async () => {
         await clearMailboxes();

src/backend/stores/notification/NotificationStore.js

@@ -315,6 +315,57 @@ export class NotificationStore extends PuterStore {
         return rows.map((r) => this.#normalizeRow(r));
     }
 
+    /**
+     * The newest row id in a mailbox slice, or null when it has none, from the
+     * primary. Scoped like `listScoped`; rows of every type count.
+     *
+     * @param {number} userId
+     * @param {{ audience: string; appUid?: string | null }} scope
+     * @returns {Promise<number | null>}
+     */
+    async latestScopedId(userId, { audience, appUid }) {
+        const { where, params } = sliceWhere(userId, {
+            audience,
+            appUid,
+            excludeTypes: [],
+        });
+        const [row] = await this.clients.db.pread(
+            `SELECT \`id\` FROM \`notification\`
+             WHERE ${where.join(' AND ')}
+             ORDER BY \`id\` DESC
+             LIMIT 1`,
+            params,
+        );
+        return row ? Number(row.id) : null;
+    }
+
+    /**
+     * The latest replacement in a mailbox slice by (`replaced_at`, `id`), or
+     * null when none was replaced, from the primary. Scoped like `listScoped`.
+     *
+     * @param {number} userId
+     * @param {{ audience: string; appUid?: string | null }} scope
+     * @returns {Promise<{ replacedAt: number; id: number } | null>}
+     */
+    async latestScopedReplacement(userId, { audience, appUid }) {
+        const { where, params } = sliceWhere(userId, {
+            audience,
+            appUid,
+            excludeTypes: [],
+        });
+        where.push('`replaced_at` IS NOT NULL');
+        const [row] = await this.clients.db.pread(
+            `SELECT \`replaced_at\`, \`id\` FROM \`notification\`
+             WHERE ${where.join(' AND ')}
+             ORDER BY \`replaced_at\` DESC, \`id\` DESC
+             LIMIT 1`,
+            params,
+        );
+        return row
+            ? { replacedAt: Number(row.replaced_at), id: Number(row.id) }
+            : null;
+    }
+
     // -- Writes -------------------------------------------------------
 
     /**

src/backend/stores/notification/NotificationStore.test.js

@@ -410,6 +410,61 @@ describe('NotificationStore', () => {
         expect(rest.map((r) => r.uid)).toEqual([c.uid]);
     });
 
+    it('finds the newest id of a slice, or null for an empty one', async () => {
+        const u = await makeUser();
+        const appUid = `app-${uuidv4()}`;
+        const write = (audience, inApp) =>
+            store.create({ userId: u.id, value: {}, audience, appUid: inApp });
+        const latest = (audience, inApp) =>
+            store.latestScopedId(u.id, { audience, appUid: inApp });
+
+        expect(await latest('app-user', appUid)).toBeNull();
+
+        await write('app-user', appUid);
+        const newest = await write('app-user', appUid);
+        await write('app-user', `app-${uuidv4()}`);
+        const unattributed = await write('developer', null);
+        const developer = await write('developer', appUid);
+        await store.create({
+            userId: other.id,
+            value: {},
+            audience: 'app-user',
+            appUid,
+        });
+
+        expect(await latest('app-user', appUid)).toBe(Number(newest.id));
+        expect(await latest('developer', null)).toBe(Number(unattributed.id));
+        expect(await latest('developer', undefined)).toBe(Number(developer.id));
+        expect(await latest('account', undefined)).toBeNull();
+    });
+
+    it('finds the latest replacement of a slice, or null when none', async () => {
+        const u = await makeUser();
+        const appUid = `app-${uuidv4()}`;
+        const write = (inApp = appUid) =>
+            store.create({
+                userId: u.id,
+                value: {},
+                audience: 'app-user',
+                appUid: inApp,
+            });
+        const latest = () =>
+            store.latestScopedReplacement(u.id, {
+                audience: 'app-user',
+                appUid,
+            });
+
+        const a = await write();
+        const b = await write();
+        const otherApp = await write(`app-${uuidv4()}`);
+        expect(await latest()).toBeNull();
+
+        await store.updateValue(a.uid, u.id, {}, { replacedAt: 2_000 });
+        await store.updateValue(b.uid, u.id, {}, { replacedAt: 2_000 });
+        await store.updateValue(otherApp.uid, u.id, {}, { replacedAt: 9_000 });
+        expect(await latest()).toEqual({ replacedAt: 2_000, id: Number(b.id) });
+    });
+
     // -- mutations -----------------------------------------------------
 
     it('acknowledges only once and only for the owning user', async () => {

src/backend/types.ts

@@ -256,12 +256,7 @@ export interface IPreludeConfig {
      * an RCS agent provisioned in the Prelude account to actually use RCS.
      */
     preferredChannel?:
-        | 'sms'
-        | 'rcs'
-        | 'whatsapp'
-        | 'viber'
-        | 'zalo'
-        | 'telegram';
+        'sms' | 'rcs' | 'whatsapp' | 'viber' | 'zalo' | 'telegram';
 }
 
 /**
@@ -1391,7 +1386,8 @@ export interface ServerDrainInfo {
 }
 
 export interface WithCostsReporting extends WithLifecycle {
-    getReportedCosts?: () => // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    getReportedCosts?: () =>
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
         | Promise<Record<string, any>[]>
         // eslint-disable-next-line @typescript-eslint/no-explicit-any
         | Record<string, any>[];

src/docs/src/Events/fetch.md

@@ -28,6 +28,7 @@ puter.events.fetch(options)
 - `subject` (String) (required): What to read. Use `notif:app-user` for your app's notifications for the user, and `notif:developer` for notifications about your app, for its owner. The full form is `notif:<appId>:<audience>`, where the audience is `account`, `developer`, or `app-user`. Apps and websites never see `account` notifications, so `notif:account` returns an empty page for them.
 - `cursor` (String): The `cursor` from the previous page. Leave it off to start from the oldest notification still kept. A read from a cursor returns the notifications created after it, plus any notification [replaced](/Notifications/notify/#groups) since that page was read, even one from an earlier page.
 - `after` (String): The older name for `cursor`, still accepted. If both are given, `cursor` is used.
+- `from` (String): Where a read without a cursor starts. `'start'`, the default, starts from the oldest notification still kept. `'end'` skips what's already there: it returns no items and a `cursor` at the newest notification, and reads from that cursor return only notifications created or replaced after it. Ignored when a cursor is given.
 - `limit` (Number): Events per page. Defaults to 50, max 200.
 
 ## Return value
@@ -35,7 +36,7 @@ puter.events.fetch(options)
 A `Promise` for `{ items, cursor }`:
 
 - `items`: The events, oldest first, in the same shape as live deliveries.
-- `cursor`: Pass it back as `cursor` to read the next page. It's absent on the last page. Treat it as opaque.
+- `cursor`: Pass it back as `cursor` to read the next page. It's absent on the last page, but always present on a `from: 'end'` read of a part of the mailbox you can see. Treat it as opaque.
 
 Each item is a notification event:
 
@@ -79,7 +80,7 @@ The promise rejects with `{ message, code }`:
 | --- | --- |
 | `invalid_subject` | `subject` is missing or doesn't parse. |
 | `invalid_subject_audience` | A `notif:` subject names an audience other than `account`, `developer`, or `app-user`. |
-| `invalid_request` | `limit` isn't a number, or `cursor` or `after` isn't a string. |
+| `invalid_request` | `limit` isn't a number, `cursor` or `after` isn't a string, or `from` isn't `'start'` or `'end'`. |
 | `fetch_unsupported_subject` | The subject is `fs:` or `kv:`, which store nothing to read. |
 | `auth_canceled` | Nobody was signed in, and the user closed the sign-in without finishing it. |
 | `too_many_requests` | Over the `fetch()` rate limit. |

src/puter-js/src/modules/events/fetch.js

@@ -12,7 +12,8 @@ import { request } from './lib/api.js';
  * for you, and the same call from anywhere returns the same answer. Keep the
  * `cursor` a page comes back with and pass it back as `cursor` to continue; a
  * page with no `cursor` is the end of what there is. `after` is the older name
- * for the same option and still works.
+ * for the same option and still works. `from: 'end'` with no cursor skips what
+ * is already there: an empty page with a cursor at the newest event.
  *
  * Only subjects with a store behind them can answer. Today that is `notif:` —
  * the notification mailbox — and any other family is refused rather than
@@ -47,12 +48,19 @@ export async function fetch (options) {
     if ( opts.after !== undefined && typeof opts.after !== 'string' ) {
         throw new PuterJSError('`after` must be a string', 'invalid_request');
     }
+    if ( opts.from !== undefined && opts.from !== 'start' && opts.from !== 'end' ) {
+        throw new PuterJSError(
+            '`from` must be `start` or `end`',
+            'invalid_request',
+        );
+    }
     const cursor = opts.cursor ?? opts.after;
 
     // Sent as `after`, which every server version reads.
     const page = await request(this.puter, '/events/fetch', undefined, {
         subject,
         ...(cursor !== undefined ? { after: cursor } : {}),
+        ...(opts.from !== undefined ? { from: opts.from } : {}),
         ...(opts.limit !== undefined ? { limit: opts.limit } : {}),
     });
 

src/puter-js/src/modules/events/types.js

@@ -110,6 +110,10 @@
  *   returns notifications replaced since, even ones from earlier pages.
  * @property {string} [after] The older name for `cursor`, still accepted.
  *   `cursor` wins when both are given.
+ * @property {'start' | 'end'} [from] Where a read with no cursor starts:
+ *   `'start'` (the default) at the oldest notification still kept, `'end'` past
+ *   the newest, answering an empty page whose `cursor` returns only what comes
+ *   after it. Ignored when a cursor is given.
  * @property {number} [limit] Events per page. Capped at 200; defaults to 50.
  */
 

src/puter-js/src/modules/notifications/lib/feed.js

@@ -192,12 +192,21 @@ export class NotificationFeed {
             let ok = false;
             try {
                 let ended = false;
+                // A server that ignores `from` answers with the oldest page.
+                let fromEnd = source === 'baseline' && !this.#cursor;
                 for (let page = 0; page < MAX_PAGES && !this.#stopped; page++) {
                     const result = await this.#puter.events.fetch({
                         subject: SUBJECT,
                         limit: PAGE_SIZE,
                         ...(this.#cursor ? { cursor: this.#cursor } : {}),
+                        ...(fromEnd ? { from: 'end' } : {}),
                     });
+                    if (fromEnd && result.items.length === 0 && result.cursor) {
+                        this.#cursor = result.cursor;
+                        ended = true;
+                        break;
+                    }
+                    fromEnd = false;
                     for (const event of result.items) {
                         this.#deliver(event, source);
                     }

src/puter-js/src/modules/notifications/notifications.test.js

@@ -601,6 +601,65 @@ describe('onNotification', () => {
         expect(seen).toEqual([['n-1', 'two']]);
     });
 
+    it('with catchUp: false, starts from the newest notification in one read', async () => {
+        const events = makeEvents({
+            pages: [{ items: [], cursor: 'end-1' }, { items: [event('n-9')] }],
+        });
+        const notifications = makeModule({ events });
+        const seen = [];
+        await notifications.onNotification((n) => seen.push(n.uid), {
+            catchUp: false,
+        });
+
+        expect(events.state.fetches).toEqual([
+            { subject: 'notif:app-user', limit: 200, from: 'end' },
+        ]);
+
+        events.state.push({ op: 'gap', reason: 'reconnect' });
+        await flush();
+
+        expect(events.state.fetches.at(-1)).toEqual({
+            subject: 'notif:app-user',
+            limit: 200,
+            cursor: 'end-1',
+        });
+        expect(seen).toEqual(['n-9']);
+    });
+
+    it('with catchUp: false, marks what is there on a server that ignores from', async () => {
+        const events = makeEvents({
+            pages: [
+                { items: [event('n-1')], cursor: 'c-1' },
+                { items: [event('n-2')] },
+                { items: [event('n-2'), event('n-3')] },
+            ],
+        });
+        const notifications = makeModule({ events });
+        const seen = [];
+        await notifications.onNotification((n) => seen.push(n.uid), {
+            catchUp: false,
+        });
+        events.state.push({ op: 'gap', reason: 'reconnect' });
+        await flush();
+
+        expect(events.state.fetches).toEqual([
+            { subject: 'notif:app-user', limit: 200, from: 'end' },
+            { subject: 'notif:app-user', limit: 200, cursor: 'c-1' },
+            { subject: 'notif:app-user', limit: 200, cursor: 'c-1' },
+        ]);
+        expect(seen).toEqual(['n-3']);
+    });
+
+    it('catching up reads from the oldest notification, not the end', async () => {
+        const events = makeEvents();
+        const notifications = makeModule({ events });
+        await notifications.onNotification(() => {});
+
+        expect(events.state.fetches).toEqual([
+            { subject: 'notif:app-user', limit: 200 },
+        ]);
+    });
+
     it('reads what it missed after a reconnect, from the last cursor', async () => {
         const events = makeEvents({
             pages: [

src/puter-js/src/modules/notifications/webPush.test.js

@@ -915,6 +915,60 @@ describe('onPush', () => {
         });
     });
 
+    it('starts its feed from the newest notification in one read', async () => {
+        const fetches = [];
+        let receive = () => {};
+        const pages = [
+            { items: [], cursor: 'end-1' },
+            {
+                items: [
+                    {
+                        id: 'n2',
+                        op: 'post',
+                        uid: 'n2',
+                        type: 'app.message',
+                        notification: { title: 'Hi', channel: 'chat' },
+                        ts: 1,
+                    },
+                ],
+            },
+        ];
+        const events = {
+            onLocal: vi.fn(async (_subject, handler) => {
+                receive = handler;
+                return { off: async () => {} };
+            }),
+            fetch: vi.fn(async (options) => {
+                fetches.push(options);
+                return pages.shift() ?? { items: [] };
+            }),
+        };
+        const module = new NotificationsModule({
+            ...puter,
+            events,
+            getAppIDFromAuthToken: () => 'app-1',
+            onAuthStateChanged: () => () => {},
+        });
+        const chat = vi.fn();
+
+        const listener = await module.onPush('chat', chat);
+        await vi.waitFor(() => expect(fetches).toHaveLength(1));
+        expect(fetches[0]).toEqual({
+            subject: 'notif:app-user',
+            limit: 200,
+            from: 'end',
+        });
+
+        receive({ event: { op: 'gap', reason: 'reconnect' } });
+        await vi.waitFor(() => expect(chat).toHaveBeenCalledTimes(1));
+        expect(fetches).toEqual([
+            { subject: 'notif:app-user', limit: 200, from: 'end' },
+            { subject: 'notif:app-user', limit: 200, cursor: 'end-1' },
+        ]);
+        expect(chat.mock.calls[0][0]).toMatchObject({ uid: 'n2' });
+        await listener.off();
+    });
+
     it('saves the handler for the worker and runs it for its channel only', async () => {
         const { module, feed } = makeModule();
         const chat = vi.fn();

src/puter-js/tests/api/suites/events.suite.ts

@@ -1457,6 +1457,40 @@ export default suite('events', {
         );
     },
 
+    'starts from the newest notification with `from: end`': async (t) => {
+        await postNotification(t, unique('before-end'));
+
+        const end = await t.puter.events.fetch({
+            subject: 'notif:account',
+            from: 'end',
+        });
+        t.assert.equal(end.items.length, 0);
+        t.assert.ok(end.cursor, 'an end read comes back with a cursor');
+
+        const title = unique('after-end');
+        await postNotification(t, title);
+        const next = await t.puter.events.fetch({
+            subject: 'notif:account',
+            cursor: end.cursor,
+        });
+        t.assert.deepEqual(
+            next.items.map(
+                (event) => (event.notification as { title?: string }).title,
+            ),
+            [title],
+        );
+    },
+
+    'rejects a `from` that is neither start nor end': async (t) => {
+        const error = await t.assert.rejects(() =>
+            t.puter.events.fetch({
+                subject: 'notif:account',
+                from: 'middle' as 'end',
+            }),
+        );
+        t.assert.equal(codeOf(error), 'invalid_request');
+    },
+
     'refuses to publish into an app this account does not own': async (t) => {
         const error = await t.assert.rejects(() =>
             t.puter.events.handlers.publish('ingestUpload', HANDLER, {