diff --git a/src/docs/src/Events.md b/src/docs/src/Events.md new file mode 100644 index 000000000..0c1ccc3ea --- /dev/null +++ b/src/docs/src/Events.md @@ -0,0 +1,105 @@ +--- +title: Events +description: Watch a user's files and react to changes as they happen. +platforms: [websites, apps, nodejs, workers] +--- + +
The Events API is in beta. Event shapes, limits, and behavior may change between releases.
+ +The Events API tells your app when something changes. Subscribe to a *subject* — a file, a directory, a path that does not exist yet — and a handler runs every time something under it is created, written, moved, or removed. + +```js +const sub = await puter.events.onLocal('fs:~/Documents', ({ event }) => { + console.log(event.op, event.path); +}); + +// ... later +await sub.off(); +``` + +## Subjects + +A subject names what you are watching, and optionally the one operation you care about: + +``` +fs:[:] +``` + +- **Path** — absolute (`/alice/Documents`) or home-relative (`~/Documents`). Subscribing to a directory covers everything under it, at any depth. +- **Uid** — the `uid` of a file or directory, for watching one specific node no matter where it moves to. +- **Op** — one of `add`, `write`, `move`, `remove`, `meta`. Leave it off to get all of them. Nothing emits `meta` yet, so a subscription limited to it stays quiet. + +```js +await puter.events.onLocal('fs:~/Documents', handler); // everything under Documents +await puter.events.onLocal('fs:~/Documents/notes.txt:write', handler); // one file, writes only +await puter.events.onLocal('fs:~/Pictures/*.png', handler); // one segment of wildcard +await puter.events.onLocal('fs:~/Projects/**/build.log', handler); // across directories +``` + +Only `fs:` subjects can be subscribed to today. + +### Watching something that does not exist yet + +A subject is allowed to name a path that is not there. The subscription anchors on the nearest directory that *does* exist and the rest of the subject becomes a pattern, so the event you get is the one where it appears: + +```js +// Nothing at this path yet — the handler runs when it is created. +await puter.events.onLocal('fs:~/Documents/inbox/trigger.json:add', ({ event }) => { + process(event.path); +}); +``` + +Wildcards work the same way: `*` matches within one path segment, `**` crosses directories, and both cost the same. + +### What you are allowed to watch + +Subscribing takes the same access as reading. A subject you cannot read — and a subject that is not there — both fail with `subject_does_not_exist`, so the call cannot be used to find out which one it was. Access is re-checked on every delivery too: when a share is revoked, deliveries stop immediately. + +## The event + +The handler is called with `{ event }`: + +| Field | Type | Description | +| --- | --- | --- | +| `id` | String | Unique id for the event. | +| `subject` | String | The subject the change was projected onto, naming the node it happened to (`fs::`) — not the subject string you subscribed with. | +| `op` | String | `add`, `write`, `move`, or `remove`. | +| `uid` | String | The uid of the node that changed. | +| `path` | String | The path of the node that changed. | +| `self` | Boolean | `true` when the change was made by the account holding the subscription. Check it to ignore your own writes. | +| `ts` | Number | When it happened, in milliseconds since the epoch. | +| `seq` | Number | Position within one dispatch, for changes that fan out to several subscriptions. | + +Nothing else is included — in particular there is no field naming *who* made the change, because on a shared folder that would tell every subscriber who else is in there. + +### Gaps + +Every per-event limit truncates the delivery rather than failing anything, and sends a **gap marker** in its place: an event with `op: 'gap'`, a `reason`, and no `uid` or `path`. A gap means something happened that you were not told the details of, so treat it as "re-read what I am watching", never as "nothing changed". + +```js +await puter.events.onLocal('fs:~/Documents', async ({ event }) => { + if (event.op === 'gap') return refreshEverything(); + apply(event); +}); +``` + +## Subscriptions live with the connection + +`onLocal()` subscriptions are **session-scoped**: nothing is stored, nothing runs while the page is closed, and the server drops them when the connection goes away. Every subscription this client makes rides one connection, which opens on the first `onLocal()` and closes when the last subscription ends. In a worker that means the subscription lasts as long as the invocation that made it, and no longer. + +When the connection drops and comes back — a reconnect, a sign-in, an API origin change — the SDK subscribes again for you. The handler and the subscription object stay the same; only `subId` changes, which is why nothing should be stored against it. If re-subscribing fails (the access is gone, the account signed out), or the server closes the connection outright (a revoked session, too many connections), the subscription ends and your `onError` callback is told: + +```js +const sub = await puter.events.onLocal('fs:~/Documents', handler, { + onError: (error) => console.warn('subscription ended:', error.code), +}); +``` + +## Limits + +Subscriptions per connection, subscribe calls per minute, and how much one event may fan out are all capped — see [Rate Limits and Quotas](/rate-limits-and-quotas/). Deliveries are coalesced over 250 ms per subject, so a multipart upload or a save loop arrives as one event rather than one per write. + +## Functions + +- **[`puter.events.onLocal()`](/Events/onLocal/)** - Subscribe to a subject for as long as this client is connected +- **[`subscription.off()`](/Events/off/)** - End a subscription diff --git a/src/docs/src/Events/off.md b/src/docs/src/Events/off.md new file mode 100644 index 000000000..0adbb9b40 --- /dev/null +++ b/src/docs/src/Events/off.md @@ -0,0 +1,55 @@ +--- +title: subscription.off() +description: End a subscription created with puter.events.onLocal(). +platforms: [websites, apps, nodejs, workers] +--- + +
The Events API is in beta. Event shapes, limits, and behavior may change between releases.
+ +Ends a subscription returned by [`puter.events.onLocal()`](/Events/onLocal/). The handler stops being called immediately, and the server is told when there is still a connection to tell it over. + +When the last subscription on this client ends, the events connection closes with it. + +## Syntax +```js +subscription.off() +``` + +## Parameters +None. + +## Return value +A `Promise` that resolves when the subscription is gone. It never rejects: calling `off()` twice, or after the connection has already dropped, is a no-op — a subscription does not outlive its connection, so there is nothing left to fail at. + +## Examples + +Watch a directory, then stop watching it + +```html + + + + + + +``` diff --git a/src/docs/src/Events/onLocal.md b/src/docs/src/Events/onLocal.md new file mode 100644 index 000000000..eaaeaa23a --- /dev/null +++ b/src/docs/src/Events/onLocal.md @@ -0,0 +1,121 @@ +--- +title: puter.events.onLocal() +description: Subscribe to changes on a file or directory for as long as this client is connected. +platforms: [websites, apps, nodejs, workers] +--- + +
The Events API is in beta. Event shapes, limits, and behavior may change between releases.
+ +Subscribes to a subject and calls `handler` every time something matching it changes. The subscription belongs to this client's connection: nothing is stored, nothing runs while the page is closed, and it ends when the connection does. See [Events](/Events/) for the subject grammar and the event shape. + +## Syntax +```js +puter.events.onLocal(subject, handler) +puter.events.onLocal(subject, handler, options) +``` + +## Parameters + +#### `subject` (String) (required) +What to watch: `fs:[:]`. The path may be absolute (`/alice/Documents`) or home-relative (`~/Documents`), may name something that does not exist yet, and may contain `*` (within a path segment) or `**` (across directories). The optional `op` is one of `add`, `write`, `move`, `remove`, `meta` — nothing emits `meta` yet. + +#### `handler` (Function) (required) +Called with a single `{ event }` object per delivery. `event.op === 'gap'` means events were dropped against a limit and the details are not available — re-read what you are watching. A handler that throws is reported on the console and does not end the subscription. + +#### `options` (Object) (optional) + +- `onError` (Function): Called with `{ message, code }` if the subscription lapses — the connection was lost and re-subscribing failed. The subscription is over at that point; call `onLocal()` again to resume. Without it, a lapse is reported on the console. +- `timeout` (Number): How long to wait for the server to confirm the subscription, in milliseconds. Defaults to `30000`. + +## Return value + +A `Promise` that resolves, once the server has confirmed the subscription, to a subscription object: + +- `subId` (String | null): The server's id for the subscription. It changes whenever the connection is rebuilt, so don't store anything against it. +- `subject` (String): The subject you subscribed with. +- `anchor` (Object): `{ uid, path }` of the node the subscription is keyed to — the nearest existing ancestor when the subject named something that does not exist yet. +- `match` (String | null): The pattern events under the anchor are matched against, if the subject had one. +- `op` (String | null): The single operation this subscription is limited to, or `null` for all of them. +- `off` (Function): Ends the subscription — see [`subscription.off()`](/Events/off/). + +The promise rejects with `{ message, code }`: + +| `code` | Meaning | +| --- | --- | +| `invalid_subject` | The subject is not a non-empty string, or the server could not parse it. | +| `invalid_handler` | `handler` is not a function. | +| `invalid_subject_op` | The `:op` suffix is not one of the five operations. | +| `invalid_subject_pattern` | The match pattern is past the compile-cost bounds (256 characters, 16 segments). | +| `subject_does_not_exist` | The subject is not there, or this account cannot read it. | +| `events_subscription_limit` | This connection already holds the maximum number of subscriptions. | +| `too_many_requests` | Over the subscribe/unsubscribe call budget. | +| `events_disabled` | Events are not enabled on this server. | +| `reauth_required` | The session backing this connection is no longer valid. | +| `events_connection_failed` | The events connection could not be established, the server did not answer in time, or the server closed the connection. | + +## Examples + +Watch a directory and print what changes + +```html + + + + + + +``` + +React to a file that does not exist yet + +```html + + + + + + +``` diff --git a/src/docs/src/sidebar.js b/src/docs/src/sidebar.js index 53d425fee..11ad84870 100755 --- a/src/docs/src/sidebar.js +++ b/src/docs/src/sidebar.js @@ -400,6 +400,30 @@ let sidebar = [ }, ], }, + { + title: 'Events', + title_tag: 'Events', + source: '/Events.md', + path: '/Events', + children: [ + { + title: 'onLocal()', + page_title: 'puter.events.onLocal()', + title_tag: 'puter.events.onLocal()', + icon: '/assets/img/function.svg', + source: '/Events/onLocal.md', + path: '/Events/onLocal', + }, + { + title: 'off()', + page_title: 'subscription.off()', + title_tag: 'subscription.off()', + icon: '/assets/img/function.svg', + source: '/Events/off.md', + path: '/Events/off', + }, + ], + }, { title: 'Serverless Workers', title_tag: 'Serverless Workers', diff --git a/src/puter-js/index.d.ts b/src/puter-js/index.d.ts index 48df4dae8..3e5cc1270 100644 --- a/src/puter-js/index.d.ts +++ b/src/puter-js/index.d.ts @@ -92,6 +92,17 @@ export type { EmailSendResult, } from './types/modules/Email.js'; +// -- puter.events -- +export type { + EventAnchor, + EventDelivery, + EventGapMarker, + EventHandler, + OnLocalOptions, + PuterEvent, +} from './types/modules/events/types.js'; +export type { EventSubscription } from './types/modules/events/lib/subscription.js'; + // -- puter.fs -- export type { CopyOptions, @@ -233,6 +244,7 @@ export type Apps = InstanceType; export type Drivers = InstanceType; export type Email = InstanceType; +export type Events = InstanceType; export type FS = InstanceType; export type Hosting = InstanceType; export type KV = InstanceType; diff --git a/src/puter-js/src/index.js b/src/puter-js/src/index.js index 7c296faa1..b7c9ffb93 100644 --- a/src/puter-js/src/index.js +++ b/src/puter-js/src/index.js @@ -13,6 +13,7 @@ import Auth from './modules/Auth.js'; import { Debug } from './modules/Debug.js'; import Drivers from './modules/Drivers.js'; import Email from './modules/Email.js'; +import { Events } from './modules/events/index.js'; import { PuterJSFileSystemModule } from './modules/FileSystem/index.js'; import FSItem from './modules/FSItem.js'; import { Hosting } from './modules/hosting/index.js'; @@ -172,6 +173,8 @@ export class Puter { kv; /** @type {InstanceType} */ email; + /** @type {InstanceType} */ + events; /** @type {InstanceType} */ perms; /** @type {InstanceType} */ @@ -298,6 +301,7 @@ export class Puter { this.ai = this.registerModule('ai', AI); this.kv = this.registerModule('kv', KV); this.email = this.registerModule('email', Email); + this.events = this.registerModule('events', Events); this.perms = this.registerModule('perms', Perms); this.drivers = this.registerModule('drivers', Drivers); this.debug = this.registerModule('debug', Debug); diff --git a/src/puter-js/src/lib/socketOptions.js b/src/puter-js/src/lib/socketOptions.js new file mode 100644 index 000000000..d77a2b760 --- /dev/null +++ b/src/puter-js/src/lib/socketOptions.js @@ -0,0 +1,22 @@ +/** @typedef {import('../index.js').Puter} Puter */ + +/** + * Whether socket.io may unref its connection so an idle socket does not keep + * a node process alive. Its autoUnref path expects `ws._socket.unref()` to + * exist, which only the `ws` package provides — Undici's WebSocket has no such + * thing, so asking for it there throws. + * + * @param {Puter} puter + * @returns {boolean} + */ +export const socketAutoUnref = (puter) => { + if ( puter.env !== 'nodejs' ) return false; + + const WebSocketImpl = globalThis.WebSocket; + if ( typeof WebSocketImpl !== 'function' ) return false; + + // ws instances are EventEmitter-like; Undici's are EventTarget-like. + const wsPrototype = /** @type {Record} */ (WebSocketImpl.prototype ?? {}); + return typeof wsPrototype.on === 'function' && + typeof wsPrototype.removeListener === 'function'; +}; diff --git a/src/puter-js/src/modules/FileSystem/index.js b/src/puter-js/src/modules/FileSystem/index.js index 8c17e11a4..572bde136 100644 --- a/src/puter-js/src/modules/FileSystem/index.js +++ b/src/puter-js/src/modules/FileSystem/index.js @@ -1,6 +1,7 @@ import path from 'path-browserify'; import { io } from 'socket.io-client'; import { PuterModule } from '../../lib/PuterModule.js'; +import { socketAutoUnref } from '../../lib/socketOptions.js'; import * as utils from '../../lib/utils.js'; // Constants @@ -113,20 +114,7 @@ export class PuterJSFileSystemModule extends PuterModule { } shouldUseSocketAutoUnref () { - if ( this.puter.env !== 'nodejs' ) { - return false; - } - - const WebSocketImpl = globalThis.WebSocket; - if ( typeof WebSocketImpl !== 'function' ) { - return false; - } - - const wsPrototype = WebSocketImpl.prototype ?? {}; - // ws package instances are EventEmitter-like; Undici WebSocket is EventTarget-like. - // autoUnref is only safe on the ws path. - return typeof wsPrototype.on === 'function' && - typeof wsPrototype.removeListener === 'function'; + return socketAutoUnref(this.puter); } bindSocketEvents () { diff --git a/src/puter-js/src/modules/events/events.test.js b/src/puter-js/src/modules/events/events.test.js new file mode 100644 index 000000000..10433af6e --- /dev/null +++ b/src/puter-js/src/modules/events/events.test.js @@ -0,0 +1,452 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +// One fake connection per `io()` call, so the tests can drive the socket +// lifecycle the server would otherwise drive: acks, drops, reconnects. +const sockets = []; + +class FakeSocket { + constructor (url, options) { + this.url = url; + this.options = options; + this.handlers = new Map(); + this.sent = []; + this.connected = true; + this.active = true; + this.disconnected = false; + sockets.push(this); + } + + on (name, handler) { + const existing = this.handlers.get(name) ?? []; + existing.push(handler); + this.handlers.set(name, existing); + return this; + } + + fire (name, ...args) { + for ( const handler of [...(this.handlers.get(name) ?? [])] ) handler(...args); + } + + emit (verb, payload, ack) { + this.sent.push({ verb, payload, ack }); + return this; + } + + /** Answer the most recent send of `verb`. */ + answer (verb, response) { + const call = [...this.sent].reverse().find(sent => sent.verb === verb); + if ( ! call ) throw new Error(`nothing was sent for ${verb}`); + call.ack(response); + } + + removeAllListeners () { + this.handlers.clear(); + return this; + } + + disconnect () { + this.disconnected = true; + this.connected = false; + return this; + } +} + +vi.mock('socket.io-client', () => ({ + io: (url, options) => new FakeSocket(url, options), +})); + +const { EventsModule } = await import('./index.js'); + +const okSub = (subId, subject) => ({ + ok: true, + sub: { + subId, + subject, + anchor: { uid: 'anchor-uid', path: '/user/Documents' }, + match: null, + op: null, + }, +}); + +const projected = (subId, path) => ({ + subId, + event: { + id: `evt-${path}`, + subject: 'fs:anchor-uid:add', + op: 'add', + uid: 'node-uid', + path, + self: true, + ts: 1, + seq: 0, + }, +}); + +let authStateListeners = []; + +const makeModule = () => { + const puter = { + env: 'web', + authToken: 'token-1', + APIOrigin: 'https://api.test', + onAuthStateChanged: listener => authStateListeners.push(listener), + }; + return new EventsModule(puter); +}; + +/** Subscribe and answer the ack the server would send. */ +const subscribed = async (events, subject, handler, options, subId = 'sub-1') => { + const pending = events.onLocal(subject, handler, options); + await Promise.resolve(); + sockets.at(-1).answer('events.subscribe', okSub(subId, subject)); + return await pending; +}; + +beforeEach(() => { + sockets.length = 0; + authStateListeners = []; +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe('connection lifetime', () => { + it('does not connect until something subscribes', async () => { + const events = makeModule(); + expect(sockets).toHaveLength(0); + + await subscribed(events, 'fs:~/Documents', () => {}); + expect(sockets).toHaveLength(1); + expect(sockets[0].url).toBe('https://api.test'); + expect(sockets[0].options.auth).toEqual({ auth_token: 'token-1' }); + }); + + it('carries several subscriptions on one socket, and closes with the last', async () => { + const events = makeModule(); + const first = await subscribed(events, 'fs:~/a', () => {}, {}, 'sub-a'); + const second = await subscribed(events, 'fs:~/b', () => {}, {}, 'sub-b'); + + expect(sockets).toHaveLength(1); + + const firstOff = first.off(); + sockets[0].answer('events.unsubscribe', { ok: true }); + await firstOff; + expect(sockets[0].disconnected).toBe(false); + + const secondOff = second.off(); + sockets[0].answer('events.unsubscribe', { ok: true }); + await secondOff; + expect(sockets[0].disconnected).toBe(true); + }); + + it('leaves no connection behind when the first subscribe fails', async () => { + const events = makeModule(); + const pending = events.onLocal('fs:~/nope', () => {}); + await Promise.resolve(); + sockets.at(-1).answer('events.subscribe', { + ok: false, + error: { code: 'subject_does_not_exist', message: 'No such entry' }, + }); + + await expect(pending).rejects.toMatchObject({ + code: 'subject_does_not_exist', + message: 'No such entry', + }); + expect(sockets[0].disconnected).toBe(true); + }); +}); + +describe('delivery routing', () => { + it('routes an event to the subscription it names', async () => { + const events = makeModule(); + const mine = []; + const theirs = []; + await subscribed(events, 'fs:~/a', ({ event }) => mine.push(event), {}, 'sub-a'); + await subscribed(events, 'fs:~/b', ({ event }) => theirs.push(event), {}, 'sub-b'); + + sockets[0].fire('events.delivery', projected('sub-a', '/user/a/one.txt')); + + expect(mine).toHaveLength(1); + expect(mine[0].path).toBe('/user/a/one.txt'); + expect(theirs).toEqual([]); + }); + + it('ignores an event for a subscription that was already ended', async () => { + const events = makeModule(); + const seen = []; + const sub = await subscribed(events, 'fs:~/a', ({ event }) => seen.push(event)); + + const off = sub.off(); + sockets[0].answer('events.unsubscribe', { ok: true }); + await off; + + sockets[0].fire('events.delivery', projected('sub-1', '/user/a/late.txt')); + expect(seen).toEqual([]); + }); + + it('keeps delivering after a handler throws', async () => { + const events = makeModule(); + const errors = vi.spyOn(console, 'error').mockImplementation(() => {}); + let calls = 0; + await subscribed(events, 'fs:~/a', () => { + calls++; + throw new Error('handler bug'); + }); + + sockets[0].fire('events.delivery', projected('sub-1', '/user/a/one.txt')); + sockets[0].fire('events.delivery', projected('sub-1', '/user/a/two.txt')); + + expect(calls).toBe(2); + expect(errors).toHaveBeenCalled(); + }); +}); + +describe('reconnect', () => { + it('re-subscribes on reconnect and keeps the same handle', async () => { + const events = makeModule(); + const seen = []; + const sub = await subscribed(events, 'fs:~/a', ({ event }) => seen.push(event)); + + sockets[0].fire('disconnect'); + expect(sub.subId).toBe(null); + + sockets[0].fire('connect'); + sockets[0].answer('events.subscribe', okSub('sub-2', 'fs:~/a')); + await Promise.resolve(); + + expect(sub.subId).toBe('sub-2'); + sockets[0].fire('events.delivery', projected('sub-2', '/user/a/after.txt')); + expect(seen).toHaveLength(1); + }); + + it('ends the subscription and reports it when re-subscribing fails', async () => { + const events = makeModule(); + const lapses = []; + const seen = []; + const sub = await subscribed( + events, + 'fs:~/a', + ({ event }) => seen.push(event), + { onError: error => lapses.push(error) }, + ); + + sockets[0].fire('disconnect'); + sockets[0].fire('connect'); + sockets[0].answer('events.subscribe', { + ok: false, + error: { code: 'subject_does_not_exist', message: 'gone' }, + }); + await Promise.resolve(); + await Promise.resolve(); + + expect(lapses).toHaveLength(1); + expect(lapses[0].code).toBe('subject_does_not_exist'); + expect(sub.subId).toBe(null); + // Nothing routes to a lapsed subscription. + sockets[0].fire('events.delivery', projected('sub-1', '/user/a/x.txt')); + expect(seen).toEqual([]); + }); + + it('keeps the subscription when the connection dies mid-resubscribe', async () => { + const events = makeModule(); + const lapses = []; + const sub = await subscribed(events, 'fs:~/a', () => {}, { + onError: error => lapses.push(error), + }); + + sockets[0].fire('disconnect'); + sockets[0].fire('connect'); + // The re-subscribe is in flight when the connection drops again. + sockets[0].fire('disconnect'); + await Promise.resolve(); + await Promise.resolve(); + + expect(lapses).toEqual([]); + expect(sub.subId).toBe(null); + + sockets[0].fire('connect'); + sockets[0].answer('events.subscribe', okSub('sub-3', 'fs:~/a')); + await Promise.resolve(); + expect(sub.subId).toBe('sub-3'); + }); + + it('rebuilds the connection when auth state changes', async () => { + const events = makeModule(); + await subscribed(events, 'fs:~/a', () => {}); + + events.puter.authToken = 'token-2'; + for ( const listener of authStateListeners ) listener(); + + expect(sockets).toHaveLength(2); + expect(sockets[0].disconnected).toBe(true); + expect(sockets[1].options.auth).toEqual({ auth_token: 'token-2' }); + }); + + it('off() after the connection dropped resolves without asking the server', async () => { + const events = makeModule(); + const sub = await subscribed(events, 'fs:~/a', () => {}); + + sockets[0].fire('disconnect'); + sockets[0].connected = false; + + await expect(sub.off()).resolves.toBeUndefined(); + expect(sockets[0].sent.filter(s => s.verb === 'events.unsubscribe')).toHaveLength(0); + }); + + it('drops a subscription off()\'d mid-resubscribe, without misrouting its id to the survivor', async () => { + const events = makeModule(); + const survivorSeen = []; + const droppedSeen = []; + + const survivor = await subscribed(events, 'fs:~/a', ({ event }) => survivorSeen.push(event), {}, 'sub-a'); + const dropped = await subscribed(events, 'fs:~/b', ({ event }) => droppedSeen.push(event), {}, 'sub-b'); + + sockets[0].fire('disconnect'); + sockets[0].fire('connect'); + // Both resubscribes are in flight now, oldest first — grab each send + // directly, since `answer()` only ever reaches the most recent one. + const [survivorResend, droppedResend] = sockets[0].sent + .filter(s => s.verb === 'events.subscribe') + .slice(-2); + + // End `dropped` before its resubscribe ack comes back. + const off = dropped.off(); + // The server had already minted a new id for it by the time the ack + // arrives — nothing points at it any more. + droppedResend.ack(okSub('sub-b2', 'fs:~/b')); + await off; + await Promise.resolve(); + await Promise.resolve(); + + // The orphaned id is handed straight back to the server... + expect( + sockets[0].sent.filter(s => s.verb === 'events.unsubscribe' && s.payload.subId === 'sub-b2'), + ).toHaveLength(1); + + // ...and the survivor's own remap is unaffected. + survivorResend.ack(okSub('sub-a2', 'fs:~/a')); + await Promise.resolve(); + expect(survivor.subId).toBe('sub-a2'); + + // A stray delivery on the dropped id must be dropped, not misrouted + // to whichever handler happens to be listening. + sockets[0].fire('events.delivery', projected('sub-b2', '/user/b/late.txt')); + sockets[0].fire('events.delivery', projected('sub-a2', '/user/a/one.txt')); + + expect(droppedSeen).toEqual([]); + expect(survivorSeen).toHaveLength(1); + expect(survivorSeen[0].path).toBe('/user/a/one.txt'); + }); + + it('ends every subscription when the server closes the connection', async () => { + const events = makeModule(); + const lapses = []; + const sub = await subscribed(events, 'fs:~/a', () => {}, { + onError: error => lapses.push(error), + }); + + // A server-side disconnect is final: socket.io will not reconnect it. + sockets[0].active = false; + sockets[0].fire('disconnect', 'io server disconnect'); + + expect(lapses).toHaveLength(1); + expect(lapses[0].code).toBe('events_connection_failed'); + expect(sub.subId).toBe(null); + expect(sockets[0].disconnected).toBe(true); + + // The next subscribe starts over on a fresh connection. + await subscribed(events, 'fs:~/b', () => {}, {}, 'sub-2'); + expect(sockets).toHaveLength(2); + }); + + it('closes the connection when the last subscription lapses', async () => { + const events = makeModule(); + await subscribed(events, 'fs:~/a', () => {}, { onError: () => {} }); + + sockets[0].fire('disconnect'); + sockets[0].fire('connect'); + sockets[0].answer('events.subscribe', { + ok: false, + error: { code: 'subject_does_not_exist', message: 'gone' }, + }); + await Promise.resolve(); + await Promise.resolve(); + + expect(sockets[0].disconnected).toBe(true); + expect(events.channel.socket).toBe(null); + }); + + it('retries a re-subscribe that was turned away for rate limiting', async () => { + vi.useFakeTimers(); + try { + const events = makeModule(); + const lapses = []; + const sub = await subscribed(events, 'fs:~/a', () => {}, { + onError: error => lapses.push(error), + }); + + sockets[0].fire('disconnect'); + sockets[0].fire('connect'); + sockets[0].answer('events.subscribe', { + ok: false, + error: { code: 'too_many_requests', message: 'slow down' }, + }); + await Promise.resolve(); + await Promise.resolve(); + + expect(lapses).toEqual([]); + expect(sub.subId).toBe(null); + const sent = () => sockets[0].sent.filter(s => s.verb === 'events.subscribe').length; + const before = sent(); + + await vi.advanceTimersByTimeAsync(10000); + expect(sent()).toBe(before + 1); + sockets[0].answer('events.subscribe', okSub('sub-2', 'fs:~/a')); + await Promise.resolve(); + expect(sub.subId).toBe('sub-2'); + } finally { + vi.useRealTimers(); + } + }); +}); + +describe('ack timeout', () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it('rejects a subscribe after 30s with no ack, by default', async () => { + vi.useFakeTimers(); + const events = makeModule(); + + const pending = events.onLocal('fs:~/a', () => {}); + const assertion = expect(pending).rejects.toMatchObject({ code: 'events_connection_failed' }); + await vi.advanceTimersByTimeAsync(30000); + await assertion; + }); + + it('honors a per-call timeout shorter than the 30s default', async () => { + vi.useFakeTimers(); + const events = makeModule(); + + const pending = events.onLocal('fs:~/a', () => {}, { timeout: 5000 }); + const assertion = expect(pending).rejects.toMatchObject({ code: 'events_connection_failed' }); + await vi.advanceTimersByTimeAsync(5000); + await assertion; + }); +}); + +describe('client-side validation', () => { + it('rejects a missing subject or handler before connecting', async () => { + const events = makeModule(); + + await expect(events.onLocal('', () => {})).rejects.toMatchObject({ + code: 'invalid_subject', + }); + await expect(events.onLocal('fs:~/a', null)).rejects.toMatchObject({ + code: 'invalid_handler', + }); + expect(sockets).toHaveLength(0); + }); +}); diff --git a/src/puter-js/src/modules/events/index.js b/src/puter-js/src/modules/events/index.js new file mode 100644 index 000000000..3fe87f48f --- /dev/null +++ b/src/puter-js/src/modules/events/index.js @@ -0,0 +1,57 @@ +import { PuterModule } from '../../lib/PuterModule.js'; +import { EventChannel } from './lib/channel.js'; +import { onLocal } from './onLocal.js'; + +/** @typedef {import('../../index.js').Puter} Puter */ + +/** + * Live change notifications. Subscribe to a subject — a file, a directory, a + * path that does not exist yet — and a handler runs whenever something under + * it changes. + * + * Method implementations live in the sibling files as `this`-context functions + * whose JSDoc is the source of truth for the public signatures — `types/` is + * generated from it, never edited by hand. + */ +export class EventsModule extends PuterModule { + // The field holds the unbound function so it keeps its full type (`bind` + // erases overloads); the constructor rebinds it so destructured calls + // (`const { onLocal } = puter.events`) work. + onLocal = onLocal; + + /** @param {Puter} puter */ + constructor (puter) { + super(puter); + + /** + * @internal The single connection every subscription multiplexes + * over. + */ + this.channel = new EventChannel(this); + + const methods = /** @type {Record unknown>} */ ( + /** @type {unknown} */ (this) + ); + methods.onLocal = methods.onLocal.bind(this); + + // The socket carries its token from the moment it connects, so a new + // token means a new connection — and the subscriptions on the old one + // have to be made again on it. + puter.onAuthStateChanged(() => this.channel.reset()); + } +} + +/** + * The public face of the module: derived from the class, with the internal + * `puter` handle, the connection plumbing, and the legacy `authToken` + * accessor omitted. + * + * @typedef {import('../../lib/types.js').OmitMembers< + * typeof EventsModule, + * 'puter' | 'authToken' | 'channel' + * >} EventsConstructor + */ + +export const Events = /** @type {EventsConstructor} */ (EventsModule); + +export default Events; diff --git a/src/puter-js/src/modules/events/lib/channel.js b/src/puter-js/src/modules/events/lib/channel.js new file mode 100644 index 000000000..101a00966 --- /dev/null +++ b/src/puter-js/src/modules/events/lib/channel.js @@ -0,0 +1,431 @@ +import { io } from 'socket.io-client'; +import { PuterJSError } from '../../../lib/PuterJSError.js'; +import { socketAutoUnref } from '../../../lib/socketOptions.js'; +import { EventSubscription } from './subscription.js'; + +/** @typedef {import('../types.js').EventGapMarker} EventGapMarker */ +/** @typedef {import('../types.js').EventHandler} EventHandler */ +/** @typedef {import('../types.js').OnLocalOptions} OnLocalOptions */ +/** @typedef {import('../types.js').PuterEvent} PuterEvent */ + +/** + * What the server says a subscription is, in its `subscribe` ack. + * + * @typedef {Object} SubscriptionView + * @property {string} subId + * @property {string} subject + * @property {import('../types.js').EventAnchor} anchor + * @property {string | null} match + * @property {string | null} op + */ + +/** @typedef {{ ok: true, sub?: SubscriptionView }} VerbAck */ + +// The wire, fixed by the server: two verbs answered with an ack, one channel +// events arrive on. +const SUBSCRIBE_VERB = 'events.subscribe'; +const UNSUBSCRIBE_VERB = 'events.unsubscribe'; +const DELIVERY_CHANNEL = 'events.delivery'; + +/** How long a verb waits for its ack before the call is called lost. */ +export const DEFAULT_TIMEOUT_MS = 30000; + +// A reconnect re-issues every subscription at once, so the ones past the +// per-minute call budget wait this long for another pass rather than lapsing. +const RESUBSCRIBE_RETRY_MS = 10000; + +/** Raised when the connection itself is the problem, never by the server. */ +const connectionError = (message) => + new PuterJSError(message, 'events_connection_failed'); + +/** The server's `{ ok: false, error }` ack, passed through code and all. */ +const ackError = (response) => { + const error = /** @type {{ code?: unknown, message?: unknown } | undefined} */ ( + response && typeof response === 'object' ? response.error : undefined + ); + return new PuterJSError( + typeof error?.message === 'string' ? error.message : 'The events request failed', + typeof error?.code === 'string' ? error.code : 'events_failed', + ); +}; + +/** The subscription an ack describes, or a failure if it describes none. */ +const viewOf = (response) => { + const view = /** @type {SubscriptionView | undefined} */ (response?.sub); + if ( ! view || typeof view.subId !== 'string' ) { + throw new PuterJSError('The events server sent an unexpected answer', 'events_failed'); + } + return view; +}; + +/** + * A rejected handshake carries its own code (`reauth_required`); anything else + * is a connection that could not be made. + */ +const handshakeError = (error) => { + const data = /** @type {{ code?: unknown } | undefined} */ ( + error && typeof error === 'object' ? error.data : undefined + ); + const message = error instanceof Error ? error.message : 'Could not connect to the events server'; + return typeof data?.code === 'string' + ? new PuterJSError(message, data.code) + : connectionError(message); +}; + +/** + * The one connection every subscription rides on, and the routing table that + * makes one socket serve all of them. + * + * Opened by the first subscription and closed by the last, so an app that + * never subscribes never connects. Session subscriptions die with the socket, + * so a reconnect is not transparent server-side — this re-issues each live + * subscription and re-points its handle at the new id, which is what keeps + * `onLocal` a thing you call once. + */ +export class EventChannel { + /** @param {import('../index.js').EventsModule} module */ + constructor (module) { + /** @internal */ + this.module = module; + /** @internal @type {import('socket.io-client').Socket | null} */ + this.socket = null; + /** @internal @type {Set} */ + this.subscriptions = new Set(); + /** @internal @type {Map} */ + this.byId = new Map(); + /** @internal Rejectors for verbs still waiting on an ack. */ + this.waiters = new Set(); + /** @internal Subscribes that have not resolved yet. */ + this.inflight = 0; + /** + * @internal Bumped every time the connection goes away, so a request + * that failed with it can be told from one a live connection + * refused. + */ + this.generation = 0; + /** @internal @type {ReturnType | null} */ + this.retryTimer = null; + } + + /** + * @internal + * @param {string} subject + * @param {EventHandler} handler + * @param {OnLocalOptions} options + * @returns {Promise} + */ + async subscribe (subject, handler, options) { + const sub = new EventSubscription(this, subject, handler, options); + this.inflight++; + try { + const response = await this.request(SUBSCRIBE_VERB, { subject }, timeoutFor(sub)); + sub.apply(viewOf(response)); + this.subscriptions.add(sub); + this.byId.set(/** @type {string} */ (sub.subId), sub); + return sub; + } finally { + this.inflight--; + this.closeIfIdle(); + } + } + + /** + * @internal + * @param {EventSubscription} sub + * @returns {Promise} + */ + async remove (sub) { + if ( ! this.subscriptions.has(sub) ) return; + this.forget(sub); + + const subId = sub.subId; + sub.subId = null; + try { + if ( subId !== null && this.socket?.connected ) { + await this.request(UNSUBSCRIBE_VERB, { subId }, timeoutFor(sub)); + } + } catch { + // A session subscription is gone with its connection anyway, so + // there is nothing an unsubscribe failure leaves behind to fix — + // and `off()` is teardown, which does not get to fail. + } finally { + this.closeIfIdle(); + } + } + + /** + * Rebuild the connection against the current token and origin. Live + * subscriptions are re-issued once it is up. + * + * @internal + * @returns {void} + */ + reset () { + this.close(); + if ( this.subscriptions.size > 0 ) this.connect(); + } + + /** + * @internal + * @returns {import('socket.io-client').Socket} + */ + connect () { + if ( this.socket ) return this.socket; + + const socket = io(this.module.APIOrigin, { + auth: { auth_token: this.module.authToken }, + autoUnref: socketAutoUnref(this.module.puter), + transports: ['websocket', 'polling'], + withCredentials: true, + }); + + socket.on('connect', () => this.resubscribe()); + socket.on('disconnect', () => { + // socket.io reconnects on its own after a transport drop, but not + // after the server hangs up: that socket is finished, and so is + // everything riding it. + if ( socket.active ) { + this.orphan(); + return; + } + this.fail(connectionError('The events connection was closed by the server')); + }); + socket.on('connect_error', error => { + // socket.io retries on its own while the socket is still active; + // only a refusal it will not retry is the client's problem. + if ( socket.active ) return; + this.fail(handshakeError(error)); + }); + socket.on(DELIVERY_CHANNEL, envelope => this.route(envelope)); + + this.socket = socket; + return socket; + } + + /** + * @internal + * @param {string} verb + * @param {Record} payload + * @param {number} timeoutMs + * @returns {Promise} + */ + request (verb, payload, timeoutMs) { + const socket = this.connect(); + return new Promise((resolve, reject) => { + let settled = false; + const abort = error => { + if ( settled ) return; + settled = true; + clearTimeout(timer); + this.waiters.delete(abort); + reject(error); + }; + const timer = setTimeout( + () => abort(connectionError(`Timed out waiting for \`${verb}\``)), + timeoutMs, + ); + timer?.unref?.(); + this.waiters.add(abort); + + socket.emit(verb, payload, response => { + if ( settled ) return; + settled = true; + clearTimeout(timer); + this.waiters.delete(abort); + if ( ! response || response.ok !== true ) { + reject(ackError(response)); + return; + } + resolve(response); + }); + }); + } + + /** + * The server dropped every subscription this socket held when it went + * away, so nothing here has a server-side id until it is re-issued — and + * nothing that was waiting on an ack is going to get one. + * + * @internal + * @returns {void} + */ + orphan () { + this.generation++; + for ( const sub of this.subscriptions ) { + if ( sub.subId !== null ) this.byId.delete(sub.subId); + sub.subId = null; + sub.pending = false; + } + this.rejectWaiters(connectionError('The events connection dropped')); + } + + /** + * @internal + * @returns {void} + */ + resubscribe () { + const generation = this.generation; + for ( const sub of [...this.subscriptions] ) { + if ( sub.subId !== null || sub.pending ) continue; + sub.pending = true; + this.request(SUBSCRIBE_VERB, { subject: sub.subject }, timeoutFor(sub)) + .then(response => { + sub.pending = false; + const view = viewOf(response); + // `off()` while this was in flight: the handle is already + // gone, so drop what the server just handed us. + if ( ! this.subscriptions.has(sub) ) { + this.dropOnServer(view.subId, timeoutFor(sub)); + return; + } + sub.apply(view); + this.byId.set(/** @type {string} */ (sub.subId), sub); + }) + .catch(error => { + sub.pending = false; + if ( ! this.subscriptions.has(sub) ) return; + // The connection went away under it — a flaky reconnect is + // not a refusal, and the next connect tries again. + if ( this.generation !== generation ) return; + const failure = PuterJSError.from(error); + // Over the call budget says nothing about this subject. + if ( failure.code === 'too_many_requests' ) { + this.retryResubscribe(generation); + return; + } + this.lapse(sub, failure); + }); + } + } + + /** + * Run `resubscribe` again once the call budget has had time to refill. + * One timer covers every subscription that was turned away. + * + * @internal + * @param {number} generation + * @returns {void} + */ + retryResubscribe (generation) { + if ( this.retryTimer ) return; + this.retryTimer = setTimeout(() => { + this.retryTimer = null; + if ( this.generation !== generation || ! this.socket?.connected ) return; + this.resubscribe(); + }, RESUBSCRIBE_RETRY_MS); + this.retryTimer?.unref?.(); + } + + /** + * @internal + * @param {{ subId?: string, event?: unknown }} envelope + * @returns {void} + */ + route (envelope) { + if ( ! envelope || typeof envelope !== 'object' ) return; + const sub = this.byId.get(/** @type {string} */ (envelope.subId)); + // An event for something this client has already unsubscribed from: + // in flight when `off()` was called, and no longer anybody's. + if ( ! sub || ! envelope.event ) return; + sub.deliver(/** @type {PuterEvent | EventGapMarker} */ (envelope.event)); + } + + /** + * The connection is not coming back: fail what is waiting on it and end + * every subscription it was carrying. + * + * @internal + * @param {PuterJSError} error + * @returns {void} + */ + fail (error) { + this.rejectWaiters(error); + for ( const sub of [...this.subscriptions] ) this.lapse(sub, error); + this.close(); + } + + /** + * @internal + * @param {EventSubscription} sub + * @param {PuterJSError} error + * @returns {void} + */ + lapse (sub, error) { + this.forget(sub); + sub.subId = null; + if ( ! sub.onError ) { + console.warn(`[puter.events] subscription to ${sub.subject} lapsed`, error); + } else { + try { + sub.onError(error); + } catch (handlerError) { + console.error('[puter.events] onError handler failed', handlerError); + } + } + this.closeIfIdle(); + } + + /** + * @internal + * @param {EventSubscription} sub + * @returns {void} + */ + forget (sub) { + this.subscriptions.delete(sub); + if ( sub.subId !== null ) this.byId.delete(sub.subId); + } + + /** + * @internal + * @param {PuterJSError} error + * @returns {void} + */ + rejectWaiters (error) { + for ( const abort of [...this.waiters] ) abort(error); + this.waiters.clear(); + } + + /** + * Best-effort removal of a subscription no handle points at any more. + * + * @internal + * @param {string} subId + * @param {number} timeoutMs + * @returns {void} + */ + dropOnServer (subId, timeoutMs) { + if ( ! this.socket?.connected ) return; + this.request(UNSUBSCRIBE_VERB, { subId }, timeoutMs).catch(() => {}); + } + + /** + * @internal + * @returns {void} + */ + closeIfIdle () { + if ( this.subscriptions.size > 0 || this.inflight > 0 ) return; + this.close(); + } + + /** + * @internal + * @returns {void} + */ + close () { + if ( this.retryTimer ) clearTimeout(this.retryTimer); + this.retryTimer = null; + const socket = this.socket; + if ( ! socket ) return; + this.socket = null; + this.orphan(); + socket.removeAllListeners(); + socket.disconnect(); + } +} + +/** + * @param {EventSubscription} sub + * @returns {number} + */ +const timeoutFor = (sub) => + typeof sub.timeout === 'number' && sub.timeout > 0 ? sub.timeout : DEFAULT_TIMEOUT_MS; diff --git a/src/puter-js/src/modules/events/lib/subscription.js b/src/puter-js/src/modules/events/lib/subscription.js new file mode 100644 index 000000000..a1a74e363 --- /dev/null +++ b/src/puter-js/src/modules/events/lib/subscription.js @@ -0,0 +1,122 @@ +/** @typedef {import('../types.js').EventAnchor} EventAnchor */ +/** @typedef {import('../types.js').EventGapMarker} EventGapMarker */ +/** @typedef {import('../types.js').EventHandler} EventHandler */ +/** @typedef {import('../types.js').PuterEvent} PuterEvent */ + +/** + * A live subscription, as returned by `puter.events.onLocal()`. + * + * The handle survives reconnects: the connection dying takes the server's + * subscription with it, the SDK makes a new one, and this object keeps + * pointing at it — with a new `subId`, which is why nothing should be stored + * against that id. + */ +export class EventSubscription { + /** The subject this was subscribed with. */ + subject; + + /** + * The server's id for the current subscription, or `null` while the + * connection is down. Changes on every reconnect. + * + * @type {string | null} + */ + subId = null; + + /** + * The node the subscription is keyed to, which is the nearest existing + * ancestor when the subject named something that does not exist yet. + * + * @type {EventAnchor | null} + */ + anchor = null; + + /** + * The pattern events under the anchor are matched against, or `null` when + * the subject named the anchor itself. + * + * @type {string | null} + */ + match = null; + + /** + * The single operation this subscription is limited to, or `null` for all + * of them. + * + * @type {string | null} + */ + op = null; + + /** + * @internal + * @param {import('./channel.js').EventChannel} channel + * @param {string} subject + * @param {EventHandler} handler + * @param {{ onError?: (error: Error & { code?: string }) => void, timeout?: number }} options + */ + constructor (channel, subject, handler, options = {}) { + /** @internal @type {import('./channel.js').EventChannel} */ + this.channel = channel; + this.subject = subject; + /** @internal @type {EventHandler} */ + this.handler = handler; + /** @internal @type {((error: Error & { code?: string }) => void) | undefined} */ + this.onError = options.onError; + /** @internal @type {number | undefined} */ + this.timeout = options.timeout; + /** @internal Set while a subscribe for this handle is in flight. */ + this.pending = false; + + this.off = this.off.bind(this); + } + + /** + * Ends the subscription. Routing stops immediately; the server is told + * when there is still a connection to tell it over. Safe to call more than + * once, and after the connection has gone away — it never throws. + * + * @returns {Promise} + */ + async off () { + await this.channel.remove(this); + } + + /** + * @internal + * @param {PuterEvent | EventGapMarker} event + * @returns {void} + */ + deliver (event) { + try { + const result = this.handler({ event }); + if ( result instanceof Promise ) { + result.catch(reportHandlerError); + } + } catch (error) { + reportHandlerError(error); + } + } + + /** + * @internal + * @param {{ subId: string, anchor?: EventAnchor, match?: string | null, op?: string | null }} view + * @returns {void} + */ + apply (view) { + this.subId = view.subId; + this.anchor = view.anchor ?? null; + this.match = view.match ?? null; + this.op = view.op ?? null; + } +} + +/** + * A handler that throws is the app's bug, not the subscription's: report it + * and keep delivering. + * + * @param {unknown} error + * @returns {void} + */ +const reportHandlerError = (error) => { + console.error('[puter.events] subscription handler failed', error); +}; diff --git a/src/puter-js/src/modules/events/lib/validate.js b/src/puter-js/src/modules/events/lib/validate.js new file mode 100644 index 000000000..6e2f564ef --- /dev/null +++ b/src/puter-js/src/modules/events/lib/validate.js @@ -0,0 +1,25 @@ +import { PuterJSError } from '../../../lib/PuterJSError.js'; + +// Cheap preconditions only. The subject grammar is parsed server-side, and +// duplicating it here would mean two parsers to keep in agreement; these are +// the checks that cost nothing and save a round trip. + +/** + * @param {unknown} subject + * @returns {void} + */ +export const assertSubject = (subject) => { + if ( typeof subject !== 'string' || subject.trim().length === 0 ) { + throw new PuterJSError('Subject must be a non-empty string', 'invalid_subject'); + } +}; + +/** + * @param {unknown} handler + * @returns {void} + */ +export const assertHandler = (handler) => { + if ( typeof handler !== 'function' ) { + throw new PuterJSError('Handler must be a function', 'invalid_handler'); + } +}; diff --git a/src/puter-js/src/modules/events/onLocal.js b/src/puter-js/src/modules/events/onLocal.js new file mode 100644 index 000000000..27394a7e2 --- /dev/null +++ b/src/puter-js/src/modules/events/onLocal.js @@ -0,0 +1,32 @@ +import { assertHandler, assertSubject } from './lib/validate.js'; + +/** @typedef {import('./lib/subscription.js').EventSubscription} EventSubscription */ +/** @typedef {import('./types.js').EventHandler} EventHandler */ +/** @typedef {import('./types.js').OnLocalOptions} OnLocalOptions */ + +/** + * Subscribes to a subject for as long as this client is connected. + * + * The subscription belongs to the connection, not to the account: it is not + * stored anywhere and nothing runs while the page is closed. Every + * subscription this client makes shares one connection, which opens on the + * first `onLocal()` and closes when the last subscription is ended. + * + * The handler is called with `{ event }` for every matching change, and with a + * gap marker (`event.op === 'gap'`) in place of events that were dropped + * against a limit. + * + * @this {import('./index.js').EventsModule} + * @param {string} subject The subject to watch, e.g. `fs:~/Documents` or + * `fs:~/Documents/inbox.txt:write`. + * @param {EventHandler} handler Called with `{ event }` per delivery. + * @param {OnLocalOptions} [options] + * @returns {Promise} Resolves once the server has confirmed + * the subscription. + */ +export async function onLocal (subject, handler, options = {}) { + assertSubject(subject); + assertHandler(handler); + + return await this.channel.subscribe(subject, handler, options); +} diff --git a/src/puter-js/src/modules/events/types.js b/src/puter-js/src/modules/events/types.js new file mode 100644 index 000000000..c9eb916db --- /dev/null +++ b/src/puter-js/src/modules/events/types.js @@ -0,0 +1,78 @@ +// Shapes shared across the `puter.events` surface. JSDoc-only; no runtime exports. + +/** + * The node a subscription is keyed to. For a subject naming something that + * does not exist yet, this is the nearest existing ancestor and the rest of + * the subject became `match`. + * + * @typedef {Object} EventAnchor + * @property {string} uid The anchor node's uid. + * @property {string} path The anchor node's absolute path. + */ + +/** + * One change, as the server projects it. Nothing internal is included: a + * subscriber gets the node, when it happened, and whether it was their own + * doing. + * + * @typedef {Object} PuterEvent + * @property {string} id Unique id for this event. Stable across the + * subscriptions it was delivered to. + * @property {string} subject The subject the delivery was projected onto, + * naming the node it happened to — `fs::`. Not the subject string + * you subscribed with. + * @property {'add' | 'write' | 'move' | 'remove' | 'meta'} op What happened. + * @property {string} uid The uid of the node the event is about. + * @property {string} path The path of the node the event is about. + * @property {boolean} self `true` when the change was made by the account + * holding the subscription — the flag to check to ignore your own writes. + * @property {number} ts Milliseconds since the epoch. + * @property {number} seq Position within one dispatch, for events that fan out + * to several subscriptions at once. + */ + +/** + * Stands in for events that happened and were not delivered — a per-event + * ceiling was hit, or deliveries were coming faster than the subscription's + * allowance. It carries no `uid` or `path`, because what was dropped is + * exactly what it cannot name: treat it as "re-read the anchor", never as + * "nothing changed". + * + * @typedef {Object} EventGapMarker + * @property {string} id Unique id for the dispatch the gap happened in. + * @property {string} subject The subject that was being delivered. + * @property {'gap'} op Always `'gap'`. + * @property {string} reason Why the delivery was dropped — + * `matched_subscription_limit`, `filter_evaluation_limit`, or + * `delivery_rate_limit`. + * @property {number} ts Milliseconds since the epoch. + */ + +/** + * What a handler is called with. An object rather than the event itself, so + * more can be added to the call without breaking existing handlers. + * + * @typedef {Object} EventDelivery + * @property {PuterEvent | EventGapMarker} event The delivered event, or a gap + * marker in place of events that were dropped. + */ + +/** + * A subscription handler. Its return value is ignored; a rejected promise is + * reported and does not affect the subscription. + * + * @typedef {(delivery: EventDelivery) => unknown} EventHandler + */ + +/** + * Options for {@link import('./onLocal.js').onLocal}. + * + * @typedef {Object} OnLocalOptions + * @property {(error: Error & { code?: string }) => void} [onError] Called if + * the subscription lapses — the connection was lost and re-subscribing + * failed. The subscription is gone by then and the handler will not be + * called again; subscribe again to resume. Without this, a lapse is reported + * on the console. + * @property {number} [timeout] How long to wait for the server to answer + * `subscribe`, in milliseconds. Default `30000`. + */ diff --git a/src/puter-js/tests/api/harness/capabilities.ts b/src/puter-js/tests/api/harness/capabilities.ts index cdf1463d4..724d0fcf0 100644 --- a/src/puter-js/tests/api/harness/capabilities.ts +++ b/src/puter-js/tests/api/harness/capabilities.ts @@ -98,6 +98,10 @@ export const loadPuterJsTestOptions = ( meteringEnforcement: { subscriptions: false }, // ~500 tests share one account; 'unlimited' means paid-base limits. unlimitedMetering: true, + // `puter.events` is a socket surface, so the suite needs the server + // half switched on. Off, every subscribe answers `events_disabled` and + // the suite would only ever cover that one branch. + events: { enabled: true }, }; for (const mapping of MAPPINGS) { diff --git a/src/puter-js/tests/api/suites/events.suite.ts b/src/puter-js/tests/api/suites/events.suite.ts new file mode 100644 index 000000000..816e9c67d --- /dev/null +++ b/src/puter-js/tests/api/suites/events.suite.ts @@ -0,0 +1,253 @@ +import { suite, type TestContext } from '../harness/types.ts'; + +// `puter.events` rides a socket; every runtime the suite runs on (node, +// browser, workerd) carries one, so a subscribe that fails is a failure here. + +/** Short enough that a runtime that cannot connect fails fast, not at 30 s. */ +const SUBSCRIBE_TIMEOUT_MS = 5000; +const DELIVERY_TIMEOUT_MS = 15000; +/** Deliveries are coalesced server-side over 250 ms. */ +const QUIET_MS = 2000; + +type Delivered = { + id: string; + subject: string; + op: string; + uid?: string; + path?: string; + self?: boolean; + ts: number; + seq?: number; +}; + +type Subscription = Awaited>; + +const codeOf = (error: unknown): string | undefined => + (error as { code?: string } | undefined)?.code; + +const sleep = (ms: number): Promise => + new Promise((resolve) => setTimeout(resolve, ms)); + +const waitFor = async ( + condition: () => boolean, + timeoutMs: number, +): Promise => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (condition()) return true; + await sleep(50); + } + return condition(); +}; + +const unique = (prefix: string): string => + `${prefix}-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`; + +/** A directory of this test's own to anchor a subscription on. */ +const makeDir = async (t: TestContext, name: string): Promise => { + const path = `/${t.env.users.user.username}/${unique(name)}`; + await t.puter.fs.mkdir(path, { createMissingParents: true }); + return path; +}; + +const open = ( + t: TestContext, + subject: string, + handler: (event: Delivered) => void, +): Promise => + t.puter.events.onLocal(subject, ({ event }) => handler(event as Delivered), { + timeout: SUBSCRIBE_TIMEOUT_MS, + }); + +export default suite('events', { + 'exposes onLocal': async (t) => { + t.assert.ok(t.puter.events, 'puter.events is registered'); + t.assert.equal(typeof t.puter.events.onLocal, 'function'); + }, + + 'rejects a subject that is not a non-empty string': async (t) => { + for (const subject of [undefined, null, '', ' ', 42, {}]) { + const error = await t.assert.rejects( + () => + t.puter.events.onLocal( + subject as unknown as string, + () => {}, + ), + `subject ${JSON.stringify(subject)} should be rejected`, + ); + t.assert.equal(codeOf(error), 'invalid_subject'); + } + }, + + 'rejects a handler that is not a function': async (t) => { + const error = await t.assert.rejects(() => + t.puter.events.onLocal( + `fs:/${t.env.users.user.username}`, + undefined as unknown as () => void, + ), + ); + t.assert.equal(codeOf(error), 'invalid_handler'); + }, + + 'delivers the projected event shape': async (t) => { + const dir = await makeDir(t, 'events-shape'); + const seen: Delivered[] = []; + + const sub = await open(t, `fs:${dir}`, (event) => seen.push(event)); + if (!sub) return; + + try { + t.assert.ok(sub.subId, 'a live subscription carries a server id'); + t.assert.equal(sub.subject, `fs:${dir}`); + t.assert.equal(sub.anchor?.path, dir); + + const file = `${dir}/note.txt`; + await t.puter.fs.write(file, 'hello'); + + await waitFor( + () => seen.some((event) => event.path === file), + DELIVERY_TIMEOUT_MS, + ); + const event = seen.find((e) => e.path === file); + t.assert.ok(event, `no delivery for ${file}; saw ${JSON.stringify(seen)}`); + t.assert.deepEqual( + Object.keys(event as Delivered).sort(), + ['id', 'op', 'path', 'self', 'seq', 'subject', 'ts', 'uid'], + 'the delivered event carries exactly the projected fields', + ); + t.assert.equal(event?.self, true, 'the writer is the subscriber'); + t.assert.ok(event?.uid, 'the event names the node'); + t.assert.ok( + ['add', 'write'].includes(event?.op as string), + `unexpected op: ${event?.op}`, + ); + } finally { + await sub.off(); + } + }, + + 'off stops delivery and can be called more than once': async (t) => { + const dir = await makeDir(t, 'events-off'); + const seen: Delivered[] = []; + + const sub = await open(t, `fs:${dir}`, (event) => seen.push(event)); + if (!sub) return; + + await sub.off(); + t.assert.equal(sub.subId, null, 'off clears the server id'); + // Teardown is idempotent, and never throws once the connection is gone. + await sub.off(); + + await t.puter.fs.write(`${dir}/after-off.txt`, 'quiet'); + await sleep(QUIET_MS); + t.assert.deepEqual(seen, [], 'nothing is delivered after off()'); + }, + + 'several subscriptions share one connection': async (t) => { + const watched = await makeDir(t, 'events-multi-a'); + const other = await makeDir(t, 'events-multi-b'); + const watchedSeen: Delivered[] = []; + const otherSeen: Delivered[] = []; + + const first = await open(t, `fs:${watched}`, (e) => watchedSeen.push(e)); + if (!first) return; + const second = await open(t, `fs:${other}`, (e) => otherSeen.push(e)); + if (!second) { + await first.off(); + return; + } + + try { + t.assert.ok( + first.subId !== second.subId, + 'each subscription gets its own id', + ); + + const file = `${watched}/only-here.txt`; + await t.puter.fs.write(file, 'x'); + await waitFor( + () => watchedSeen.some((event) => event.path === file), + DELIVERY_TIMEOUT_MS, + ); + + t.assert.ok( + watchedSeen.some((event) => event.path === file), + 'the subscription on the written directory hears it', + ); + t.assert.deepEqual( + otherSeen, + [], + 'the subscription on the other directory hears nothing', + ); + } finally { + await first.off(); + await second.off(); + } + }, + + 'resubscribes when the connection is rebuilt': async (t) => { + const dir = await makeDir(t, 'events-reconnect'); + const seen: Delivered[] = []; + + const sub = await open(t, `fs:${dir}`, (event) => seen.push(event)); + if (!sub) return; + + try { + const before = sub.subId; + + // Rebuilding auth state drops the socket the way a reconnect + // does; the server's session subscriptions go with it. + t.puter.setAPIOrigin(t.puter.APIOrigin); + + const back = await waitFor( + () => sub.subId !== null && sub.subId !== before, + DELIVERY_TIMEOUT_MS, + ); + t.assert.ok(back, 'the subscription was re-established with a new id'); + + const file = `${dir}/after-reconnect.txt`; + await t.puter.fs.write(file, 'still listening'); + await waitFor( + () => seen.some((event) => event.path === file), + DELIVERY_TIMEOUT_MS, + ); + t.assert.ok( + seen.some((event) => event.path === file), + 'the same handler keeps receiving events after the rebuild', + ); + } finally { + await sub.off(); + } + }, + + 'passes server error codes through unchanged': async (t) => { + const dir = await makeDir(t, 'events-errors'); + const cases: Array<[string, string]> = [ + // An op the subject grammar does not define. + [`fs:${dir}:frobnicate`, 'invalid_subject_op'], + // A pattern past the compile-cost bounds (16 segments). + [`fs:${dir}/${'deep/'.repeat(20)}x`, 'invalid_subject_pattern'], + // Another account's home: refused as absent, so the call cannot be + // used to find out what exists. + [ + `fs:/${t.env.users.other.username}/${unique('nope')}`, + 'subject_does_not_exist', + ], + ]; + + for (const [subject, expected] of cases) { + const error = await t.assert.rejects( + () => + t.puter.events.onLocal(subject, () => {}, { + timeout: SUBSCRIBE_TIMEOUT_MS, + }), + `${subject} should be refused`, + ); + t.assert.equal( + codeOf(error), + expected, + `${subject} answered ${codeOf(error)}`, + ); + } + }, +}); diff --git a/src/puter-js/tests/api/suites/index.ts b/src/puter-js/tests/api/suites/index.ts index 5eda206c2..bab77f3b8 100644 --- a/src/puter-js/tests/api/suites/index.ts +++ b/src/puter-js/tests/api/suites/index.ts @@ -3,6 +3,7 @@ import ai from './ai.suite.ts'; import apps from './apps.suite.ts'; import auth from './auth.suite.ts'; import components from './components.suite.ts'; +import events from './events.suite.ts'; import fs from './fs.suite.ts'; import hosting from './hosting.suite.ts'; import kv from './kv.suite.ts'; @@ -23,6 +24,7 @@ export const suites: Suite[] = [ apps, auth, components, + events, fs, hosting, kv,