;
@@ -1056,10 +1055,7 @@ export type MatchingEvents = 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 {
diff --git a/src/backend/controllers/events/EventsController.ts b/src/backend/controllers/events/EventsController.ts
index 13ab2d9aa..8248dadfe 100644
--- a/src/backend/controllers/events/EventsController.ts
+++ b/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 {
const actor = this.#requireActor(req);
const query = (req.query ?? {}) as Record;
+ 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, {
diff --git a/src/backend/services/events/EventsService.ts b/src/backend/services/events/EventsService.ts
index 5b35de43a..786e37e78 100644
--- a/src/backend/services/events/EventsService.ts
+++ b/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> {
+ 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
diff --git a/src/backend/services/events/fetch.integration.test.ts b/src/backend/services/events/fetch.integration.test.ts
index 01c62e475..bf655ecb0 100644
--- a/src/backend/services/events/fetch.integration.test.ts
+++ b/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 =>
+ (
+ 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 => {
+ 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> = [];
+ 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) => {
+ 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();
diff --git a/src/backend/stores/notification/NotificationStore.js b/src/backend/stores/notification/NotificationStore.js
index d8cb86d2a..ae6a872dc 100644
--- a/src/backend/stores/notification/NotificationStore.js
+++ b/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}
+ */
+ 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 -------------------------------------------------------
/**
diff --git a/src/backend/stores/notification/NotificationStore.test.js b/src/backend/stores/notification/NotificationStore.test.js
index afba9b0d6..ba15f64eb 100644
--- a/src/backend/stores/notification/NotificationStore.test.js
+++ b/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 () => {
diff --git a/src/backend/types.ts b/src/backend/types.ts
index 9948fc87b..2edb4cb9a 100644
--- a/src/backend/types.ts
+++ b/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[]>
// eslint-disable-next-line @typescript-eslint/no-explicit-any
| Record[];
diff --git a/src/docs/src/Events/fetch.md b/src/docs/src/Events/fetch.md
index 37b7fc234..40e928764 100644
--- a/src/docs/src/Events/fetch.md
+++ b/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::`, 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. |
diff --git a/src/puter-js/src/modules/events/fetch.js b/src/puter-js/src/modules/events/fetch.js
index 505105f5a..8691c43d1 100644
--- a/src/puter-js/src/modules/events/fetch.js
+++ b/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 } : {}),
});
diff --git a/src/puter-js/src/modules/events/types.js b/src/puter-js/src/modules/events/types.js
index 6e597e74e..99c091b9a 100644
--- a/src/puter-js/src/modules/events/types.js
+++ b/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.
*/
diff --git a/src/puter-js/src/modules/notifications/lib/feed.js b/src/puter-js/src/modules/notifications/lib/feed.js
index 360096c36..8ebb5db3b 100644
--- a/src/puter-js/src/modules/notifications/lib/feed.js
+++ b/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);
}
diff --git a/src/puter-js/src/modules/notifications/notifications.test.js b/src/puter-js/src/modules/notifications/notifications.test.js
index 341f035e0..d02e889fd 100644
--- a/src/puter-js/src/modules/notifications/notifications.test.js
+++ b/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: [
diff --git a/src/puter-js/src/modules/notifications/webPush.test.js b/src/puter-js/src/modules/notifications/webPush.test.js
index ddebd9cf2..3632a3969 100644
--- a/src/puter-js/src/modules/notifications/webPush.test.js
+++ b/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();
diff --git a/src/puter-js/tests/api/suites/events.suite.ts b/src/puter-js/tests/api/suites/events.suite.ts
index 0741232f5..e6a231e40 100644
--- a/src/puter-js/tests/api/suites/events.suite.ts
+++ b/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, {