diff --git a/src/puter-js/index.d.ts b/src/puter-js/index.d.ts index ca01b2764..586407af1 100644 --- a/src/puter-js/index.d.ts +++ b/src/puter-js/index.d.ts @@ -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'; diff --git a/src/puter-js/src/lib/networkUtils.js b/src/puter-js/src/lib/networkUtils.js index f5921116d..6e020aae8 100644 --- a/src/puter-js/src/lib/networkUtils.js +++ b/src/puter-js/src/lib/networkUtils.js @@ -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; } diff --git a/src/puter-js/src/lib/networkUtils.test.js b/src/puter-js/src/lib/networkUtils.test.js index b4ae28580..e05b263f5 100644 --- a/src/puter-js/src/lib/networkUtils.test.js +++ b/src/puter-js/src/lib/networkUtils.test.js @@ -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: {} })), diff --git a/src/puter-js/src/modules/events/index.js b/src/puter-js/src/modules/events/index.js index 138e2ff8a..393b637c3 100644 --- a/src/puter-js/src/modules/events/index.js +++ b/src/puter-js/src/modules/events/index.js @@ -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} */ onPersistent = onPersistent; + /** @type {OmitThisParameter} */ unsubscribe = unsubscribe; update = update; list = list; fetch = fetch; + /** @type {OmitThisParameter} */ + onSchedule = onSchedule; + /** @type {OmitThisParameter} */ + listSchedules = listSchedules; + /** @type {OmitThisParameter} */ + getSchedule = getSchedule; + /** @type {OmitThisParameter} */ + 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); } diff --git a/src/puter-js/src/modules/events/schedules.js b/src/puter-js/src/modules/events/schedules.js new file mode 100644 index 000000000..baa5893a3 --- /dev/null +++ b/src/puter-js/src/modules/events/schedules.js @@ -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} 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 Promise }>} 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} */ + 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} + */ +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} + */ +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} + */ +export async function removeSchedule (scheduleId) { + assertScheduleId(scheduleId); + await request(this.puter, '/events/schedules/remove', { scheduleId }); +} diff --git a/src/puter-js/src/modules/events/schedules.test.js b/src/puter-js/src/modules/events/schedules.test.js new file mode 100644 index 000000000..020082bfa --- /dev/null +++ b/src/puter-js/src/modules/events/schedules.test.js @@ -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(); + }); +}); diff --git a/src/puter-js/src/modules/events/types.js b/src/puter-js/src/modules/events/types.js index b4b84e9c7..0b264182e 100644 --- a/src/puter-js/src/modules/events/types.js +++ b/src/puter-js/src/modules/events/types.js @@ -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:`. + * @property {'tick'} op Always `'tick'`. + * @property {string} id `:`, 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>} [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} [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} [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} [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. diff --git a/src/puter-js/src/modules/notifications/cancel.js b/src/puter-js/src/modules/notifications/cancel.js new file mode 100644 index 000000000..7bf58e250 --- /dev/null +++ b/src/puter-js/src/modules/notifications/cancel.js @@ -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} + */ +export async function cancel(scheduleId) { + await this.puter.events.removeSchedule(scheduleId); +} diff --git a/src/puter-js/src/modules/notifications/index.js b/src/puter-js/src/modules/notifications/index.js index 258552a22..b62c8cc51 100644 --- a/src/puter-js/src/modules/notifications/index.js +++ b/src/puter-js/src/modules/notifications/index.js @@ -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} */ + 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', diff --git a/src/puter-js/src/modules/notifications/notifications.test.js b/src/puter-js/src/modules/notifications/notifications.test.js index 645427c89..22f21471d 100644 --- a/src/puter-js/src/modules/notifications/notifications.test.js +++ b/src/puter-js/src/modules/notifications/notifications.test.js @@ -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')", () => { diff --git a/src/puter-js/src/modules/notifications/notify.js b/src/puter-js/src/modules/notifications/notify.js index 06163d40b..cee4203ee 100644 --- a/src/puter-js/src/modules/notifications/notify.js +++ b/src/puter-js/src/modules/notifications/notify.js @@ -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} + * @param {( + * | import('./types.js').NotificationInput + * | import('./types.js').ShareNotificationInput + * ) & + * Partial} 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') { diff --git a/src/puter-js/src/modules/notifications/types.js b/src/puter-js/src/modules/notifications/types.js index 633dcf7c1..e41f1ab1e 100644 --- a/src/puter-js/src/modules/notifications/types.js +++ b/src/puter-js/src/modules/notifications/types.js @@ -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. * diff --git a/src/puter-js/tests/api/suites/events.suite.ts b/src/puter-js/tests/api/suites/events.suite.ts index d6f1b181d..0ef984122 100644 --- a/src/puter-js/tests/api/suites/events.suite.ts +++ b/src/puter-js/tests/api/suites/events.suite.ts @@ -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]> = [ + [ + '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]> = [ + [{ 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[0], + ), + ); + t.assert.equal(codeOf(error), code, JSON.stringify(options)); + } + }, }); diff --git a/src/puter-js/tests/api/suites/notifications.suite.ts b/src/puter-js/tests/api/suites/notifications.suite.ts index b942a5ea2..0321c8a65 100644 --- a/src/puter-js/tests/api/suites/notifications.suite.ts +++ b/src/puter-js/tests/api/suites/notifications.suite.ts @@ -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) => { + 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( diff --git a/src/puter-js/tests/types/events.ts b/src/puter-js/tests/types/events.ts new file mode 100644 index 000000000..4ea632001 --- /dev/null +++ b/src/puter-js/tests/types/events.ts @@ -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 => { + 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(); +}; diff --git a/tools/checkPuterjsTypes.mjs b/tools/checkPuterjsTypes.mjs index f85f3b519..fb210b087 100644 --- a/tools/checkPuterjsTypes.mjs +++ b/tools/checkPuterjsTypes.mjs @@ -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.