feat(puter.js): schedules, notify({ sendAt }) and notifications.cancel(), and treat a 503 events_disabled as final

This commit is contained in:
Daniel Salazar committed 2026-11-01 20:30:56 -07:00
1 parent 35d53dbbcc
commit d360c2c350
16 files changed
+986 -6

No files matched your search

+9
View File
@@ -139,11 +139,18 @@ export type {
HandlerSummary,
OnLocalOptions,
OnPersistentOptions,
OnScheduleOptions,
PersistentSubscription,
PublishedHandler,
PuterEvent,
PuterKvEvent,
PuterNotifEvent,
PuterScheduleEvent,
Schedule,
ScheduleDelivery,
ScheduleErrorCode,
ScheduleHandler,
ScheduleListOptions,
UpdatePersistentOptions,
} from './types/modules/events/types.js';
export type { EventSubscription } from './types/modules/events/lib/subscription.js';
@@ -236,6 +243,7 @@ export type { PSocket, PTLSSocket } from './types/modules/networking/PSocket.js'
// -- puter.notifications --
export type {
AppNotification,
DelayedNotificationOptions,
NotificationClick,
NotificationHandler,
NotificationInput,
@@ -249,6 +257,7 @@ export type {
PushPlatform,
PushShown,
PushStatus,
ScheduledNotification,
ShareNotificationInput,
} from './types/modules/notifications/types.js';
+23 -1
View File
@@ -504,6 +504,16 @@ async function bodyForLog(xhr) {
// can mean the request was half-applied upstream, so only a read replays.
const RETRYABLE_STATUS = new Set([502, 503, 504]);
// A 503 that names a feature this server doesn't have is final: asking again
// gets the same answer.
const FINAL_UNAVAILABLE_CODES = new Set(['events_disabled']);
const isFinalUnavailable = (status, parsed) =>
status === 503 &&
[parsed?.code, parsed?.error?.code].some((code) =>
FINAL_UNAVAILABLE_CODES.has(code),
);
// A rate/concurrency gate rejects in middleware, before the handler that would
// have done the work — so nothing happened, and replaying is safe for a write
// exactly as it is for a read. Treating this like the statuses above meant a
@@ -844,7 +854,19 @@ async function classifyRetry(outcome, ctx) {
}
// transient status — read-safe only, honors kill switch, fixed schedule.
if (RETRYABLE_STATUS.has(status)) return transientRetry(ctx);
if (RETRYABLE_STATUS.has(status)) {
if (status === 503 && ctx.retrySafe) {
if (outcome.parsed === undefined) {
try {
outcome.parsed = await bodyJson(xhr);
} catch {
outcome.parsed = null;
}
}
if (isFinalUnavailable(status, outcome.parsed)) return null;
}
return transientRetry(ctx);
}
return null;
}
+41
View File
@@ -613,6 +613,47 @@ describe('transient retry', () => {
vi.useRealTimers();
});
// A server without a feature answers 503 `events_disabled` every time.
it.each([
[{ code: 'events_disabled', message: 'Schedules are not available' }],
[{ error: { code: 'events_disabled', message: 'Not available' } }],
])(
'answers a GET whose 503 is a structured events_disabled at once (%j)',
async (body) => {
const xhrs = installFakeXHR(
sequence(respond({ status: 503, body })),
);
const resp = await fetchUrl('https://api.example/x'); // GET
expect(resp.status).toBe(503);
expect(xhrs.length).toBe(1);
expect(JSON.stringify(await resp.json())).toContain(
'events_disabled',
);
},
);
it.each([
[{ status: 503, body: { code: 'service_unavailable' } }],
[
{
status: 503,
body: 'upstream unavailable',
contentType: 'text/plain',
},
],
[{ status: 502, body: { code: 'events_disabled' } }],
])('keeps retrying any other 502/503 (%j)', async (first) => {
vi.useFakeTimers();
const xhrs = installFakeXHR(
sequence(respond(first), respond({ status: 200, body: { ok: 1 } })),
);
const p = fetchUrl('https://api.example/x');
await vi.advanceTimersByTimeAsync(60_000);
expect((await p).status).toBe(200);
expect(xhrs.length).toBe(2);
vi.useRealTimers();
});
it('retry:false disables retry even for a GET', async () => {
const xhrs = installFakeXHR(
sequence(respond({ status: 503, body: {} })),
+24
View File
@@ -6,6 +6,12 @@ import { fetch } from './fetch.js';
import { list } from './list.js';
import { onLocal } from './onLocal.js';
import { onPersistent } from './onPersistent.js';
import {
getSchedule,
listSchedules,
onSchedule,
removeSchedule,
} from './schedules.js';
import { unsubscribe } from './unsubscribe.js';
import { update } from './update.js';
@@ -24,6 +30,8 @@ import { update } from './update.js';
* `fetch()` is the other half: what happened while nothing was listening, read
* from the subject's own store a page at a time.
*
* `onSchedule()` runs a published handler on a timer rather than on a change.
*
* A published handler set stands up an **events worker** per app —
* `puter.events.workers` is where an owner sees and destroys them.
*
@@ -35,12 +43,24 @@ export class EventsModule extends PuterModule {
// The fields hold the unbound functions so they keep their full types
// (`bind` erases overloads); the constructor rebinds them so destructured
// calls (`const { onLocal } = puter.events`) work.
// A field typed without `this` keeps `puter.events.x()` compiling: its
// `this` names this class, which the public type omits members of.
onLocal = onLocal;
/** @type {OmitThisParameter<typeof onPersistent>} */
onPersistent = onPersistent;
/** @type {OmitThisParameter<typeof unsubscribe>} */
unsubscribe = unsubscribe;
update = update;
list = list;
fetch = fetch;
/** @type {OmitThisParameter<typeof onSchedule>} */
onSchedule = onSchedule;
/** @type {OmitThisParameter<typeof listSchedules>} */
listSchedules = listSchedules;
/** @type {OmitThisParameter<typeof getSchedule>} */
getSchedule = getSchedule;
/** @type {OmitThisParameter<typeof removeSchedule>} */
removeSchedule = removeSchedule;
/** @param {Puter} puter */
constructor (puter) {
@@ -68,6 +88,10 @@ export class EventsModule extends PuterModule {
'update',
'list',
'fetch',
'onSchedule',
'listSchedules',
'getSchedule',
'removeSchedule',
] ) {
methods[name] = methods[name].bind(this);
}
@@ -0,0 +1,176 @@
import { PuterJSError } from '../../lib/PuterJSError.js';
import { request } from './lib/api.js';
import { prepareHandler, serializeContext } from './lib/handlerSource.js';
/** @typedef {import('./types.js').OnScheduleOptions} OnScheduleOptions */
/** @typedef {import('./types.js').Schedule} Schedule */
/** @typedef {import('./types.js').ScheduleListOptions} ScheduleListOptions */
/** @typedef {import('../../lib/types.js').ListPage<Schedule>} SchedulePage */
/** The options sent as they are; the server validates and rounds them. */
const TIMING_OPTIONS = /** @type {const} */ ([
'every',
'cron',
'tz',
'at',
'startAt',
'expiresAt',
'maxRuns',
]);
const noSuchSchedule = () =>
new PuterJSError('No such schedule', 'events_schedule_not_found');
/** @param {unknown} scheduleId */
const assertScheduleId = (scheduleId) => {
if ( typeof scheduleId !== 'string' || scheduleId.trim().length === 0 ) {
throw noSuchSchedule();
}
};
/**
* Runs the app's published handler on a timer: every interval, on a cron
* expression in a time zone, or once at a set time. The schedule is stored
* against the account and runs with nobody present, in the app's events
* worker, never in this client.
*
* Give exactly one of `every`, `cron` or `at`. `handlerName` names the
* published handler each run calls; passing `handler` too checks that it is
* the source published under that name.
*
* `context` is evaluated **here, now** and delivered to every run as a frozen
* `ctx`. It can't be changed later: remove the schedule and create a new one.
*
* Needs an app, and the account's `events:background` permission for it. On a
* website with nobody signed in, opens the sign-in first; rejects
* `auth_canceled` if the visitor closes it.
*
* @this {import('./index.js').EventsModule}
* @param {OnScheduleOptions} options
* @returns {Promise<Schedule & { off: () => Promise<void> }>} The schedule,
* with `off()` to remove it.
*/
export async function onSchedule (options) {
const { puter } = this;
if ( ! options || typeof options !== 'object' ) {
throw new PuterJSError(
'`onSchedule()` takes an options object',
'invalid_request',
);
}
const { handler, handlerName } = options;
if ( handler !== undefined && handler !== null && ! handlerName ) {
throw new PuterJSError(
'An inline `handler` needs a `handlerName` to check it against',
'events_handler_name_required',
);
}
if ( ! handlerName ) {
throw new PuterJSError(
'A schedule needs a `handlerName`',
'events_handler_required',
);
}
for ( const key of ['at', 'startAt', 'expiresAt'] ) {
const value = options[key];
if ( value instanceof Date && ! Number.isFinite(value.getTime()) ) {
throw new PuterJSError(
`\`${key}\` is not a valid date`,
'events_schedule_invalid',
);
}
}
const inline = handler === undefined || handler === null
? null
: await prepareHandler(puter, handler);
const handlerHash = inline?.hash ?? options.handlerHash;
/** @type {Record<string, unknown>} */
const body = { handlerName };
for ( const key of TIMING_OPTIONS ) {
if ( options[key] !== undefined && options[key] !== null )
body[key] = options[key];
}
if ( handlerHash !== undefined && handlerHash !== null )
body.handlerHash = handlerHash;
// Serialized only to check it against the cap before the round trip.
if ( serializeContext(options.context) !== undefined )
body.context = options.context;
const schedule = /** @type {Schedule} */ (
await request(puter, '/events/schedules', body)
);
const off = async () => this.removeSchedule(schedule.scheduleId);
return Object.assign(schedule, { off });
}
/**
* Lists the schedules this caller holds, one page at a time. Ended schedules
* stay listed for 7 days with why they ended.
*
* An app sees only the schedules it created. A session acting for the account
* sees them all, including ones left behind by an app it no longer uses.
* `context` values are never returned; a schedule reports its key names and a
* hash instead.
*
* On a website with nobody signed in, opens the sign-in first; rejects
* `auth_canceled` if the visitor closes it.
*
* @this {import('./index.js').EventsModule}
* @param {ScheduleListOptions} [options]
* @returns {Promise<SchedulePage>}
*/
export async function listSchedules (options = {}) {
const { limit, cursor, includeTotal } = options ?? {};
if ( includeTotal !== undefined && typeof includeTotal !== 'boolean' ) {
throw new PuterJSError(
'`includeTotal` must be a boolean',
'invalid_request',
);
}
return /** @type {SchedulePage} */ (
/** @type {unknown} */ (
await request(this.puter, '/events/schedules', undefined, {
...(limit !== undefined ? { limit } : {}),
...(cursor ? { cursor } : {}),
...(includeTotal ? { includeTotal: true } : {}),
})
)
);
}
/**
* Reads one schedule, ended or not.
*
* An id this caller does not hold reads as absent, the same as one that never
* existed: `events_schedule_not_found`.
*
* @this {import('./index.js').EventsModule}
* @param {string} scheduleId The `scheduleId` of the schedule to read.
* @returns {Promise<Schedule>}
*/
export async function getSchedule (scheduleId) {
assertScheduleId(scheduleId);
return /** @type {Schedule} */ (
await request(
this.puter,
`/events/schedules/${encodeURIComponent(scheduleId)}`,
)
);
}
/**
* Ends a schedule.
*
* An id this caller does not hold, or one already ended, reads as absent:
* `events_schedule_not_found`.
*
* @this {import('./index.js').EventsModule}
* @param {string} scheduleId The `scheduleId` of the schedule to end.
* @returns {Promise<void>}
*/
export async function removeSchedule (scheduleId) {
assertScheduleId(scheduleId);
await request(this.puter, '/events/schedules/remove', { scheduleId });
}
@@ -0,0 +1,245 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { PuterJSError } from '../../lib/PuterJSError.js';
// Every schedule verb goes through the one HTTP helper, so mocking it needs no
// server and still exercises the real request bodies.
const mockRequest = vi.fn();
vi.mock('./lib/api.js', () => ({
request: (...args) => mockRequest(...args),
}));
const { getSchedule, listSchedules, onSchedule, removeSchedule } =
await import('./schedules.js');
const HANDLER = ({ event, ctx }) => fetch(ctx.url, { body: event.id });
const SCHEDULE_ID = '2f7c1b6e-4d1a-4c55-9a8e-0e6f3c2b1a90';
const makeModule = () => {
const module = {
puter: { APIOrigin: 'https://api.test', fs: { read: vi.fn() } },
onSchedule,
listSchedules,
getSchedule,
removeSchedule,
};
for ( const name of ['onSchedule', 'listSchedules', 'getSchedule', 'removeSchedule'] )
module[name] = module[name].bind(module);
return module;
};
const routeOf = (index = 0) => mockRequest.mock.calls[index][1];
const bodyOf = (index = 0) => mockRequest.mock.calls[index][2];
const queryOf = (index = 0) => mockRequest.mock.calls[index][3];
const rejects = async (run) => {
try {
await run();
} catch ( error ) {
return error;
}
throw new Error('expected a rejection');
};
beforeEach(() => {
mockRequest.mockReset();
mockRequest.mockResolvedValue({});
});
describe('onSchedule', () => {
it('sends the timing options and the handler name, and returns the server`s schedule', async () => {
const view = { scheduleId: SCHEDULE_ID, kind: 'recurring', nextRunAt: 1 };
mockRequest.mockResolvedValue(view);
const schedule = await makeModule().onSchedule({
cron: '0 9 * * 1-5',
tz: 'America/New_York',
startAt: '2026-11-02',
expiresAt: 1798761600000,
maxRuns: 10,
handlerName: 'digest',
unknownOption: true,
});
expect(routeOf()).toBe('/events/schedules');
expect(bodyOf()).toEqual({
cron: '0 9 * * 1-5',
tz: 'America/New_York',
startAt: '2026-11-02',
expiresAt: 1798761600000,
maxRuns: 10,
handlerName: 'digest',
});
expect(schedule).toBe(view);
expect(schedule.nextRunAt).toBe(1);
});
it('passes `every`, `at` and a `Date` through for the server to check', async () => {
const at = new Date('2026-11-01T18:30:00Z');
await makeModule().onSchedule({ every: '6h', handlerName: 'poll' });
await makeModule().onSchedule({ at, handlerName: 'remind' });
expect(bodyOf(0)).toEqual({ every: '6h', handlerName: 'poll' });
expect(bodyOf(1)).toEqual({ at, handlerName: 'remind' });
expect(JSON.parse(JSON.stringify(bodyOf(1))).at).toBe(
'2026-11-01T18:30:00.000Z',
);
});
it('rejects an invalid Date for `at`, `startAt` or `expiresAt` before any request', async () => {
for ( const key of ['at', 'startAt', 'expiresAt'] ) {
const error = await rejects(() =>
makeModule().onSchedule({
...(key === 'at' ? {} : { every: '1h' }),
[key]: new Date(NaN),
handlerName: 'tick',
handler: HANDLER,
}));
expect(error).toBeInstanceOf(PuterJSError);
expect(error.code).toBe('events_schedule_invalid');
expect(error.message).toContain(key);
}
expect(mockRequest).not.toHaveBeenCalled();
});
it('hands back a handle whose off() removes the schedule', async () => {
const module = makeModule();
mockRequest.mockResolvedValueOnce({ scheduleId: SCHEDULE_ID });
const schedule = await module.onSchedule({ every: '1h', handlerName: 'poll' });
await schedule.off();
expect(routeOf(1)).toBe('/events/schedules/remove');
expect(bodyOf(1)).toEqual({ scheduleId: SCHEDULE_ID });
});
it('sends an inline handler as a hash, never as source', async () => {
const { hashSource } = await import('./lib/handlerSource.js');
await makeModule().onSchedule({
every: '1h',
handlerName: 'poll',
handler: HANDLER,
});
expect(bodyOf().handlerHash).toBe(
await hashSource(Function.prototype.toString.call(HANDLER)),
);
expect(JSON.stringify(bodyOf())).not.toContain('fetch(');
});
it('passes a handlerHash given in place of the handler', async () => {
const handlerHash = 'a'.repeat(64);
await makeModule().onSchedule({ every: '1h', handlerName: 'poll', handlerHash });
expect(bodyOf().handlerHash).toBe(handlerHash);
});
it('forwards any handlerHash it is given, an empty one included, for the server to judge', async () => {
await makeModule().onSchedule({ every: '1h', handlerName: 'poll', handlerHash: '' });
await makeModule().onSchedule({ every: '1h', handlerName: 'poll', handlerHash: null });
expect(bodyOf(0)).toHaveProperty('handlerHash', '');
expect(bodyOf(1)).not.toHaveProperty('handlerHash');
});
it('needs a handler name before it calls anything', async () => {
const unnamed = await rejects(() => makeModule().onSchedule({ every: '1h' }));
expect(unnamed.code).toBe('events_handler_required');
const inline = await rejects(() =>
makeModule().onSchedule({ every: '1h', handler: HANDLER }),
);
expect(inline.code).toBe('events_handler_name_required');
const none = await rejects(() => makeModule().onSchedule());
expect(none.code).toBe('invalid_request');
expect(mockRequest).not.toHaveBeenCalled();
});
it('refuses a context over the cap before the round trip', async () => {
const error = await rejects(() =>
makeModule().onSchedule({
every: '1h',
handlerName: 'poll',
context: { blob: 'x'.repeat(5000) },
}),
);
expect(error.code).toBe('events_context_too_large');
expect(mockRequest).not.toHaveBeenCalled();
});
it('sends the context it was given', async () => {
await makeModule().onSchedule({
every: '1h',
handlerName: 'poll',
context: { team: 'design' },
});
expect(bodyOf().context).toEqual({ team: 'design' });
});
it('passes a server refusal through unchanged, a disabled server included', async () => {
for ( const code of ['events_disabled', 'events_schedule_invalid', 'events_schedule_busy'] ) {
const refusal = new PuterJSError('refused', code);
mockRequest.mockRejectedValueOnce(refusal);
const error = await rejects(() =>
makeModule().onSchedule({ every: '1h', handlerName: 'poll' }),
);
expect(error).toBe(refusal);
}
});
});
describe('listSchedules', () => {
it('asks for one page, with only the paging options given', async () => {
const page = { items: [], cursor: 'next' };
mockRequest.mockResolvedValue(page);
const result = await makeModule().listSchedules({
limit: 2,
cursor: 'abc',
includeTotal: true,
});
await makeModule().listSchedules();
await makeModule().listSchedules({ cursor: null, includeTotal: false });
expect(result).toBe(page);
expect(routeOf(0)).toBe('/events/schedules');
expect(bodyOf(0)).toBeUndefined();
expect(queryOf(0)).toEqual({ limit: 2, cursor: 'abc', includeTotal: true });
expect(queryOf(1)).toEqual({});
expect(queryOf(2)).toEqual({});
});
it('refuses an includeTotal that is not a boolean', async () => {
const error = await rejects(() =>
makeModule().listSchedules({ includeTotal: 'yes' }),
);
expect(error.code).toBe('invalid_request');
expect(mockRequest).not.toHaveBeenCalled();
});
});
describe('getSchedule and removeSchedule', () => {
it('name the schedule in the path and the body', async () => {
await makeModule().getSchedule(SCHEDULE_ID);
await makeModule().removeSchedule(SCHEDULE_ID);
expect(routeOf(0)).toBe(`/events/schedules/${SCHEDULE_ID}`);
expect(bodyOf(0)).toBeUndefined();
expect(routeOf(1)).toBe('/events/schedules/remove');
expect(bodyOf(1)).toEqual({ scheduleId: SCHEDULE_ID });
});
it('keep an id inside its path segment', async () => {
await makeModule().getSchedule('../subscriptions');
expect(routeOf()).toBe('/events/schedules/..%2Fsubscriptions');
});
it('answer an id that is not one as absent, without a round trip', async () => {
for ( const id of [undefined, null, '', ' ', 42] ) {
for ( const verb of ['getSchedule', 'removeSchedule'] ) {
const error = await rejects(() => makeModule()[verb](id));
expect(error.code).toBe('events_schedule_not_found');
}
}
expect(mockRequest).not.toHaveBeenCalled();
});
});
+143
View File
@@ -286,6 +286,149 @@
* `onPersistent()` returns, not on one a listing reports.
*/
/**
* What a handler a schedule runs gets as `event`. Runs are at least once: a
* retry, or a run that outlives its timeout, can run one slot twice, with the
* same `id`. Skip work already done for an `id`.
*
* @typedef {Object} PuterScheduleEvent
* @property {string} subject `schedule:<scheduleId>`.
* @property {'tick'} op Always `'tick'`.
* @property {string} id `<scheduleId>:<scheduledFor>`, the same on every
* attempt at one run.
* @property {string} scheduleId The schedule this run belongs to.
* @property {number} scheduledFor When this run was due, in milliseconds since
* the epoch.
* @property {number} firedAt When this attempt was sent to the handler, in
* milliseconds since the epoch.
* @property {number} missed Runs skipped since the last one that ran, such as
* while the schedule was suspended. Only this run happens for them.
* @property {true} [missedCapped] Present when `missed` stopped counting at
* its cap of 1,000.
* @property {number} attempt Attempts at this run, this one included.
*/
/**
* What a handler a schedule runs is called with, in the app's events worker.
*
* @typedef {Object} ScheduleDelivery
* @property {PuterScheduleEvent} event The run.
* @property {Readonly<Record<string, unknown>>} [ctx] The `context` the
* schedule was created with, frozen.
* @property {unknown} [user] A `puter` bound to the account holding the
* schedule.
* @property {(input: unknown, init?: unknown) => Promise<unknown>} [fetch]
* `puter.net.fetch` where it exists, the environment's `fetch` otherwise.
*/
/**
* A handler a schedule runs. Its return value is ignored.
*
* @typedef {(delivery: ScheduleDelivery) => unknown} ScheduleHandler
*/
/**
* Options for {@link import('./schedules.js').onSchedule}. Give exactly one
* of `every`, `cron` or `at`. Times are an ISO-8601 string with a zone (or a
* plain date, read as UTC midnight), a `Date`, or milliseconds since the
* epoch, and are rounded up to the next whole second.
*
* @typedef {Object} OnScheduleOptions
* @property {number | string} [every] Run at this interval: milliseconds, or
* a duration like `'30m'`, `'6h'` or `'1d'` (`s`, `m`, `h`, `d`, `w`). A
* whole number of seconds, at least the plan's minimum, at most 366 days.
* @property {string} [cron] Run on a 5-field cron expression, or a nickname
* such as `'@daily'`. Standard fields only.
* @property {string} [tz] The IANA time zone `cron` runs in. Defaults to
* `'UTC'`. Only with `cron`.
* @property {number | string | Date} [at] Run once, at this time. At least 1
* second and at most 366 days ahead.
* @property {number | string | Date} [startAt] For `every`, the first run;
* without it, the first run is one interval after create. A past `startAt`
* aligns the runs to it, starting at the next one. For `cron`, the earliest
* run.
* @property {number | string | Date} [expiresAt] The last time the schedule may
* run, inclusive.
* @property {number} [maxRuns] Stop after this many runs (1 to 100,000). With
* `expiresAt` too, the earlier end wins.
* @property {string} handlerName The published handler each run calls.
* @property {ScheduleHandler | string | { file: string }} [handler] The handler
* source this schedule was written against. Sent as a hash: the schedule is
* refused unless it matches what is published under `handlerName`. It never
* runs in this client.
* @property {string} [handlerHash] The SHA-256 hex digest of that source, in
* place of `handler`.
* @property {Record<string, unknown>} [context] Values the handler needs,
* evaluated **now** and delivered as a frozen `ctx` on every run. Capped at
* 4 KB serialized, and can't be changed after create.
*/
/**
* A schedule, as its holder sees it. Times are milliseconds since the epoch.
* `context` values are never returned, only which keys it sets and a hash.
*
* @typedef {Object} Schedule
* @property {string} scheduleId The schedule's id, and what
* `removeSchedule()` and `getSchedule()` name.
* @property {'recurring' | 'once' | 'deferred'} kind `recurring` for `every`
* and `cron`, `once` for `at`, `deferred` for a delayed notification.
* @property {string} appUid The app that created it.
* @property {string | null} handlerName The handler each run calls, or `null`
* once the schedule has ended.
* @property {string[] | null} contextKeys Key names of its context, or `null`
* when it has none or has ended.
* @property {string | null} contextHash Hash of its context.
* @property {number | null} every The interval of an `every` schedule, in
* milliseconds.
* @property {string | null} cron The cron expression of a `cron` schedule.
* @property {string} tz The time zone `cron` runs in; `'UTC'` otherwise.
* @property {number} startAt The first allowed run. For `once`, its only run.
* @property {number | null} endsAt The last allowed run, inclusive, from
* `expiresAt` or `maxRuns`. For `once`, its only run.
* @property {'active' | 'suspended' | 'sending' | 'ended'} state `suspended`
* while it doesn't run but can resume, `sending` once a delayed
* notification has started to go (it can't be cancelled then), `ended` once
* it never runs again.
* @property {string | null} suspendedReason Why it is suspended:
* `handler_not_found`, `failures` or `no_credit`.
* @property {string | null} endReason Why it ended: `removed`, `ended` (its
* last run is done), `app_deleted`, `uninstalled`, `permission_revoked`,
* `holder_gone`, `app_gone`, `orphaned`, `delivered` or `dropped`.
* @property {number | null} endedAt When it ended.
* @property {number | null} nextRunAt The next slot of its definition, or
* `null` while suspended or sending, and once ended. A run starts within about a minute
* of it; a late or retried run for an earlier slot may fire first.
* @property {number} createdAt When it was created.
* @property {() => Promise<void>} [off] Removes the schedule. Present on the
* schedule `onSchedule()` returns, not on one a listing reports.
*/
/**
* Options for {@link import('./schedules.js').listSchedules}.
*
* @typedef {Object} ScheduleListOptions
* @property {number} [limit] Schedules per page. Defaults to 50, capped at 200.
* @property {string | null} [cursor] The `cursor` from the previous page.
* Absent or `null` starts from the first page.
* @property {boolean} [includeTotal] Adds `total`, the count across every page.
*/
/**
* Error codes the schedule methods reject with, besides the shared events
* codes such as `events_disabled` and `too_many_requests`.
*
* @typedef {'events_schedule_invalid'
* | 'events_schedule_never_fires'
* | 'events_schedule_too_frequent'
* | 'events_schedule_immutable'
* | 'events_schedule_app_required'
* | 'events_schedule_limit'
* | 'events_schedule_budget'
* | 'events_schedule_not_found'
* | 'events_schedule_sending'
* | 'events_schedule_busy'} ScheduleErrorCode
*/
/**
* Where a handler operation applies. An app token publishes into its own app
* and needs neither field; an account session has to name an app it owns.
@@ -0,0 +1,13 @@
/**
* Cancels a notification `notify()` stored to send later. One that has started
* sending and isn't finished yet can't be cancelled: `events_schedule_sending`.
* One already sent or cancelled, or an id this app doesn't hold, reads as
* `events_schedule_not_found`.
*
* @this {import('./index.js').NotificationsModule}
* @param {string} scheduleId The `scheduleId` `notify()` resolved with.
* @returns {Promise<void>}
*/
export async function cancel(scheduleId) {
await this.puter.events.removeSchedule(scheduleId);
}
@@ -1,6 +1,7 @@
import EventListener from '../../lib/EventListener.js';
import { PuterJSError } from '../../lib/PuterJSError.js';
import { needsSignIn } from '../events/lib/signIn.js';
import { cancel } from './cancel.js';
import { disable } from './disable.js';
import { enable } from './enable.js';
import { enablePush } from './enablePush.js';
@@ -151,6 +152,8 @@ const showThroughWorker = (registration, title, options) => {
export class NotificationsModule extends EventListener {
// Rebound in the constructor so destructured calls keep `this`.
notify = notify;
/** @type {OmitThisParameter<typeof cancel>} */
cancel = cancel;
onNotification = onNotification;
enable = enable;
disable = disable;
@@ -213,6 +216,7 @@ export class NotificationsModule extends EventListener {
);
for (const name of [
'notify',
'cancel',
'onNotification',
'enable',
'disable',
@@ -221,6 +221,81 @@ describe('notify', () => {
const error = await rejects(() => makeModule().notify({ title: 'Hi' }));
expect(error.code).toBe('notifications_failed');
});
it('stores a send for later and resolves with its schedule', async () => {
mockDriverCall.mockResolvedValue({
scheduled: { scheduleId: 'sched-1', sendAt: 1_800_000_000_000 },
});
const result = await makeModule().notify({
title: 'Hi',
sendAt: new Date(1_800_000_000_000),
});
expect(result).toEqual({
scheduleId: 'sched-1',
sendAt: 1_800_000_000_000,
});
expect(mockDriverCall.mock.calls[0][0].args.object.value.sendAt).toBe(
'2027-01-15T08:00:00.000Z',
);
});
it('resolves with the schedule of a later send through a share too', async () => {
mockDriverCall.mockResolvedValue({
scheduled: { scheduleId: 'sched-2', sendAt: 5_000 },
});
const result = await makeModule().notify({
title: 'Hi',
to: { handle: 'kvh-1' },
sendAt: 5_000,
});
expect(result).toEqual({ scheduleId: 'sched-2', sendAt: 5_000 });
});
it('rejects an invalid Date before calling the server', async () => {
const error = await rejects(() =>
makeModule().notify({ title: 'Hi', sendAt: new Date(NaN) }),
);
expect(error).toBeInstanceOf(PuterJSError);
expect(error.code).toBe('events_schedule_invalid');
expect(mockDriverCall).not.toHaveBeenCalled();
});
it('resolves as an immediate send when the server sent it at once', async () => {
mockDriverCall.mockResolvedValue({ uid: 'n-3' });
expect(
await makeModule().notify({ title: 'Hi', sendAt: Date.now() }),
).toBe('n-3');
});
});
describe('cancel', () => {
it('removes the stored send through the schedule surface', async () => {
const removeSchedule = vi.fn(async () => {});
const notifications = makeModule({
events: { ...makeEvents(), removeSchedule },
});
const { cancel } = notifications;
await expect(cancel('sched-1')).resolves.toBeUndefined();
expect(removeSchedule).toHaveBeenCalledWith('sched-1');
});
it('passes a refusal through', async () => {
const refusal = new PuterJSError(
'This delayed send has already started',
'events_schedule_sending',
);
const notifications = makeModule({
events: {
...makeEvents(),
removeSchedule: vi.fn(async () => {
throw refusal;
}),
},
});
expect(await rejects(() => notifications.cancel('sched-1'))).toBe(
refusal,
);
});
});
describe("on('click')", () => {
@@ -9,6 +9,29 @@ const toError = (error) => {
return PuterJSError.from(error);
};
/**
* What the server answers a create with: the written notification, or the send
* it stored for later.
*
* @typedef {{
* uid?: unknown;
* scheduled?: import('./types.js').ScheduledNotification;
* }} CreateAnswer
*/
/**
* @overload
* @param {(
* | import('./types.js').NotificationInput
* | import('./types.js').ShareNotificationInput
* ) &
* import('./types.js').DelayedNotificationOptions} notification
* @returns {Promise<
* import('./types.js').ScheduledNotification | string | void
* >}
* The stored send. When `sendAt` is within 5 seconds or past, the notification
* is sent at once and this resolves as it would without it.
*/
/**
* @overload
* @param {import('./types.js').NotificationInput & { to?: undefined }} notification
@@ -23,12 +46,18 @@ const toError = (error) => {
*/
/**
* Sends the user this app runs for a notification, or with `to` the other party
* of a KV share. Only an app can send one.
* of a KV share. Only an app can send one. With `sendAt` it is stored and sent
* then; `cancel()` takes it back before it goes.
*
* @this {import('./index.js').NotificationsModule}
* @param {import('./types.js').NotificationInput
* | import('./types.js').ShareNotificationInput} notification
* @returns {Promise<string | void>}
* @param {(
* | import('./types.js').NotificationInput
* | import('./types.js').ShareNotificationInput
* ) &
* Partial<import('./types.js').DelayedNotificationOptions>} notification
* @returns {Promise<
* import('./types.js').ScheduledNotification | string | void
* >}
*/
export async function notify(notification) {
if (
@@ -54,10 +83,19 @@ export async function notify(notification) {
if (field !== undefined) value[key] = field;
}
value.type = APP_MESSAGE_TYPE;
if (value.sendAt instanceof Date) {
if (!Number.isFinite(value.sendAt.getTime())) {
throw new PuterJSError(
'`sendAt` is not a valid date',
'events_schedule_invalid',
);
}
value.sendAt = value.sendAt.toISOString();
}
let created;
try {
created = /** @type {{ uid?: unknown } | null} */ (
created = /** @type {CreateAnswer | null} */ (
await driverCall({
puter: this.puter,
iface: 'puter-notifications',
@@ -70,6 +108,10 @@ export async function notify(notification) {
throw toError(error);
}
if (created?.scheduled) {
const { scheduleId, sendAt } = created.scheduled;
return { scheduleId, sendAt };
}
// A notification to someone else answers with nothing about it.
if (value.to !== undefined) return;
if (typeof created?.uid !== 'string') {
@@ -31,6 +31,25 @@
* @typedef {NotificationInput & { to: NotificationRecipient }} ShareNotificationInput
*/
/**
* Sends a notification later instead of now.
*
* @typedef {Object} DelayedNotificationOptions
* @property {number | string | Date} sendAt When to send it: an ISO-8601 time
* with a zone (or a plain date, read as UTC midnight), a `Date`, or
* milliseconds since the epoch, rounded up to the next whole second. At most
* 366 days ahead. Within 5 seconds, or in the past, it is sent at once.
*/
/**
* A notification stored to be sent later.
*
* @typedef {Object} ScheduledNotification
* @property {string} scheduleId What `cancel()` and `puter.events.getSchedule()`
* name.
* @property {number} sendAt When it goes, in milliseconds since the epoch.
*/
/**
* Who a notification goes to, when it isn't the user the app runs for.
*
@@ -1342,4 +1342,76 @@ export default suite('events', {
);
t.assert.equal(codeOf(error), 'events_handler_app_required');
},
// -- Schedules ---------------------------------------------------
//
// This server has no schedules service, as a self-hosted one doesn't:
// every schedule call answers `events_disabled`.
'exposes the schedule surface': async (t) => {
for (const method of [
'onSchedule',
'listSchedules',
'getSchedule',
'removeSchedule',
] as const) {
t.assert.equal(
typeof t.puter.events[method],
'function',
`puter.events.${method} is a function`,
);
}
},
'answers every schedule call events_disabled without schedules': async (
t,
) => {
const appUid = await makeApp(t);
const scheduleId = '2f7c1b6e-4d1a-4c55-9a8e-0e6f3c2b1a90';
const calls: Array<[string, () => Promise<unknown>]> = [
[
'onSchedule',
() =>
t.puter.events.onSchedule({
every: '6h',
handlerName: 'digest',
}),
],
['listSchedules', () => t.puter.events.listSchedules()],
['getSchedule', () => t.puter.events.getSchedule(scheduleId)],
['removeSchedule', () => t.puter.events.removeSchedule(scheduleId)],
];
for (const [name, call] of calls) {
// Final, not replayed as a transient 503.
const started = Date.now();
const asTheApp = await asApp(t, appUid, () =>
t.assert.rejects(call),
);
t.assert.equal(codeOf(asTheApp), 'events_disabled', name);
t.assert.ok(
Date.now() - started < 2_000,
`${name} answers at once`,
);
const asTheAccount = await t.assert.rejects(call);
t.assert.equal(codeOf(asTheAccount), 'events_disabled', name);
}
},
'checks schedule options in the client before asking': async (t) => {
const cases: Array<[Record<string, unknown>, string]> = [
[{ every: '1h' }, 'events_handler_required'],
[
{ at: new Date(Number.NaN), handlerName: 'digest' },
'events_schedule_invalid',
],
];
for (const [options, code] of cases) {
const error = await t.assert.rejects(() =>
t.puter.events.onSchedule(
options as Parameters<typeof t.puter.events.onSchedule>[0],
),
);
t.assert.equal(codeOf(error), code, JSON.stringify(options));
}
},
});
@@ -85,6 +85,7 @@ export default suite('notifications', {
'status',
'listDevices',
'removeDevice',
'cancel',
] as const) {
t.assert.equal(
typeof t.puter.notifications[method],
@@ -384,6 +385,58 @@ export default suite('notifications', {
);
},
'answers sendAt and cancel events_disabled without schedules, and sends without sendAt':
async (t) => {
const appUid = await makeApp(t);
const { later, now, cancelled, invalid, sent } = await asApp(
t,
appUid,
async () => {
const refusal = async (run: () => Promise<unknown>) => {
try {
await run();
} catch (error) {
return error;
}
throw new Error('expected a refusal');
};
return {
later: await refusal(() =>
t.puter.notifications.notify({
title: unique('later'),
sendAt: new Date(Date.now() + 60 * 60 * 1000),
}),
),
now: await refusal(() =>
t.puter.notifications.notify({
title: unique('now'),
sendAt: Date.now(),
}),
),
cancelled: await refusal(() =>
t.puter.notifications.cancel(
'2f7c1b6e-4d1a-4c55-9a8e-0e6f3c2b1a90',
),
),
invalid: await refusal(() =>
t.puter.notifications.notify({
title: unique('never'),
sendAt: new Date(Number.NaN),
}),
),
sent: await t.puter.notifications.notify({
title: unique('plain'),
}),
};
},
);
t.assert.equal(codeOf(later), 'events_disabled');
t.assert.equal(codeOf(now), 'events_disabled');
t.assert.equal(codeOf(cancelled), 'events_disabled');
t.assert.equal(codeOf(invalid), 'events_schedule_invalid');
t.assert.equal(typeof sent, 'string');
},
'status resolves one of its four states without asking': async (t) => {
const state = await t.puter.notifications.status();
t.assert.ok(
+37
View File
@@ -0,0 +1,37 @@
// Compiled by `npm run check:puterjs:types` against the generated declarations;
// never run. A line marked `@ts-expect-error` must fail to compile.
import type {
EventHandler,
OnScheduleOptions,
Puter,
ScheduleHandler,
} from '../../index.js';
// A subscription handler reads the fields every subscription event carries.
export const persistent: EventHandler = ({ event }) => event.ts;
export const scheduled: ScheduleHandler = ({ event, ctx }) =>
`${event.id} ${event.scheduledFor} ${event.attempt} ${String(ctx?.team)}`;
// @ts-expect-error A schedule run has no node to name.
export const noPath: ScheduleHandler = ({ event }) => event.path;
export const options: OnScheduleOptions = {
every: '6h',
startAt: new Date(),
handlerName: 'digest',
handler: scheduled,
};
// The schedule verbs, unsubscribing, and cancelling a stored notification,
// called on `puter`.
export const onPuter = async (puter: Puter): Promise<void> => {
const schedule = await puter.events.onSchedule(options);
await puter.events.getSchedule(schedule.scheduleId);
const page = await puter.events.listSchedules({ limit: 10 });
await puter.events.removeSchedule(page.items[0].scheduleId);
await puter.notifications.cancel(schedule.scheduleId);
await puter.events.unsubscribe('sub-id');
// The handle onSchedule() returns always has off().
await schedule.off();
};
+5
View File
@@ -18,6 +18,7 @@
*/
import { execFileSync } from 'node:child_process';
import { readdirSync } from 'node:fs';
import { dirname, join, sep } from 'node:path';
import { fileURLToPath } from 'node:url';
@@ -52,6 +53,10 @@ const checked = tsc([
'--module', 'nodenext',
'--moduleResolution', 'nodenext',
join(SDK, 'index.d.ts'),
// Usage the declarations must keep accepting (and refusing).
...readdirSync(join(SDK, 'tests', 'types'))
.filter((name) => name.endsWith('.ts'))
.map((name) => join(SDK, 'tests', 'types', name)),
]);
// Errors inside third-party `@types` packages are not this project's to fix.