{
+ await this.clients.redis.del(holdKey(subId));
+ }
+
// -- Reads -------------------------------------------------------
/**
diff --git a/src/backend/stores/index.ts b/src/backend/stores/index.ts
index 8e1cef9c1..ce0149e47 100644
--- a/src/backend/stores/index.ts
+++ b/src/backend/stores/index.ts
@@ -22,6 +22,7 @@ import { AppStore } from './app/AppStore.js';
import { FSEntryStore } from './fs/FSEntryStore.js';
import { GroupStore } from './group/GroupStore.js';
import { DurableSubscriptionStore } from './events/DurableSubscriptionStore.js';
+import { EventHandlerStore } from './events/EventHandlerStore.js';
import { EventSubscriptionStore } from './events/EventSubscriptionStore.js';
import { PendingDeliveryStore } from './events/PendingDeliveryStore.js';
import { CreditHoldStore } from './metering/CreditHoldStore.js';
@@ -65,6 +66,7 @@ declare module './types.js' {
userBlock: UserBlockStore;
eventSubscription: EventSubscriptionStore;
durableSubscription: DurableSubscriptionStore;
+ eventHandler: EventHandlerStore;
pendingDelivery: PendingDeliveryStore;
}
}
@@ -100,4 +102,6 @@ export const puterStores = {
pendingDelivery: PendingDeliveryStore,
// Writes through the Redis keyspace above, so it comes after it.
durableSubscription: DurableSubscriptionStore,
+ // Table only, and reads the subscription table for its dependent counts.
+ eventHandler: EventHandlerStore,
} satisfies IPuterStoreRegistry;
diff --git a/src/docs/src/Events.md b/src/docs/src/Events.md
index bb356f4d5..c13040d54 100644
--- a/src/docs/src/Events.md
+++ b/src/docs/src/Events.md
@@ -119,7 +119,7 @@ Emptying a whole store with [`puter.kv.flush()`](/KV/flush/) delivers nothing: n
### 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".
+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". A persistent subscription that was suspended long enough for its held backlog to lapse gets one too, with `reason: 'suspended_backlog_expired'`.
```js
await puter.events.onLocal('fs:~/Documents', async ({ event }) => {
@@ -128,7 +128,7 @@ await puter.events.onLocal('fs:~/Documents', async ({ event }) => {
});
```
-## Subscriptions live with the connection
+## Two kinds of subscription
`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.
@@ -140,11 +140,43 @@ const sub = await puter.events.onLocal('fs:~/Documents', handler, {
});
```
+[`onPersistent()`](/Events/onPersistent/) subscriptions are **stored against the account**. They keep matching with nothing open, survive every reconnect, and end only when you call [`unsubscribe()`](/Events/unsubscribe/) or their `expiresAt` passes. What runs is a *handler* your app deployed by name:
+
+```js
+// Once, at deploy time
+await puter.events.handlers.publish('ingestUpload', async ({ event, ctx }) => {
+ await fetch(ctx.endpoint, { method: 'POST', body: event.path });
+}, { appUid });
+
+// Per user, when they opt in
+await puter.events.onPersistent({
+ subject: 'fs:~/inbox',
+ handlerName: 'ingestUpload',
+ context: { endpoint: 'https://example.com/ingest' },
+});
+```
+
+### Handlers cannot close over anything
+
+A handler is deployed, not called: it is serialized with `Function.prototype.toString()` and run later, somewhere else, with nothing around it. A closed-over variable is not discouraged — it is *unrepresentable*. Every identifier a handler names has to be a parameter, something it declares itself, a standard global, or reached through `ctx`; the SDK checks that before the call and rejects with `events_handler_free_variable`, naming what it could not resolve.
+
+Values reach a handler through **`context`**, which is evaluated **once, at subscribe time**, serialized, and delivered to every invocation as a frozen `ctx`. It never re-evaluates: `ctx.endpoint` is whatever the value was when the subscription was created, forever, until it is created again.
+
+**`context` is capped at a hard 4 KB.** These are database rows read on every delivery, and `context` is the one field a developer controls the size of — over the cap the call fails with `events_context_too_large`, client-side, before the request. It is stored in plaintext and read only on the delivery path: [`list()`](/Events/list/) returns its **key names and a content hash**, never its values. For anything larger, store it in a file and put the path in `context`; a wider column is not the upgrade path.
+
+See [`puter.events.handlers`](/Events/handlers/) for the deploy side — publishing, replacing, and what removing a name does to the subscriptions bound to it.
+
+A persistent subscription can also stop without you unsubscribing: its handler was removed, its holder ran out of credit, or the share it was made under was withdrawn. It is then *suspended* rather than deleted, and [`list()`](/Events/list/) reports `suspendedAt` and `suspendedReason`. Everything but a withdrawn grant can resume.
+
## 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.
+Subscriptions per connection, persistent subscriptions per account, published handlers per app, 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
+- **[`subscription.off()`](/Events/off/)** - End a session subscription
+- **[`puter.events.onPersistent()`](/Events/onPersistent/)** - Subscribe with a subscription that keeps running when your app is closed
+- **[`puter.events.list()`](/Events/list/)** - List the persistent subscriptions this caller holds
+- **[`puter.events.unsubscribe()`](/Events/unsubscribe/)** - End a persistent subscription
+- **[`puter.events.handlers`](/Events/handlers/)** - Publish, list and remove the named handlers a persistent subscription runs
diff --git a/src/docs/src/Events/handlers.md b/src/docs/src/Events/handlers.md
new file mode 100644
index 000000000..2309a9481
--- /dev/null
+++ b/src/docs/src/Events/handlers.md
@@ -0,0 +1,203 @@
+---
+title: puter.events.handlers
+description: Publish, list and remove the named handlers a persistent subscription runs.
+platforms: [websites, apps, nodejs, workers]
+---
+
+The Events API is in beta. Event shapes, limits, and behavior may change between releases.
+
+A **handler** is a function your app deploys once, under a name, that persistent subscriptions bind to. A name is a label for deployed code, not an event: nothing triggers by name, and a handler runs only when a subscription bound to it has a delivery.
+
+Publishing is a **developer** operation. An app token publishes into its own app; an account session has to name an app it owns with `appUid`. Either way the account must own the app.
+
+```js
+await puter.events.handlers.publish('ingestUpload', async ({ event, ctx }) => {
+ await fetch(ctx.endpoint, { method: 'POST', body: event.path });
+}, { appUid });
+
+await puter.events.handlers.list({ appUid }); // [{ name, hash, updatedAt, subscriptions }]
+await puter.events.handlers.remove('indexDocument', { appUid });
+```
+
+## Handlers cannot close over anything
+
+A handler is serialized with `Function.prototype.toString()` and run later, somewhere else. A closed-over variable is not discouraged — it is **unrepresentable**, because nothing around the function survives the trip.
+
+So every identifier a handler names must be one of: a parameter, something the handler itself declares, a standard global (`fetch`, `JSON`, `Math`, `console`, `URL`, `crypto`, …), or reached through `ctx`. The SDK checks this before the request and rejects with `events_handler_free_variable`, naming the identifier:
+
+```js
+const endpoint = 'https://example.com/ingest';
+
+// Rejected: `endpoint` is not a parameter, a local, or a known global.
+await puter.events.handlers.publish('ingestUpload', ({ event }) => fetch(endpoint), { appUid });
+
+// Accepted: the value travels with the subscription, not with the code.
+await puter.events.handlers.publish('ingestUpload', ({ event, ctx }) => fetch(ctx.endpoint), { appUid });
+await puter.events.onPersistent({ subject: 'fs:~/inbox', handlerName: 'ingestUpload', context: { endpoint } });
+```
+
+The check is deliberately conservative: anything it cannot resolve is refused with a clear message, rather than accepted and failed on first delivery in production.
+
+## `publish()`
+
+```js
+puter.events.handlers.publish(name, handler)
+puter.events.handlers.publish(name, handler, options)
+```
+
+- `name` (String) (required): The name subscriptions bind to. Letters, digits and `_ . : -`, starting alphanumeric, up to 128 characters. Unique per app, and stable across source changes.
+- `handler` (Function | String | Object) (required): A function (serialized with `toString()`), a source string, or `{ file: '~/AppData/…/handler.js' }`. **A file reference resolves now, not at delivery** — the bytes as they are at this call are what gets deployed, so editing the file afterwards changes nothing until you publish again.
+- `options.replace` (Boolean): Take the name whatever is published under it.
+- `options.appUid` (String): The app to publish into. Required for an account session.
+
+Resolves to `{ name, hash, updatedAt, outcome, resumed }`. `outcome` is `'created'`, `'updated'`, or `'unchanged'` when the same source was already published. `resumed` counts subscriptions this publish brought back out of suspension.
+
+### Two build steps must not silently pick a winner
+
+The source hash is a change detector and an idempotency key: publishing the **same** source again is a no-op. Publishing **different** source is an update — but only from a caller that knows what it is updating.
+
+The SDK remembers the hash it last saw published for each name and sends it as the base. A publish whose base has moved under it — a second build step got there first — is refused with `events_handler_conflict`. Pass `replace: true` to say you mean to take the name regardless.
+
+A client that has never published or listed that name sends no base, so its publish can only create, or be idempotent.
+
+## `publishAll()`
+
+```js
+puter.events.handlers.publishAll(handlers)
+puter.events.handlers.publishAll(handlers, options)
+```
+
+Publishes a set in one call — what a build step has. `handlers` is an array of `{ name, handler, replace? }`, capped at 50 entries and taken in order. An item the server refuses stops the pass, so a deploy never reports success over a half-published set; items before it are published, and the error names where it stopped.
+
+Resolves to an array of the same objects `publish()` returns.
+
+## `list()`
+
+```js
+puter.events.handlers.list()
+puter.events.handlers.list(options)
+```
+
+Resolves to `[{ name, hash, updatedAt, subscriptions }]` for everything the app has published, ordered by name. `subscriptions` counts what is bound to that name, **suspended ones included** — a suspended subscription is still a dependent, and it is the reason removing a name is not just a delete.
+
+**Source is never returned.** It is the app's own code, read only on the delivery path.
+
+## `remove()`
+
+```js
+puter.events.handlers.remove(name)
+puter.events.handlers.remove(name, options)
+```
+
+Resolves to `{ name, removed, suspended }`.
+
+| Situation | What happens |
+| --- | --- |
+| Nothing is bound to the name | The handler is deleted outright. |
+| Subscriptions are bound to it | The handler is deleted **and** every subscription on it is *suspended* with `suspendedReason: 'handler_not_found'` — not deleted. The app's developer is notified. |
+
+**Publishing the name again resumes them.** That is what makes a bad deploy recoverable: the subscriptions keep their ids, their context and their place, and start delivering again on the next publish.
+
+Renaming is publish-new plus remove-old, and subscriptions do **not** follow — that is a re-subscribe, deliberately: silently repointing someone's subscription at different code is exactly what consent is protecting against.
+
+### What a suspension does to the backlog
+
+A suspended subscription stops being delivered to and stops being metered — so it cannot go on holding a full backlog for free. On suspension its undelivered deliveries are trimmed to **100** and given a deadline: **24 hours** for `handler_not_found` and `failures`, **1 hour** for `no_credit`. Past the deadline they are dropped and one `gap` marker with `reason: 'suspended_backlog_expired'` takes their place, so a resumed subscription learns there were events rather than reading the silence as "nothing changed". A subscription suspended by `permission_revoked` has its backlog **purged at once** and never resumes.
+
+## Errors
+
+All four methods reject with `{ message, code }`:
+
+| `code` | Meaning |
+| --- | --- |
+| `events_handler_free_variable` | The handler names something it cannot carry. The message names the identifier. |
+| `events_handler_invalid` | `handler` is not a function, a source string, or `{ file }`. |
+| `events_handler_name_invalid` | The name is empty, too long, or not an addressable identifier. |
+| `events_handler_conflict` | Different source is published under this name and the caller did not name it as the base. Pass `replace: true` to take it. |
+| `events_handler_app_required` | An account session did not name an app. |
+| `events_handler_forbidden` | The caller does not own the app — and an app that is not there answers the same way. |
+| `events_handler_too_large` | The serialized handler is over 64 KB. |
+| `events_handler_source_invalid` | The handler source is empty. |
+| `events_handler_limit` | The app already has the maximum number of published handlers. |
+| `too_many_requests` | Over the handler publish/remove budget. |
+| `events_disabled` | Events are not enabled on this server. |
+
+## Examples
+
+Publish a handler, bind a subscription to it, then take it away
+
+```html
+
+
+
+
+
+
+```
+
+Deploy a whole set from a build step
+
+```html
+
+
+
+
+
+
+```
diff --git a/src/docs/src/Events/list.md b/src/docs/src/Events/list.md
new file mode 100644
index 000000000..d0278fff8
--- /dev/null
+++ b/src/docs/src/Events/list.md
@@ -0,0 +1,89 @@
+---
+title: puter.events.list()
+description: List the persistent subscriptions this caller holds.
+platforms: [websites, apps, nodejs, workers]
+---
+
+The Events API is in beta. Event shapes, limits, and behavior may change between releases.
+
+Lists the persistent subscriptions created with [`puter.events.onPersistent()`](/Events/onPersistent/). Session subscriptions made with `onLocal()` are not listed — they live with the connection and are not stored anywhere.
+
+An app sees only the subscriptions it created. A session acting for the account sees them all, **including ones left behind by an app that is gone** — which is what makes the account the place a stray subscription is revoked from.
+
+## Syntax
+```js
+puter.events.list()
+puter.events.list(options)
+```
+
+## Parameters
+
+#### `options` (Object) (optional)
+
+- `limit` (Number): Maximum subscriptions per request. Capped at 200; defaults to 50.
+- `cursor` (String | null): Continuation token from a previous page. Passing it — `null` included — switches the return value to a single page envelope.
+- `includeTotal` (Boolean): Adds `total` to the envelope. Request it on the first page only; it costs more the more subscriptions exist.
+- `stream` (Boolean): Returns an async iterator of page envelopes instead of a promise.
+
+## Return value
+
+With no pagination params, a `Promise` for an array of every subscription, fetched page by page under the hood. With `cursor` or `includeTotal`, a `Promise` for one page: `{ items, cursor?, total? }` — `cursor` is present only while more pages exist. With `stream: true`, an async iterator of those envelopes.
+
+**Pages may be short.** Never read `items.length < limit` as the end of the list; iterate until `cursor` is absent.
+
+Each subscription is the object [`onPersistent()`](/Events/onPersistent/) returns. In particular:
+
+- `contextKeys` (Array | null) and `contextHash` (String | null) describe the stored `context`. **The values are never returned** — the context is where an API key lives, and a listing is the one surface an app can call repeatedly. The hash changes whenever any value does, which is enough to tell two subscriptions apart or to notice one was re-created.
+- `suspendedAt` (Number | null) and `suspendedReason` (String | null) say whether a subscription stopped delivering without being removed, and why: `handler_not_found`, `failures`, `no_credit`, or `permission_revoked`.
+
+The promise rejects with `{ message, code }` — `too_many_requests` over the listing budget, `events_disabled` where events are off.
+
+## Examples
+
+List everything this account is watching
+
+```html
+
+
+
+
+
+
+```
+
+Find the ones that stopped, and why
+
+```html
+
+
+
+
+
+
+```
diff --git a/src/docs/src/Events/onPersistent.md b/src/docs/src/Events/onPersistent.md
new file mode 100644
index 000000000..75d40376f
--- /dev/null
+++ b/src/docs/src/Events/onPersistent.md
@@ -0,0 +1,156 @@
+---
+title: puter.events.onPersistent()
+description: Subscribe to changes with a subscription that keeps running when your app is closed.
+platforms: [websites, apps, nodejs, workers]
+---
+
+The Events API is in beta. Event shapes, limits, and behavior may change between releases.
+
+Creates a subscription that outlives this connection. It is stored against the account, keeps matching while your app is closed, and runs a handler your app published with [`puter.events.handlers.publish()`](/Events/handlers/). Contrast [`puter.events.onLocal()`](/Events/onLocal/), which lives and dies with the page.
+
+See [Events](/Events/) for the subject grammar and the event shape.
+
+## Syntax
+```js
+puter.events.onPersistent(options)
+```
+
+## Parameters
+
+#### `options` (Object) (required)
+
+- `subject` (String) (required): What to watch — the same grammar `onLocal()` takes, e.g. `fs:~/Documents` or `fs:~/inbox/*.json:add`.
+- `delivery` (String): `'broadcast'` (default) delivers to everything listening. `'single'` delivers each event to exactly one consumer, which must acknowledge it, and requires `handlerName`.
+- `targets` (Array): Transports deliveries may take — any of `'socket'`, `'worker'`, `'push'`. Defaults to `['socket', 'worker']` for a subscription an app made, `['socket']` for one an account session made naming no app. A `single` subscription may not target `'push'`; a subscription with no app may not target `'worker'` — there is exactly one events worker per app, and no app means no worker to invoke.
+- `handlerName` (String): The published handler this subscription binds to. Required for `single`.
+- `handler` (Function | String | Object): The handler source this subscription was written against. Sent as a **hash**, never as source: the subscription binds only if that hash matches what is published under `handlerName`, which is why `handlerName` is required alongside it. Accepts a function, a source string, or `{ file: '~/AppData/…/handler.js' }`.
+- `context` (Object): Values the handler needs, delivered to it as a frozen `ctx`. **Capped at 4 KB serialized** — see below.
+- `expiresAt` (Number | String): When the subscription ends by itself — unix seconds or an ISO-8601 string, and it has to be in the future.
+
+## `context` is evaluated once, and capped at 4 KB
+
+A handler is deployed, not called: it is serialized and run later, somewhere else, so it cannot close over anything. `context` is how values reach it — and it is evaluated **at this call**, serialized, and never re-evaluated. `ctx.endpoint` is whatever `process.env.INGEST_URL` was when you subscribed, forever, until you subscribe again.
+
+```js
+await puter.events.onPersistent({
+ subject: 'fs:~/inbox',
+ handlerName: 'ingestUpload',
+ context: { endpoint: process.env.INGEST_URL, apiKey: process.env.INGEST_KEY },
+});
+```
+
+**The cap is a hard 4 KB.** These are database rows read on every delivery, and `context` is the one field you control the size of; over the cap the call fails with `events_context_too_large`, client-side, before the request. Context is stored in plaintext and is read only on the delivery path — [`puter.events.list()`](/Events/list/) returns its **key names and a content hash**, never its values. If you need to hand a handler more than 4 KB, put it in a file and pass the path in `context`; a wider column is not the upgrade path.
+
+## Return value
+
+A `Promise` that resolves to the subscription:
+
+- `subId` (String): Its id, and what [`puter.events.unsubscribe()`](/Events/unsubscribe/) names. Stable for the life of the subscription.
+- `subject`, `anchor`, `match`, `op`: as `onLocal()` returns them.
+- `delivery` (String), `targets` (Array), `handlerName` (String | null).
+- `appUid` (String | null): The app that created it, or `null` for one an account session made.
+- `contextKeys` (Array | null), `contextHash` (String | null): the shape of the stored context, never its values.
+- `createdAt`, `expiresAt` (Number | null): unix seconds.
+- `suspendedAt` (Number | null), `suspendedReason` (String | null): why it stopped delivering without being removed — see [`puter.events.handlers.remove()`](/Events/handlers/).
+
+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. |
+| `events_handler_name_required` | An inline `handler` was given with no `handlerName` to publish it under. |
+| `events_handler_free_variable` | The handler names something it cannot carry — a closed-over variable. The message names the identifier. |
+| `events_handler_invalid` | `handler` is not a function, a source string, or `{ file }`. |
+| `events_handler_hash_unavailable` | This environment provides no `crypto.subtle`, so an inline handler cannot be hashed. Publish it first and pass `handlerName` alone. |
+| `events_handler_not_found` | No handler is published under `handlerName`. The subscription is **not** created. |
+| `events_handler_hash_mismatch` | The published handler is not the source this subscription was written against. |
+| `events_handler_required` | `delivery: 'single'` without a `handlerName`. |
+| `events_context_too_large` | The serialized `context` is over 4 KB. |
+| `events_context_invalid` | `context` is not JSON-serializable. |
+| `invalid_targets` | A target outside `socket`/`worker`/`push`, `push` on a `single` subscription, or `worker` on a subscription with no app. |
+| `invalid_expires_at` | `expiresAt` is not a future time. |
+| `subject_does_not_exist` | The subject is not there, or this account cannot read it. |
+| `events_subscription_limit` | This account already holds the maximum number of persistent subscriptions. |
+| `too_many_requests` | Over the subscribe/unsubscribe call budget. |
+| `events_disabled` | Events are not enabled on this server. |
+
+## Examples
+
+Watch a folder with a handler that keeps running
+
+```html
+
+
+
+
+
+
+```
+
+Bind to the exact source you wrote against
+
+```html
+
+
+
+
+
+
+```
diff --git a/src/docs/src/Events/unsubscribe.md b/src/docs/src/Events/unsubscribe.md
new file mode 100644
index 000000000..4a6bfdedf
--- /dev/null
+++ b/src/docs/src/Events/unsubscribe.md
@@ -0,0 +1,66 @@
+---
+title: puter.events.unsubscribe()
+description: End a persistent subscription.
+platforms: [websites, apps, nodejs, workers]
+---
+
+The Events API is in beta. Event shapes, limits, and behavior may change between releases.
+
+Ends a subscription created with [`puter.events.onPersistent()`](/Events/onPersistent/). It stops matching immediately and everything it was still owed goes with it — a backlog held for a subscription nobody can consume is memory, and the paths it names are ones its holder just stopped asking about.
+
+For a session subscription made with [`puter.events.onLocal()`](/Events/onLocal/), use [`subscription.off()`](/Events/off/) instead.
+
+## Syntax
+```js
+puter.events.unsubscribe(subId)
+```
+
+## Parameters
+
+#### `subId` (String) (required)
+The `subId` of the subscription to end, as `onPersistent()` returned it or as [`puter.events.list()`](/Events/list/) reports it.
+
+## Return value
+
+A `Promise` that resolves when the subscription is gone.
+
+An id this caller does not hold — one already ended, or one another app created — **reads as absent** rather than refused, so the call cannot be used to find out which subscriptions exist. It rejects with `{ message, code }`:
+
+| `code` | Meaning |
+| --- | --- |
+| `subscription_does_not_exist` | No such subscription, or not this caller's. |
+| `too_many_requests` | Over the subscribe/unsubscribe call budget. |
+| `events_disabled` | Events are not enabled on this server. |
+
+An app may only end the subscriptions it created. A session acting for the account may end any of them, including ones left behind by an app that is gone.
+
+## Examples
+
+Create a persistent subscription, then end it
+
+```html
+
+
+
+
+
+
+```
diff --git a/src/docs/src/rate-limits-and-quotas.md b/src/docs/src/rate-limits-and-quotas.md
index 2d56dd9ef..08d1fcfb9 100644
--- a/src/docs/src/rate-limits-and-quotas.md
+++ b/src/docs/src/rate-limits-and-quotas.md
@@ -170,15 +170,34 @@ One write can reach many subscriptions, so events are bounded on both halves: ho
| Deliveries per minute, per subscription | 600 |
| Acknowledgements per minute | 600 |
| Undelivered deliveries per subscription | 10,000 |
+| Undelivered deliveries per *suspended* subscription | 100 |
| Suspended subscriptions kept for | 30 days |
+| Published handlers per app | 100 |
+| Handler source size | 64 KB |
+| Handlers per `publishAll` call | 50 |
+| Handler publish / remove calls per minute | 60 |
+| Handler listings per minute | 120 |
Subscriptions come in two kinds. A **session** subscription lives with the connection that made it: it is dropped when the connection closes, and a reconnecting client subscribes again. A **durable** subscription outlives every connection — it is created over the API, listed and revoked from the account, and keeps delivering until you remove it or it expires.
The 51st subscription on one connection, and the 501st durable subscription on one account, both fail with `events_subscription_limit`. Over the call budget, `subscribe` and `unsubscribe` fail with `too_many_requests`. Subscribing to something you cannot read fails with `subject_does_not_exist` — the same answer as subscribing to something that is not there, so the call cannot be used to find out which.
-A durable subscription may carry a `context`: JSON that is stored with it and handed to its handler on every delivery, capped at **4 KB** and rejected over that with `events_context_too_large`. Listings never return it. An app sees and revokes only the subscriptions it created; a session acting for the account sees them all, including ones left behind by an app that has since been removed.
+A durable subscription may carry a `context`: JSON that is stored with it and handed to its handler on every delivery, capped at a hard **4 KB** and rejected over that with `events_context_too_large` — client-side, before the request. It is stored in plaintext and read only on the delivery path; listings return its **key names and a content hash**, never its values. For anything larger, store it in a file and put the path in `context`. An app sees and revokes only the subscriptions it created; a session acting for the account sees them all, including ones left behind by an app that has since been removed.
-**A subscription can end without you unsubscribing.** Access is re-checked against the stored permission on every delivery, so a share that is taken back stops delivering immediately; the subscription is then *suspended*, with `suspendedAt` and `suspendedReason: 'permission_revoked'` in `list` and a notification to whoever holds it. The same happens to every subscription an app holds for you when you withdraw that app's access. Re-granting does not bring a suspended subscription back — subscribe again, which is how consent to watch is re-established — and a suspended row is deleted **30 days** after it stops. Deleting the node a subscription is anchored on ends it too, unless the subject named a path or a pattern, in which case it follows that path up to the nearest folder that still exists and keeps watching, so recreating the path resumes delivery.
+A durable subscription runs a **handler** its app published by name. An app may publish **100** of them, each up to **64 KB** of source, and a name is unique inside one app. Publishing is a developer operation: the account has to own the app. Publishing the same source again is a no-op; publishing different source under a name whose current source the caller did not name as its base is refused with `events_handler_conflict`, so two racing build steps never silently pick a winner — `replace: true` is how a caller says it means to take the name. Handler source is never returned by any listing.
+
+**A subscription can end or stop without you unsubscribing.** Access is re-checked against the stored permission on every delivery, so a share that is taken back stops delivering immediately; the subscription is then *suspended*, with `suspendedAt` and `suspendedReason` in `list`. There are four reasons:
+
+| `suspendedReason` | Cause | Resumes when |
+| --- | --- | --- |
+| `handler_not_found` | The handler it is bound to was removed | The name is published again |
+| `failures` | Its handler failed or timed out repeatedly | The subscription is republished against a working handler |
+| `no_credit` | Its holder ran out of credit | The balance is restored |
+| `permission_revoked` | The grant it was made under was withdrawn | **Never** — subscribe again |
+
+A suspended subscription stops delivering and stops being metered, so it cannot go on holding a full backlog for free: what it is owed is trimmed to **100** deliveries and given a deadline — **24 hours** for `handler_not_found` and `failures`, **1 hour** for `no_credit` — after which they are dropped and one `gap` marker with `reason: 'suspended_backlog_expired'` takes their place. A subscription suspended by `permission_revoked` has its backlog **purged immediately**: it names paths its holder has just lost the right to see, and holding them for a resume that by design never comes would turn a revocation into a delayed disclosure. A suspended row itself is deleted **30 days** after it stops.
+
+Deleting the node a subscription is anchored on ends it too, unless the subject named a path or a pattern, in which case it follows that path up to the nearest folder that still exists and keeps watching, so recreating the path resumes delivery.
Match patterns are compiled once when you subscribe and are capped at **256 characters** and **16 segments**; anything larger is rejected with `invalid_subject_pattern`. `**` crosses directories and costs no more than `*`.
diff --git a/src/docs/src/sidebar.js b/src/docs/src/sidebar.js
index 11ad84870..11e4ebc39 100755
--- a/src/docs/src/sidebar.js
+++ b/src/docs/src/sidebar.js
@@ -422,6 +422,38 @@ let sidebar = [
source: '/Events/off.md',
path: '/Events/off',
},
+ {
+ title: 'onPersistent()',
+ page_title: 'puter.events.onPersistent()',
+ title_tag: 'puter.events.onPersistent()',
+ icon: '/assets/img/function.svg',
+ source: '/Events/onPersistent.md',
+ path: '/Events/onPersistent',
+ },
+ {
+ title: 'list()',
+ page_title: 'puter.events.list()',
+ title_tag: 'puter.events.list()',
+ icon: '/assets/img/function.svg',
+ source: '/Events/list.md',
+ path: '/Events/list',
+ },
+ {
+ title: 'unsubscribe()',
+ page_title: 'puter.events.unsubscribe()',
+ title_tag: 'puter.events.unsubscribe()',
+ icon: '/assets/img/function.svg',
+ source: '/Events/unsubscribe.md',
+ path: '/Events/unsubscribe',
+ },
+ {
+ title: 'handlers',
+ page_title: 'puter.events.handlers',
+ title_tag: 'puter.events.handlers',
+ icon: '/assets/img/function.svg',
+ source: '/Events/handlers.md',
+ path: '/Events/handlers',
+ },
],
},
{
diff --git a/src/puter-js/index.d.ts b/src/puter-js/index.d.ts
index e54454d79..188e37198 100644
--- a/src/puter-js/index.d.ts
+++ b/src/puter-js/index.d.ts
@@ -98,11 +98,18 @@ export type {
EventDelivery,
EventGapMarker,
EventHandler,
+ HandlerOptions,
+ HandlerPublication,
+ HandlerSummary,
OnLocalOptions,
+ OnPersistentOptions,
+ PersistentSubscription,
+ PublishedHandler,
PuterEvent,
PuterKvEvent,
} from './types/modules/events/types.js';
export type { EventSubscription } from './types/modules/events/lib/subscription.js';
+export type { EventHandlers } from './types/modules/events/lib/handlers.js';
// -- puter.fs --
export type {
diff --git a/src/puter-js/src/modules/events/index.js b/src/puter-js/src/modules/events/index.js
index 3fe87f48f..750254512 100644
--- a/src/puter-js/src/modules/events/index.js
+++ b/src/puter-js/src/modules/events/index.js
@@ -1,6 +1,10 @@
import { PuterModule } from '../../lib/PuterModule.js';
import { EventChannel } from './lib/channel.js';
+import { EventHandlers } from './lib/handlers.js';
+import { list } from './list.js';
import { onLocal } from './onLocal.js';
+import { onPersistent } from './onPersistent.js';
+import { unsubscribe } from './unsubscribe.js';
/** @typedef {import('../../index.js').Puter} Puter */
@@ -9,15 +13,23 @@ import { onLocal } from './onLocal.js';
* path that does not exist yet — and a handler runs whenever something under
* it changes.
*
+ * Two kinds of subscription: `onLocal()` lives with this connection and is
+ * gone when the page is, while `onPersistent()` is stored against the account
+ * and keeps matching with nothing open. A persistent subscription runs a
+ * handler the app published through `puter.events.handlers`.
+ *
* 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.
+ // 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.
onLocal = onLocal;
+ onPersistent = onPersistent;
+ unsubscribe = unsubscribe;
+ list = list;
/** @param {Puter} puter */
constructor (puter) {
@@ -29,10 +41,15 @@ export class EventsModule extends PuterModule {
*/
this.channel = new EventChannel(this);
+ /** The named functions this app has deployed. */
+ this.handlers = new EventHandlers(this);
+
const methods = /** @type {Record unknown>} */ (
/** @type {unknown} */ (this)
);
- methods.onLocal = methods.onLocal.bind(this);
+ for ( const name of ['onLocal', 'onPersistent', 'unsubscribe', 'list'] ) {
+ methods[name] = methods[name].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
diff --git a/src/puter-js/src/modules/events/lib/api.js b/src/puter-js/src/modules/events/lib/api.js
new file mode 100644
index 000000000..ead31617c
--- /dev/null
+++ b/src/puter-js/src/modules/events/lib/api.js
@@ -0,0 +1,64 @@
+import { fetchUrl } from '../../../lib/networkUtils.js';
+import { PuterJSError } from '../../../lib/PuterJSError.js';
+
+/**
+ * The HTTP half of `puter.events`. The socket verbs carry session
+ * subscriptions; everything that outlives a connection — durable
+ * subscriptions and the handlers they bind — is a route.
+ *
+ * The server's `{ message, code }` is passed through untouched: its codes are
+ * the API surface callers branch on, and re-wrapping them here would make the
+ * SDK a second place they are defined.
+ */
+
+/** The failure shape for a response that carried no usable body. */
+const requestFailed = (status) =>
+ new PuterJSError(
+ `The events request failed (HTTP ${status})`,
+ 'events_failed',
+ );
+
+/**
+ * @param {import('../../../index.js').Puter} puter
+ * @param {string} route
+ * @param {Record} [body] Present makes it a POST.
+ * @param {Record} [query]
+ * @returns {Promise>}
+ */
+export async function request (puter, route, body, query) {
+ const search = new URLSearchParams();
+ for ( const [key, value] of Object.entries(query ?? {}) ) {
+ if ( value === undefined || value === null ) continue;
+ search.set(key, String(value));
+ }
+ const suffix = search.toString();
+
+ const response = await fetchUrl(
+ `${puter.APIOrigin}${route}${suffix ? `?${suffix}` : ''}`,
+ {
+ method: body ? 'POST' : 'GET',
+ includePuterAuth: true,
+ headers: { 'Content-Type': 'application/json' },
+ ...(body ? { body: JSON.stringify(body) } : {}),
+ },
+ );
+
+ const isJson = response.headers.get('content-type')?.includes('application/json');
+ const parsed = isJson ? await response.json() : null;
+
+ if ( response.status !== 200 ) {
+ if ( ! parsed ) throw requestFailed(response.status);
+ const { message, error, code, ...rest } = parsed;
+ throw new PuterJSError(
+ typeof message === 'string'
+ ? message
+ : typeof error === 'string'
+ ? error
+ : `The events request failed (HTTP ${response.status})`,
+ typeof code === 'string' ? code : 'events_failed',
+ rest,
+ );
+ }
+
+ return parsed ?? {};
+}
diff --git a/src/puter-js/src/modules/events/lib/channel.js b/src/puter-js/src/modules/events/lib/channel.js
index 676ec3836..a2f155d25 100644
--- a/src/puter-js/src/modules/events/lib/channel.js
+++ b/src/puter-js/src/modules/events/lib/channel.js
@@ -328,7 +328,10 @@ export class EventChannel {
// 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 | PuterKvEvent | EventGapMarker} */ (envelope.event));
+ sub.deliver(
+ /** @type {PuterEvent | PuterKvEvent | EventGapMarker} */ (envelope.event),
+ /** @type {Record | undefined} */ (envelope.ctx),
+ );
}
/**
diff --git a/src/puter-js/src/modules/events/lib/freeVariables.js b/src/puter-js/src/modules/events/lib/freeVariables.js
new file mode 100644
index 000000000..ae47d3c2d
--- /dev/null
+++ b/src/puter-js/src/modules/events/lib/freeVariables.js
@@ -0,0 +1,311 @@
+import { PuterJSError } from '../../../lib/PuterJSError.js';
+import { tokenize } from './tokenize.js';
+
+/**
+ * The free-variable scan a handler is held to at subscribe time.
+ *
+ * A handler is deployed, not called: it is serialized with
+ * `Function.prototype.toString()` and run later, somewhere else, with nothing
+ * around it. A closed-over variable is therefore not discouraged, it is
+ * unrepresentable — so anything the source names that it does not also bind has
+ * to come from `ctx`, from a parameter, or from the runtime. Catching that here
+ * turns a rule that would otherwise fail on the first delivery, in production,
+ * into a rejected `subscribe`.
+ *
+ * The scan collects every name the source *binds* anywhere — parameters,
+ * destructured names, `var`/`let`/`const`/`function`/`class`/`catch` — and then
+ * requires every identifier *reference* to be one of those or a known global.
+ *
+ * Known limitation: bindings are collected flat rather than per scope, so a
+ * name bound in one block counts as bound in the whole handler. That direction
+ * is deliberate — it can miss a shadowing case, and it never rejects code that
+ * would have worked.
+ */
+
+/**
+ * Reserved words and the contextual keywords that read as identifiers. Skipped
+ * rather than resolved: none of them is a variable reference, and treating
+ * `async` or `get` as one would reject perfectly ordinary handlers.
+ */
+const KEYWORDS = new Set([
+ 'await', 'break', 'case', 'catch', 'class', 'const', 'continue', 'debugger',
+ 'default', 'delete', 'do', 'else', 'enum', 'export', 'extends', 'false',
+ 'finally', 'for', 'function', 'if', 'import', 'in', 'instanceof', 'let',
+ 'new', 'null', 'return', 'super', 'switch', 'this', 'throw', 'true', 'try',
+ 'typeof', 'var', 'void', 'while', 'with', 'yield',
+ 'async', 'as', 'from', 'get', 'set', 'of', 'static', 'accessor',
+]);
+
+/** Declaration keywords whose head is a binding pattern. */
+const DECLARATORS = new Set(['var', 'let', 'const']);
+
+/**
+ * Names the runtime provides. Curated rather than derived from `globalThis`:
+ * the handler runs in a worker isolate, not in the environment doing the scan,
+ * so what is present here says nothing about what is present there.
+ */
+export const HANDLER_GLOBALS = new Set([
+ // Language
+ 'globalThis', 'undefined', 'NaN', 'Infinity', 'arguments',
+ 'Object', 'Array', 'String', 'Number', 'Boolean', 'Symbol', 'BigInt',
+ 'Math', 'JSON', 'Date', 'RegExp', 'Function', 'Promise', 'Proxy', 'Reflect',
+ 'Map', 'Set', 'WeakMap', 'WeakSet', 'WeakRef', 'FinalizationRegistry',
+ 'Error', 'TypeError', 'RangeError', 'SyntaxError', 'ReferenceError',
+ 'EvalError', 'URIError', 'AggregateError', 'Intl',
+ 'ArrayBuffer', 'SharedArrayBuffer', 'DataView',
+ 'Int8Array', 'Uint8Array', 'Uint8ClampedArray', 'Int16Array', 'Uint16Array',
+ 'Int32Array', 'Uint32Array', 'Float32Array', 'Float64Array',
+ 'BigInt64Array', 'BigUint64Array',
+ 'parseInt', 'parseFloat', 'isNaN', 'isFinite',
+ 'encodeURI', 'encodeURIComponent', 'decodeURI', 'decodeURIComponent',
+ 'structuredClone', 'queueMicrotask', 'atob', 'btoa',
+ // Runtime
+ 'console', 'fetch', 'Request', 'Response', 'Headers', 'FormData', 'Blob',
+ 'File', 'URL', 'URLSearchParams', 'AbortController', 'AbortSignal',
+ 'TextEncoder', 'TextDecoder', 'ReadableStream', 'WritableStream',
+ 'TransformStream', 'CompressionStream', 'DecompressionStream',
+ 'crypto', 'Crypto', 'SubtleCrypto', 'performance', 'WebSocket',
+ 'Event', 'EventTarget', 'CustomEvent', 'MessageChannel', 'MessagePort',
+ 'setTimeout', 'clearTimeout', 'setInterval', 'clearInterval',
+ // The SDK the worker runs inside.
+ 'puter',
+]);
+
+/** Raised for the identifier that could not be resolved, naming it. */
+const freeVariable = (name) =>
+ new PuterJSError(
+ `Handler refers to \`${name}\`, which is not a parameter, a local, or a known global. ` +
+ 'A handler is serialized and run elsewhere, so it cannot close over anything — ' +
+ 'pass the value in `context` and read it from `ctx`.',
+ 'events_handler_free_variable',
+ );
+
+const isName = (token) => token?.type === 'name';
+const isPunct = (token, value) => token?.type === 'punct' && token.value === value;
+
+const OPENERS = { '(': ')', '[': ']', '{': '}' };
+const CLOSERS = new Set([')', ']', '}']);
+
+/** Index of the token closing the group that opens at `start`, or -1. */
+const matchGroup = (tokens, start) => {
+ const stack = [];
+ for ( let i = start; i < tokens.length; i++ ) {
+ const token = tokens[i];
+ if ( token.type !== 'punct' ) continue;
+ if ( OPENERS[token.value] ) { stack.push(OPENERS[token.value]); continue; }
+ if ( ! CLOSERS.has(token.value) ) continue;
+ if ( stack.pop() !== token.value ) return -1;
+ if ( stack.length === 0 ) return i;
+ }
+ return -1;
+};
+
+/**
+ * Collect the names a binding pattern introduces, between `start` and `end`.
+ * Everything after an `=` is an initializer — a reference, not a binding — so
+ * it is skipped until the comma that ends that binder.
+ */
+const collectPattern = (tokens, start, end, into) => {
+ let depth = 0;
+ let inInitializer = false;
+ for ( let i = start; i < end; i++ ) {
+ const token = tokens[i];
+ if ( token.type === 'punct' ) {
+ if ( OPENERS[token.value] ) depth++;
+ else if ( CLOSERS.has(token.value) ) depth--;
+ else if ( token.value === '=' ) inInitializer = true;
+ else if ( token.value === ',' && depth <= 0 ) inInitializer = false;
+ continue;
+ }
+ if ( inInitializer || ! isName(token) || KEYWORDS.has(token.value) ) continue;
+ // `.b` in a pattern is a member target, which binds nothing new.
+ if ( isPunct(tokens[i - 1], '.') ) continue;
+ into.add(token.value);
+ }
+};
+
+/** Names a `var`/`let`/`const` head introduces, and where the head ends. */
+const collectDeclaration = (tokens, start, into) => {
+ let depth = 0;
+ let i = start;
+ let inInitializer = false;
+ for ( ; i < tokens.length; i++ ) {
+ const token = tokens[i];
+ if ( token.type === 'punct' ) {
+ if ( OPENERS[token.value] ) { depth++; continue; }
+ if ( CLOSERS.has(token.value) ) {
+ if ( depth === 0 ) return i;
+ depth--;
+ continue;
+ }
+ if ( depth > 0 ) {
+ if ( token.value === '=' ) inInitializer = true;
+ else if ( token.value === ',' ) inInitializer = false;
+ continue;
+ }
+ if ( token.value === ';' ) return i;
+ if ( token.value === '=' ) inInitializer = true;
+ else if ( token.value === ',' ) inInitializer = false;
+ continue;
+ }
+ if ( isName(token) && depth === 0 && (token.value === 'of' || token.value === 'in') )
+ return i;
+ if ( inInitializer || ! isName(token) || KEYWORDS.has(token.value) ) continue;
+ if ( isPunct(tokens[i - 1], '.') ) continue;
+ into.add(token.value);
+ }
+ return i;
+};
+
+/**
+ * Every name the source binds, wherever it binds it. Over-approximate on
+ * purpose: the alternative is a scope tree, and the cost of getting one wrong
+ * is rejecting a handler that works.
+ *
+ * @param {import('./tokenize.js').Token[]} tokens
+ * @returns {Set}
+ */
+export const collectBindings = (tokens) => {
+ /** @type {Set} */
+ const bound = new Set();
+
+ for ( let i = 0; i < tokens.length; i++ ) {
+ const token = tokens[i];
+
+ if ( isPunct(token, '=>') ) {
+ const before = tokens[i - 1];
+ if ( isPunct(before, ')') ) {
+ // Walk back to the `(` this `)` closes.
+ let depth = 0;
+ for ( let j = i - 1; j >= 0; j-- ) {
+ const back = tokens[j];
+ if ( back.type !== 'punct' ) continue;
+ if ( CLOSERS.has(back.value) ) depth++;
+ else if ( OPENERS[back.value] ) {
+ depth--;
+ if ( depth === 0 ) {
+ collectPattern(tokens, j + 1, i - 1, bound);
+ break;
+ }
+ }
+ }
+ } else if ( isName(before) && ! KEYWORDS.has(before.value) ) {
+ bound.add(before.value);
+ }
+ continue;
+ }
+
+ if ( ! isName(token) ) continue;
+
+ if ( DECLARATORS.has(token.value) ) {
+ i = collectDeclaration(tokens, i + 1, bound) - 1;
+ continue;
+ }
+
+ if ( token.value === 'function' || token.value === 'class' ) {
+ const next = tokens[i + 1];
+ // `function *gen()` and `function ()` both leave the name absent.
+ const nameAt = isPunct(next, '*') ? i + 2 : i + 1;
+ if ( isName(tokens[nameAt]) && ! KEYWORDS.has(tokens[nameAt].value) )
+ bound.add(tokens[nameAt].value);
+ continue;
+ }
+
+ if ( token.value === 'catch' && isPunct(tokens[i + 1], '(') ) {
+ const close = matchGroup(tokens, i + 1);
+ if ( close !== -1 ) collectPattern(tokens, i + 2, close, bound);
+ continue;
+ }
+
+ // `name(...) {` is a function or method definition — every construct
+ // that reads the same way (`if`, `for`, `while`, `switch`, `catch`) is
+ // a keyword and never reaches here. Its parameters are bindings, and so
+ // is the name itself.
+ if ( ! KEYWORDS.has(token.value) && isPunct(tokens[i + 1], '(') ) {
+ const close = matchGroup(tokens, i + 1);
+ if ( close !== -1 && isPunct(tokens[close + 1], '{') ) {
+ bound.add(token.value);
+ collectPattern(tokens, i + 2, close, bound);
+ }
+ continue;
+ }
+ }
+
+ // An anonymous `function (a, b) {`, whose parameters the pass above only
+ // reaches when the function is named.
+ for ( let i = 0; i < tokens.length; i++ ) {
+ if ( ! isName(tokens[i]) || tokens[i].value !== 'function' ) continue;
+ let open = i + 1;
+ while ( open < tokens.length && ! isPunct(tokens[open], '(') ) {
+ if ( isPunct(tokens[open], '{') ) break;
+ open++;
+ }
+ if ( ! isPunct(tokens[open], '(') ) continue;
+ const close = matchGroup(tokens, open);
+ if ( close !== -1 ) collectPattern(tokens, open + 1, close, bound);
+ }
+
+ return bound;
+};
+
+/**
+ * Identifiers the source *reads*, in order and without duplicates. Property
+ * names, keys and labels are not reads: `a.b` reaches `b` through `a`, and only
+ * `a` has to resolve to anything.
+ *
+ * @param {import('./tokenize.js').Token[]} tokens
+ * @returns {string[]}
+ */
+export const collectReferences = (tokens) => {
+ const seen = new Set();
+ /** @type {string[]} */
+ const names = [];
+
+ for ( let i = 0; i < tokens.length; i++ ) {
+ const token = tokens[i];
+ if ( ! isName(token) || KEYWORDS.has(token.value) ) continue;
+
+ const before = tokens[i - 1];
+ const after = tokens[i + 1];
+
+ // `a.b`, `a?.b`, `#private`, and the target of `break`/`continue`.
+ if ( isPunct(before, '.') || isPunct(before, '?.') || isPunct(before, '#') )
+ continue;
+ if ( isName(before) && (before.value === 'break' || before.value === 'continue') )
+ continue;
+ // A property key or a label. Also swallows the middle of a ternary,
+ // which is a name this scan then does not check — the safe direction.
+ if ( isPunct(after, ':') ) continue;
+ // A method or function definition, whose name is not a read.
+ if ( isPunct(after, '(') ) {
+ const close = matchGroup(tokens, i + 1);
+ if ( close !== -1 && isPunct(tokens[close + 1], '{') ) continue;
+ }
+
+ if ( seen.has(token.value) ) continue;
+ seen.add(token.value);
+ names.push(token.value);
+ }
+
+ return names;
+};
+
+/**
+ * Throws for the first identifier a handler names and cannot reach. Returns the
+ * bound names, which is only useful to a test.
+ *
+ * @param {string} source Serialized handler source.
+ * @returns {{ bound: Set, references: string[] }}
+ */
+export const scanHandlerSource = (source) => {
+ const tokens = tokenize(source);
+ const bound = collectBindings(tokens);
+ const references = collectReferences(tokens);
+
+ for ( const name of references ) {
+ if ( bound.has(name) || HANDLER_GLOBALS.has(name) ) continue;
+ throw freeVariable(name);
+ }
+
+ return { bound, references };
+};
diff --git a/src/puter-js/src/modules/events/lib/freeVariables.test.js b/src/puter-js/src/modules/events/lib/freeVariables.test.js
new file mode 100644
index 000000000..cc47c19af
--- /dev/null
+++ b/src/puter-js/src/modules/events/lib/freeVariables.test.js
@@ -0,0 +1,181 @@
+import { describe, expect, it } from 'vitest';
+import { scanHandlerSource } from './freeVariables.js';
+import { tokenize } from './tokenize.js';
+
+const scan = (source) => scanHandlerSource(source);
+
+const rejects = (source) => {
+ try {
+ scan(source);
+ } catch (error) {
+ return error;
+ }
+ throw new Error(`expected a rejection for: ${source}`);
+};
+
+/** Handlers a developer would plausibly write, none of which close over anything. */
+const ACCEPTED = [
+ [
+ 'the design`s example handler',
+ `async ({ event, ctx, user, fetch, ack }) => {
+ await fetch(ctx.endpoint, {
+ method: 'POST',
+ body: JSON.stringify({ path: event.path, key: ctx.apiKey }),
+ });
+ await ack();
+ }`,
+ ],
+ ['a bare arrow with one parameter', 'delivery => console.log(delivery.event.op)'],
+ ['a named function declaration', 'function onWrite ({ event }) { console.log(event.uid); }'],
+ ['an anonymous function expression', 'function ({ event, ctx }) { return ctx.prefix + event.path; }'],
+ ['locals declared with const and let', '({ event }) => { const p = event.path; let n = p.length; return n; }'],
+ ['a destructured local with a default', '({ ctx }) => { const { retries = 3, url } = ctx; return url.repeat(retries); }'],
+ ['an array destructuring local', '({ event }) => { const [head, ...rest] = event.path.split("/"); return rest.concat(head); }'],
+ ['a for-of loop variable', '({ ctx }) => { for (const item of ctx.items) console.log(item); }'],
+ ['a classic for loop', '({ ctx }) => { for (let i = 0; i < ctx.n; i++) console.log(i); }'],
+ ['a catch parameter', '({ ctx }) => { try { JSON.parse(ctx.body); } catch (err) { console.warn(err); } }'],
+ ['a nested function and its parameters', '({ event }) => { const f = (a, b) => a + b; return f(1, event.seq); }'],
+ ['a class declaration with methods', '({ ctx }) => { class Sink { constructor (url) { this.url = url; } send (body) { return fetch(this.url, { body }); } } return new Sink(ctx.url); }'],
+ ['object property keys that share a name with nothing', '({ event }) => ({ endpoint: event.path, retries: 2 })'],
+ ['a template literal reading only ctx', '({ ctx, event }) => `${ctx.base}/${event.uid}`'],
+ ['a regex literal that looks like division', '({ event }) => /\\/tmp\\/[a-z]+/.test(event.path)'],
+ ['a comment naming something undeclared', '({ event }) => { /* endpoint is gone now */ return event.uid; }'],
+ ['a string naming something undeclared', '({ event }) => event.path + "endpoint"'],
+ ['runtime globals', '({ event }) => { console.log(Date.now(), Math.max(1, event.seq), JSON.stringify(event), new URL("https://x.example")); }'],
+ ['the SDK global a worker runs inside', '({ user }) => user.puter.fs.read("/x").then(r => puter.print(r))'],
+ ['optional chaining and computed member access', '({ event, ctx }) => event?.meta?.[ctx.key]'],
+ ['a shorthand method on an object literal', '({ event }) => ({ run (x) { return x + event.seq; } })'],
+ ['an async generator with a yield', 'async function* ({ ctx }) { yield ctx.first; }'],
+ ['a label and a break to it', '({ ctx }) => { outer: for (const a of ctx.rows) { break outer; } }'],
+ ['a label and a continue to it', '({ ctx, event }) => { loop: while (ctx.n-- > 0) { if (ctx.skip) continue loop; event.push(ctx.n); } }'],
+ ['a getter on a class', '({ ctx }) => { class C { get url () { return ctx.url; } } return new C(); }'],
+ [
+ 'the design doc`s own example, verbatim',
+ `async ({ event, ctx, user, fetch, ack }) => {
+ const meta = await user.fs.stat(event.path);
+ if (meta.size < ctx.minSize) return ack();
+ await fetch(ctx.endpoint, {
+ method: 'POST',
+ body: JSON.stringify({ uid: event.uid, size: meta.size }),
+ });
+ await ack();
+ }`,
+ ],
+ ['object shorthand naming a declared local', '({ event }) => { const endpoint = event.path; return { endpoint }; }'],
+ ['rest in a destructured object parameter', 'async ({ event, ...rest }) => { return rest.foo + event.seq; }'],
+ ['typeof on a bound parameter', '({ event }) => typeof event === "object"'],
+ // Regex directly after a block-closing `}`, with no `return`/other
+ // regex-triggering keyword in between — the tokenizer has to decide this
+ // is a regex from the `}` alone, not from what came before it.
+ [
+ 'a regex literal right after a closed block, not division',
+ '({ ctx }) => { if (ctx.on) { console.log(ctx.on); } /ab+c/.test(ctx.body); }',
+ ],
+];
+
+/** Handlers that close over something the serialized source cannot carry. */
+const REJECTED = [
+ ['a closure over an outer const', '({ event }) => fetch(endpoint, { body: event.path })', 'endpoint'],
+ ['a closure used as a bare value', '({ event }) => event.path + suffix', 'suffix'],
+ ['a closure inside a template hole', '({ event }) => `${base}/${event.uid}`', 'base'],
+ ['a closure inside a nested function', '({ event }) => { const f = () => apiKey; return f(); }', 'apiKey'],
+ ['a closure used as a call target', '({ event }) => publish(event)', 'publish'],
+ ['a closure in a default parameter value', '({ event }, retries = maxRetries) => retries + event.seq', 'maxRetries'],
+ ['a closure in a destructuring default', '({ event, timeout = defaultTimeout }) => timeout + event.seq', 'defaultTimeout'],
+ ['a closure in a for-of subject', '() => { for (const row of rows) console.log(row); }', 'rows'],
+ ['a closure used with new', '({ ctx }) => new Sink(ctx.url)', 'Sink'],
+ ['a closure in a declaration initializer', '({ event }) => { const target = destination; return target + event.uid; }', 'destination'],
+ ['a closure in an object value position', '({ event }) => ({ endpoint: outerEndpoint, path: event.path })', 'outerEndpoint'],
+ ['a closure in a computed key', '({ event }) => ({ [outerKey]: event.uid })', 'outerKey'],
+ // Shorthand `{ endpoint }` is sugar for `{ endpoint: endpoint }` — a
+ // *reference*, not a key — and has to be told apart from `{ endpoint: x }`
+ // above, where `endpoint` is a label nothing needs to resolve.
+ ['a closure read through object shorthand', '({ event }) => ({ endpoint, path: event.path })', 'endpoint'],
+ ['typeof on an undeclared name', '({ event }) => typeof missingGlobal === "undefined" ? event.seq : 0', 'missingGlobal'],
+];
+
+describe('handlers a scan accepts', () => {
+ it.each(ACCEPTED)('accepts %s', (_label, source) => {
+ expect(() => scan(source)).not.toThrow();
+ });
+});
+
+describe('handlers a scan rejects', () => {
+ it.each(REJECTED)('rejects %s', (_label, source, identifier) => {
+ const error = rejects(source);
+ expect(error.code).toBe('events_handler_free_variable');
+ expect(error.message).toContain(`\`${identifier}\``);
+ });
+
+ it('names the identifier so the developer knows what to move into context', () => {
+ const error = rejects('({ event }) => fetch(ingestUrl, { body: event.path })');
+ expect(error.message).toContain('`ingestUrl`');
+ expect(error.message).toContain('ctx');
+ });
+});
+
+describe('what the tokenizer hides from the scan', () => {
+ it('drops strings, comments and regex bodies', () => {
+ const values = tokenize(
+ '({ a }) => { /* comment */ const s = "text"; return /pattern/.test(s) && a; }',
+ ).map(token => token.value);
+
+ expect(values).not.toContain('comment');
+ expect(values).not.toContain('text');
+ expect(values).not.toContain('pattern');
+ expect(values).toContain('a');
+ });
+
+ it('keeps the code inside a template hole', () => {
+ const values = tokenize('`prefix ${value} suffix`').map(token => token.value);
+ expect(values).toContain('value');
+ expect(values).not.toContain('prefix');
+ expect(values).not.toContain('suffix');
+ });
+
+ it('reads a nested template inside a hole', () => {
+ expect(() => scan('({ ctx }) => `${`${ctx.a}`}`')).not.toThrow();
+ expect(rejects('({ ctx }) => `${`${nested}`}`').message).toContain('`nested`');
+ });
+
+ it('does not mistake division for a regex', () => {
+ expect(() => scan('({ ctx }) => (ctx.a + ctx.b) / 2')).not.toThrow();
+ });
+});
+
+/**
+ * Known misses, not bugs: the scan collects bindings flat rather than
+ * per-scope and treats a name before `:` as a label/key rather than a
+ * reference (see the module doc). Both directions only ever *accept* code
+ * that closes over something real — they never reject code that would have
+ * worked, which is the safe side to be wrong on. Pinned here so a future
+ * tightening of the scan is a deliberate choice, not an accidental one.
+ */
+describe('known accept-biased misses (documented, not fixed)', () => {
+ it('does not resolve the truthy arm of a ternary, so a free name there slips through', () => {
+ // `freeVar` sits directly before the ternary`s `:` and reads the same
+ // as a label, so the scan skips it — even though it is a real,
+ // undeclared reference here.
+ expect(() => scan('({ event }) => event.ok ? freeVar : event.seq')).not.toThrow();
+ });
+
+ it('over-binds a destructuring rename`s source key', () => {
+ // `{ event: renamed }` binds only `renamed` — `event` is the property
+ // being read off the parameter, not a local. The scan collects every
+ // name in a pattern as bound, so it treats `event` as available too,
+ // and a bare reference to it below is not caught even though it would
+ // be a ReferenceError at runtime.
+ expect(() =>
+ scan('({ event: renamed }) => { return renamed.x + event; }'),
+ ).not.toThrow();
+ });
+});
+
+describe('what the scan reports back', () => {
+ it('lists what the source binds and what it reads', () => {
+ const { bound, references } = scan('({ event, ctx }) => { const n = ctx.n; return event.seq + n; }');
+
+ expect([...bound].sort()).toEqual(['ctx', 'event', 'n']);
+ expect(references).toEqual(['event', 'ctx', 'n']);
+ });
+});
diff --git a/src/puter-js/src/modules/events/lib/handlerSource.js b/src/puter-js/src/modules/events/lib/handlerSource.js
new file mode 100644
index 000000000..ba08900b2
--- /dev/null
+++ b/src/puter-js/src/modules/events/lib/handlerSource.js
@@ -0,0 +1,151 @@
+import { PuterJSError } from '../../../lib/PuterJSError.js';
+import { scanHandlerSource } from './freeVariables.js';
+
+/**
+ * Turning what a developer wrote into what gets deployed.
+ *
+ * A handler is not called where it is written — it is serialized, stored, and
+ * run later in the app's events worker. So the three accepted forms all reduce
+ * to one string, that string is scanned for anything it cannot carry with it,
+ * and its hash goes along so the server can tell whether the code a
+ * subscription was written against is still what is published.
+ */
+
+/** Hard cap on a subscription's serialized `context`, matching the column. */
+export const CONTEXT_MAX_BYTES = 4096;
+
+const invalidHandler = (message) =>
+ new PuterJSError(message, 'events_handler_invalid');
+
+const contextTooLarge = () =>
+ new PuterJSError(
+ `Subscription context may not exceed ${CONTEXT_MAX_BYTES} bytes`,
+ 'events_context_too_large',
+ );
+
+/**
+ * Hashing is `crypto.subtle`, which an insecure browser origin does not
+ * provide. Failing loudly beats binding a subscription to whatever happens to
+ * be published under the name.
+ */
+const hashUnavailable = () =>
+ new PuterJSError(
+ 'This environment provides no `crypto.subtle`, so an inline handler cannot be ' +
+ 'hashed. Publish it with `puter.events.handlers.publish()` and subscribe with ' +
+ '`handlerName` instead.',
+ 'events_handler_hash_unavailable',
+ );
+
+const encoder = new TextEncoder();
+
+/** Bytes a string takes on the wire, which is what every cap is measured in. */
+export const byteLength = (text) => encoder.encode(text).length;
+
+/**
+ * SHA-256 of the source, hex, matching what the server stores. Async because
+ * `crypto.subtle` is, and it is the only digest all three runtimes share.
+ *
+ * @param {string} source
+ * @returns {Promise}
+ */
+export const hashSource = async (source) => {
+ const subtle = globalThis.crypto?.subtle;
+ if ( ! subtle ) throw hashUnavailable();
+ const digest = await subtle.digest('SHA-256', encoder.encode(source));
+ return [...new Uint8Array(digest)]
+ .map(byte => byte.toString(16).padStart(2, '0'))
+ .join('');
+};
+
+/**
+ * The source of a handler given as a function or a source string. A
+ * `{ file }` form is read separately, because reading is asynchronous and
+ * everything else here is not.
+ *
+ * @param {unknown} handler
+ * @returns {string | null} `null` when the handler is a `{ file }` reference.
+ */
+export const sourceOf = (handler) => {
+ if ( typeof handler === 'function' ) return Function.prototype.toString.call(handler);
+ if ( typeof handler === 'string' ) {
+ if ( handler.trim().length === 0 )
+ throw invalidHandler('A handler source string may not be empty');
+ return handler;
+ }
+ if ( handler && typeof handler === 'object' && 'file' in handler ) return null;
+ throw invalidHandler(
+ 'A handler must be a function, a source string, or `{ file: }`',
+ );
+};
+
+/**
+ * Resolve a handler to its source, reading a `{ file }` reference through the
+ * caller's own filesystem.
+ *
+ * File references resolve **at this call**, not at delivery: what is deployed
+ * is the bytes as they were when the handler was published or subscribed, so
+ * editing the file afterwards changes nothing until it is published again.
+ *
+ * @param {import('../../../index.js').Puter} puter
+ * @param {unknown} handler
+ * @returns {Promise}
+ */
+export const resolveSource = async (puter, handler) => {
+ const inline = sourceOf(handler);
+ if ( inline !== null ) return inline;
+
+ const path = /** @type {{ file: unknown }} */ (handler).file;
+ if ( typeof path !== 'string' || path.trim().length === 0 )
+ throw invalidHandler('`file` must be a non-empty path');
+
+ const blob = await puter.fs.read(path);
+ const source = typeof blob === 'string' ? blob : await blob.text();
+ if ( source.trim().length === 0 )
+ throw invalidHandler(`\`${path}\` is empty`);
+ return source;
+};
+
+/**
+ * Everything the wire needs about a handler: its source, its hash, and the
+ * guarantee that it names nothing it cannot carry.
+ *
+ * @param {import('../../../index.js').Puter} puter
+ * @param {unknown} handler
+ * @returns {Promise<{ source: string, hash: string }>}
+ */
+export const prepareHandler = async (puter, handler) => {
+ const source = await resolveSource(puter, handler);
+ scanHandlerSource(source);
+ return { source, hash: await hashSource(source) };
+};
+
+/**
+ * The `context` a subscription carries, serialized and checked against the cap
+ * before the request rather than after it.
+ *
+ * Evaluated **now**: `ctx` is a snapshot of these values as they are at
+ * subscribe time, and it never changes again for the life of the subscription.
+ *
+ * @param {unknown} context
+ * @returns {string | undefined}
+ */
+export const serializeContext = (context) => {
+ if ( context === undefined || context === null ) return undefined;
+
+ let json;
+ try {
+ json = JSON.stringify(context);
+ } catch {
+ throw new PuterJSError(
+ 'context must be JSON-serializable',
+ 'events_context_invalid',
+ );
+ }
+ if ( json === undefined )
+ throw new PuterJSError(
+ 'context must be JSON-serializable',
+ 'events_context_invalid',
+ );
+ if ( byteLength(json) > CONTEXT_MAX_BYTES ) throw contextTooLarge();
+ return json;
+};
diff --git a/src/puter-js/src/modules/events/lib/handlers.js b/src/puter-js/src/modules/events/lib/handlers.js
new file mode 100644
index 000000000..f61d5369c
--- /dev/null
+++ b/src/puter-js/src/modules/events/lib/handlers.js
@@ -0,0 +1,172 @@
+import { PuterJSError } from '../../../lib/PuterJSError.js';
+import { request } from './api.js';
+import { prepareHandler } from './handlerSource.js';
+
+/** @typedef {import('../types.js').PublishedHandler} PublishedHandler */
+/** @typedef {import('../types.js').HandlerSummary} HandlerSummary */
+/** @typedef {import('../types.js').HandlerOptions} HandlerOptions */
+/** @typedef {import('../types.js').HandlerPublication} HandlerPublication */
+
+/**
+ * `puter.events.handlers` — the named functions an app deploys once and its
+ * users' subscriptions bind to.
+ *
+ * Publishing is a developer operation: an app token publishes into its own app,
+ * and a plain session has to name an app it owns. Nothing here triggers a
+ * handler — a name is a label for deployed code, and it runs only when a
+ * subscription bound to it has a delivery.
+ *
+ * Two build steps publishing different source under one name is a race with no
+ * right winner, so this sends the hash it last saw published (`ifHash`) and
+ * lets the server refuse a publish whose base has moved. `replace: true` is how
+ * a caller says it means to take the name regardless.
+ */
+
+/** One name in one app. The same name means different code in another. */
+const baseKey = (appUid, name) => `${appUid ?? ''}|${name}`;
+
+const invalidName = () =>
+ new PuterJSError(
+ 'A handler name must be a non-empty string',
+ 'events_handler_name_invalid',
+ );
+
+export class EventHandlers {
+ /** @param {import('../index.js').EventsModule} module */
+ constructor (module) {
+ /** @internal */
+ this.module = module;
+ /**
+ * @internal The hash last seen published, keyed by app and name — the
+ * base a publish claims it is updating. Empty until this client has
+ * published or listed, which is what makes a first publish
+ * create-or-idempotent. Keyed by app as well as name because one
+ * name means different code in two apps.
+ * @type {Map}
+ */
+ this.known = new Map();
+
+ for ( const name of ['publish', 'publishAll', 'list', 'remove'] ) {
+ this[name] = this[name].bind(this);
+ }
+ }
+
+ /**
+ * Publishes one named handler.
+ *
+ * @param {string} name The name subscriptions bind to.
+ * @param {Function | string | { file: string }} handler The handler: a
+ * function (serialized with `toString()`), its source, or a path to read
+ * it from. A file resolves now, not at delivery.
+ * @param {HandlerOptions} [options]
+ * @returns {Promise}
+ */
+ async publish (name, handler, options = {}) {
+ const [published] = await this.#send(
+ [{ name, handler, replace: options.replace }],
+ options.appUid,
+ '/events/handlers/publish',
+ );
+ return published;
+ }
+
+ /**
+ * Publishes a set of handlers in one call — what a build step has. Items
+ * are taken in order, and one the server refuses stops the pass, so a
+ * deploy never reports success over a half-published set.
+ *
+ * @param {HandlerPublication[]} handlers
+ * @param {HandlerOptions} [options]
+ * @returns {Promise}
+ */
+ async publishAll (handlers, options = {}) {
+ if ( ! Array.isArray(handlers) || handlers.length === 0 ) {
+ throw new PuterJSError(
+ '`handlers` must be a non-empty array',
+ 'invalid_request',
+ );
+ }
+ return this.#send(handlers, options.appUid, '/events/handlers/publishAll');
+ }
+
+ /**
+ * What this app has published: names, source hashes, and how many
+ * subscriptions each is carrying. Never the source.
+ *
+ * @param {HandlerOptions} [options]
+ * @returns {Promise}
+ */
+ async list (options = {}) {
+ const response = await request(
+ this.module.puter,
+ '/events/handlers/list',
+ undefined,
+ options.appUid ? { appUid: options.appUid } : undefined,
+ );
+ const handlers = /** @type {HandlerSummary[]} */ (response.handlers ?? []);
+ for ( const handler of handlers )
+ this.known.set(baseKey(options.appUid, handler.name), handler.hash);
+ return handlers;
+ }
+
+ /**
+ * Removes a name. With nothing bound to it the handler simply goes; with
+ * subscriptions on it they are **suspended**, not deleted, and publishing
+ * the name again resumes them.
+ *
+ * @param {string} name
+ * @param {HandlerOptions} [options]
+ * @returns {Promise<{ name: string, removed: boolean, suspended: number }>}
+ */
+ async remove (name, options = {}) {
+ if ( typeof name !== 'string' || name.trim().length === 0 ) throw invalidName();
+ const removed = /** @type {{ name: string, removed: boolean, suspended: number }} */ (
+ await request(this.module.puter, '/events/handlers/remove', {
+ name,
+ ...(options.appUid ? { appUid: options.appUid } : {}),
+ })
+ );
+ this.known.delete(baseKey(options.appUid, name));
+ return removed;
+ }
+
+ /**
+ * @internal Serialize, scan and send one or more publications, then record
+ * what is now published so the next publish can name its base.
+ * @param {HandlerPublication[]} items
+ * @param {string | undefined} appUid
+ * @param {string} route
+ * @returns {Promise}
+ */
+ async #send (items, appUid, route) {
+ const handlers = [];
+ for ( const item of items ) {
+ const name = item?.name;
+ if ( typeof name !== 'string' || name.trim().length === 0 ) throw invalidName();
+
+ const { source } = await prepareHandler(this.module.puter, item.handler);
+ const ifHash = this.known.get(baseKey(appUid, name));
+ handlers.push({
+ name,
+ source,
+ ...(item.replace === true ? { replace: true } : {}),
+ ...(ifHash && item.replace !== true ? { ifHash } : {}),
+ });
+ }
+
+ const body = {
+ ...(appUid ? { appUid } : {}),
+ ...(handlers.length === 1 && route.endsWith('/publish')
+ ? handlers[0]
+ : { handlers }),
+ };
+
+ const response = await request(this.module.puter, route, body);
+ const published = /** @type {PublishedHandler[]} */ (
+ Array.isArray(response.handlers) ? response.handlers : [response]
+ );
+ for ( const handler of published )
+ this.known.set(baseKey(appUid, handler.name), handler.hash);
+ return published;
+ }
+}
diff --git a/src/puter-js/src/modules/events/lib/subscription.js b/src/puter-js/src/modules/events/lib/subscription.js
index dd115908f..81f4dbdc6 100644
--- a/src/puter-js/src/modules/events/lib/subscription.js
+++ b/src/puter-js/src/modules/events/lib/subscription.js
@@ -85,11 +85,16 @@ export class EventSubscription {
/**
* @internal
* @param {PuterEvent | PuterKvEvent | EventGapMarker} event
+ * @param {Record} [ctx] The subscription's stored
+ * context, which the handler must not be able to mutate: it is one
+ * snapshot shared across every delivery.
* @returns {void}
*/
- deliver (event) {
+ deliver (event, ctx) {
try {
- const result = this.handler({ event });
+ const result = this.handler(
+ ctx === undefined ? { event } : { event, ctx: Object.freeze(ctx) },
+ );
if ( result instanceof Promise ) {
result.catch(reportHandlerError);
}
diff --git a/src/puter-js/src/modules/events/lib/subscription.test.js b/src/puter-js/src/modules/events/lib/subscription.test.js
new file mode 100644
index 000000000..3c5e7a917
--- /dev/null
+++ b/src/puter-js/src/modules/events/lib/subscription.test.js
@@ -0,0 +1,49 @@
+import { describe, expect, it, vi } from 'vitest';
+import { EventSubscription } from './subscription.js';
+
+/**
+ * `context` is one snapshot shared across every delivery (R2-16): a handler
+ * that could mutate it would have every later delivery see the mutation, on
+ * every subscriber sharing that context. `deliver()` is where a value crosses
+ * from "stored" to "handed to the developer's code", so it is where the
+ * freeze has to happen.
+ */
+
+const fakeChannel = { remove: vi.fn() };
+
+describe('what a delivery hands the handler', () => {
+ it('freezes ctx before the handler ever sees it', () => {
+ const handler = vi.fn();
+ const sub = new EventSubscription(fakeChannel, 'fs:~/Documents', handler);
+
+ sub.deliver({ id: 'e1', op: 'write' }, { url: 'https://ingest.example' });
+
+ expect(handler).toHaveBeenCalledTimes(1);
+ const [{ ctx }] = handler.mock.calls[0];
+ expect(Object.isFrozen(ctx)).toBe(true);
+ expect(ctx).toEqual({ url: 'https://ingest.example' });
+ });
+
+ it('omits ctx entirely for a subscription that carries none', () => {
+ const handler = vi.fn();
+ const sub = new EventSubscription(fakeChannel, 'fs:~/Documents', handler);
+
+ sub.deliver({ id: 'e1', op: 'write' });
+
+ expect(handler).toHaveBeenCalledWith({ event: { id: 'e1', op: 'write' } });
+ expect('ctx' in handler.mock.calls[0][0]).toBe(false);
+ });
+
+ it('does not let the handler write back into the shared context', () => {
+ const handler = vi.fn((arg) => {
+ expect(() => {
+ arg.ctx.url = 'https://tampered.example';
+ }).toThrow();
+ });
+ const sub = new EventSubscription(fakeChannel, 'fs:~/Documents', handler);
+
+ sub.deliver({ id: 'e1', op: 'write' }, { url: 'https://ingest.example' });
+
+ expect(handler).toHaveBeenCalledTimes(1);
+ });
+});
diff --git a/src/puter-js/src/modules/events/lib/tokenize.js b/src/puter-js/src/modules/events/lib/tokenize.js
new file mode 100644
index 000000000..41e1d55b4
--- /dev/null
+++ b/src/puter-js/src/modules/events/lib/tokenize.js
@@ -0,0 +1,160 @@
+// A tokenizer good enough to tell an identifier *reference* from everything
+// that merely looks like one. It is not a parser: it produces a flat token
+// stream with strings, comments and regex literals removed, and template
+// literals reduced to the tokens inside their `${}` holes.
+//
+// This exists because the handler scan has to run in the browser, in node and
+// in a worker isolate with no parser available and no dependency to add for
+// one. Everything it cannot decide, it decides in the direction that produces a
+// clear error rather than a silent misreading.
+
+/** One token. Strings, comments and regex bodies never reach here. */
+/** @typedef {{ type: 'name' | 'num' | 'punct', value: string }} Token */
+
+const WHITESPACE = /\s/;
+const IDENT_START = /[A-Za-z_$\u00A0-\uFFFF]/;
+const IDENT_PART = /[A-Za-z0-9_$\u00A0-\uFFFF]/;
+const DIGIT = /[0-9]/;
+
+const NUMBER = /^(?:0[xX][0-9a-fA-F_]+|0[bB][01_]+|0[oO][0-7_]+|(?:[0-9][0-9_]*)?\.?[0-9][0-9_]*(?:[eE][+-]?[0-9]+)?|[0-9][0-9_]*\.)n?/;
+
+// Longest first, so `===` is never read as `==` followed by `=`.
+const PUNCTUATORS = [
+ '>>>=', '...', '===', '!==', '**=', '<<=', '>>=', '>>>', '&&=', '||=', '??=',
+ '=>', '==', '!=', '<=', '>=', '&&', '||', '??', '?.', '++', '--',
+ '+=', '-=', '*=', '/=', '%=', '&=', '|=', '^=', '**', '<<', '>>',
+];
+
+/**
+ * After these, a `/` opens a regex rather than dividing. `)` and `]` are
+ * deliberately absent — `(a + b) / 2` is far more common than a regex there —
+ * while `}` is present, because reading a regex as division would tokenize its
+ * body and invent identifiers that were never in the code.
+ */
+const REGEX_AFTER_KEYWORD = new Set([
+ 'return', 'typeof', 'instanceof', 'in', 'of', 'new', 'delete', 'void',
+ 'case', 'do', 'else', 'yield', 'await', 'throw',
+]);
+
+const NO_REGEX_AFTER = new Set([')', ']', '++', '--']);
+
+/**
+ * Splits source into tokens.
+ *
+ * @param {string} source
+ * @returns {Token[]}
+ */
+export const tokenize = (source) => {
+ /** @type {Token[]} */
+ const tokens = [];
+ /** Braces that are `${` holes, so `}` can hand the template back. */
+ const braces = [];
+ let inTemplate = false;
+ let i = 0;
+
+ const push = (type, value) => tokens.push({ type, value });
+ const previous = () => tokens[tokens.length - 1];
+
+ const regexAllowed = () => {
+ const prev = previous();
+ if ( ! prev ) return true;
+ if ( prev.type === 'num' ) return false;
+ if ( prev.type === 'name' ) return REGEX_AFTER_KEYWORD.has(prev.value);
+ return ! NO_REGEX_AFTER.has(prev.value);
+ };
+
+ /** Walk to the end of a quoted string, honouring escapes. */
+ const skipString = (quote) => {
+ i++;
+ while ( i < source.length ) {
+ if ( source[i] === '\\' ) { i += 2; continue; }
+ if ( source[i] === quote ) { i++; return; }
+ i++;
+ }
+ };
+
+ /** Walk to the end of a regex literal, including its character classes. */
+ const skipRegex = () => {
+ i++;
+ let inClass = false;
+ while ( i < source.length ) {
+ const ch = source[i];
+ if ( ch === '\\' ) { i += 2; continue; }
+ if ( ch === '\n' ) return;
+ if ( ch === '[' ) inClass = true;
+ else if ( ch === ']' ) inClass = false;
+ else if ( ch === '/' && ! inClass ) {
+ i++;
+ while ( i < source.length && IDENT_PART.test(source[i]) ) i++;
+ return;
+ }
+ i++;
+ }
+ };
+
+ while ( i < source.length ) {
+ const ch = source[i];
+
+ if ( inTemplate ) {
+ if ( ch === '\\' ) { i += 2; continue; }
+ if ( ch === '`' ) { inTemplate = false; i++; continue; }
+ if ( ch === '$' && source[i + 1] === '{' ) {
+ // The hole is code, and code is what this is here to read.
+ braces.push('template');
+ inTemplate = false;
+ i += 2;
+ continue;
+ }
+ i++;
+ continue;
+ }
+
+ if ( WHITESPACE.test(ch) ) { i++; continue; }
+
+ if ( ch === '/' && source[i + 1] === '/' ) {
+ while ( i < source.length && source[i] !== '\n' ) i++;
+ continue;
+ }
+ if ( ch === '/' && source[i + 1] === '*' ) {
+ const end = source.indexOf('*/', i + 2);
+ i = end === -1 ? source.length : end + 2;
+ continue;
+ }
+ if ( ch === '/' && regexAllowed() ) { skipRegex(); continue; }
+
+ if ( ch === '"' || ch === "'" ) { skipString(ch); continue; }
+ if ( ch === '`' ) { inTemplate = true; i++; continue; }
+
+ if ( ch === '{' ) { braces.push('brace'); push('punct', '{'); i++; continue; }
+ if ( ch === '}' ) {
+ if ( braces.pop() === 'template' ) { inTemplate = true; i++; continue; }
+ push('punct', '}');
+ i++;
+ continue;
+ }
+
+ if ( DIGIT.test(ch) || (ch === '.' && DIGIT.test(source[i + 1] ?? '')) ) {
+ const match = NUMBER.exec(source.slice(i));
+ const text = match ? match[0] : ch;
+ push('num', text);
+ i += text.length;
+ continue;
+ }
+
+ if ( IDENT_START.test(ch) ) {
+ let end = i + 1;
+ while ( end < source.length && IDENT_PART.test(source[end]) ) end++;
+ push('name', source.slice(i, end));
+ i = end;
+ continue;
+ }
+
+ const punct = PUNCTUATORS.find(candidate => source.startsWith(candidate, i));
+ if ( punct ) { push('punct', punct); i += punct.length; continue; }
+
+ push('punct', ch);
+ i++;
+ }
+
+ return tokens;
+};
diff --git a/src/puter-js/src/modules/events/list.js b/src/puter-js/src/modules/events/list.js
new file mode 100644
index 000000000..c6a558934
--- /dev/null
+++ b/src/puter-js/src/modules/events/list.js
@@ -0,0 +1,75 @@
+import { fetchAllPages, iteratePages } from '../../lib/pagination.js';
+import { PuterJSError } from '../../lib/PuterJSError.js';
+import { request } from './lib/api.js';
+
+/** @typedef {import('./types.js').PersistentSubscription} PersistentSubscription */
+/** @typedef {import('../../lib/types.js').ListPage} SubscriptionPage */
+
+/**
+ * @overload
+ * @param {import('../../lib/types.js').ListStreamOptions} options
+ * @returns {AsyncIterableIterator}
+ */
+/**
+ * @overload
+ * @param {import('../../lib/types.js').ListPaginationOptions & ({ cursor: string | null } | { includeTotal: true })} options
+ * @returns {Promise}
+ */
+/**
+ * @overload
+ * @param {{ limit?: number }} [options]
+ * @returns {Promise}
+ */
+/**
+ * Lists the persistent subscriptions this caller holds, page by page under the
+ * hood, resolving to a plain array. Passing any pagination param
+ * (`cursor`/`includeTotal`) switches to a single-request page envelope, and
+ * `stream: true` returns an async iterator of page envelopes.
+ *
+ * An app sees only the subscriptions it created. A session acting for the
+ * account sees them all, including ones left behind by an app that is gone —
+ * which is what makes the account the place a stray subscription is revoked
+ * from. `context` values are never returned; a row reports its key names and a
+ * hash instead.
+ *
+ * @this {import('./index.js').EventsModule}
+ * @param {...unknown} args
+ * @returns {Promise | Promise | AsyncIterableIterator}
+ */
+export function list (...args) {
+ const { puter } = this;
+ const opts = /** @type {Record} */ (
+ typeof args[0] === 'object' && args[0] !== null ? args[0] : {}
+ );
+ const { limit, cursor, includeTotal, stream } = opts;
+ const hasCursor = Object.prototype.hasOwnProperty.call(opts, 'cursor');
+
+ const fetchPage = pageParams =>
+ request(puter, '/events/subscriptions', undefined, {
+ ...(limit !== undefined ? { limit } : {}),
+ ...(pageParams.cursor ? { cursor: pageParams.cursor } : {}),
+ ...(pageParams.includeTotal ? { includeTotal: true } : {}),
+ });
+
+ if ( stream === true ) {
+ return iteratePages(fetchPage, {
+ cursor: /** @type {string | null | undefined} */ (cursor),
+ includeTotal: includeTotal === true,
+ });
+ }
+
+ if ( hasCursor || includeTotal !== undefined ) {
+ if ( includeTotal !== undefined && typeof includeTotal !== 'boolean' ) {
+ throw new PuterJSError(
+ '`includeTotal` must be a boolean',
+ 'invalid_request',
+ );
+ }
+ return fetchPage({
+ cursor: /** @type {string | null} */ (cursor ?? null),
+ includeTotal: includeTotal === true,
+ });
+ }
+
+ return fetchAllPages(fetchPage);
+}
diff --git a/src/puter-js/src/modules/events/onPersistent.js b/src/puter-js/src/modules/events/onPersistent.js
new file mode 100644
index 000000000..7b91bb95e
--- /dev/null
+++ b/src/puter-js/src/modules/events/onPersistent.js
@@ -0,0 +1,62 @@
+import { PuterJSError } from '../../lib/PuterJSError.js';
+import { request } from './lib/api.js';
+import { prepareHandler, serializeContext } from './lib/handlerSource.js';
+import { assertSubject } from './lib/validate.js';
+
+/** @typedef {import('./types.js').OnPersistentOptions} OnPersistentOptions */
+/** @typedef {import('./types.js').PersistentSubscription} PersistentSubscription */
+
+/**
+ * Subscribes to a subject with a subscription that outlives this connection.
+ *
+ * Unlike `onLocal()`, nothing about this lives in the page: the subscription is
+ * stored against the account, keeps matching while the app is closed, and is
+ * ended by `puter.events.unsubscribe()` rather than by navigating away. What
+ * runs is the app's published handler, named by `handlerName`.
+ *
+ * `context` is evaluated **here, now** — serialized once and delivered to every
+ * invocation as a frozen `ctx`. It never re-evaluates, so a value read from the
+ * environment is the value that subscription carries forever.
+ *
+ * @this {import('./index.js').EventsModule}
+ * @param {OnPersistentOptions} options
+ * @returns {Promise}
+ */
+export async function onPersistent (options = {}) {
+ const { puter } = this;
+ assertSubject(options?.subject);
+
+ const { handler, handlerName } = options;
+ // An inline handler is source the server has to match against something it
+ // already has, and a name is the only thing it can match against.
+ if ( handler !== undefined && handler !== null && ! handlerName ) {
+ throw new PuterJSError(
+ 'An inline `handler` needs a `handlerName` to publish it under',
+ 'events_handler_name_required',
+ );
+ }
+
+ const inline = handler === undefined || handler === null
+ ? null
+ : await prepareHandler(puter, handler);
+
+ const body = {
+ subject: options.subject,
+ ...(options.delivery ? { delivery: options.delivery } : {}),
+ ...(options.targets ? { targets: options.targets } : {}),
+ ...(handlerName ? { handlerName } : {}),
+ ...(inline ? { handlerHash: inline.hash } : {}),
+ ...(options.expiresAt !== undefined && options.expiresAt !== null
+ ? { expiresAt: options.expiresAt }
+ : {}),
+ };
+
+ // Serialized only to check it against the cap before the round trip; the
+ // request carries the value, which the server stores the same way.
+ if ( serializeContext(options.context) !== undefined )
+ body.context = options.context;
+
+ return /** @type {PersistentSubscription} */ (
+ await request(puter, '/events/subscribe', body)
+ );
+}
diff --git a/src/puter-js/src/modules/events/persistent.test.js b/src/puter-js/src/modules/events/persistent.test.js
new file mode 100644
index 000000000..42f1b1754
--- /dev/null
+++ b/src/puter-js/src/modules/events/persistent.test.js
@@ -0,0 +1,379 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+// Every persistent 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 { EventHandlers } = await import('./lib/handlers.js');
+const { list } = await import('./list.js');
+const { onPersistent } = await import('./onPersistent.js');
+const { unsubscribe } = await import('./unsubscribe.js');
+
+const SUBJECT = 'fs:~/Documents';
+const HANDLER = ({ event, ctx }) => fetch(ctx.url, { body: event.path });
+/** SHA-256 of the serialized `HANDLER`, computed the same way the SDK does. */
+let handlerHash;
+
+const makeModule = (fsRead) => {
+ const module = {
+ puter: {
+ APIOrigin: 'https://api.test',
+ fs: { read: fsRead ?? vi.fn() },
+ },
+ onPersistent,
+ unsubscribe,
+ list,
+ };
+ module.handlers = new EventHandlers(module);
+ return module;
+};
+
+const bodyOf = (index = 0) => mockRequest.mock.calls[index][2];
+const routeOf = (index = 0) => mockRequest.mock.calls[index][1];
+
+const rejects = async (run) => {
+ try {
+ await run();
+ } catch (error) {
+ return error;
+ }
+ throw new Error('expected a rejection');
+};
+
+beforeEach(async () => {
+ mockRequest.mockReset();
+ mockRequest.mockResolvedValue({});
+ if ( ! handlerHash ) {
+ const { hashSource } = await import('./lib/handlerSource.js');
+ handlerHash = await hashSource(Function.prototype.toString.call(HANDLER));
+ }
+});
+
+describe('onPersistent', () => {
+ it('sends the subject and the server`s answer comes straight back', async () => {
+ const view = { subId: 'app-1#a', subject: SUBJECT };
+ mockRequest.mockResolvedValue(view);
+
+ const sub = await makeModule().onPersistent({ subject: SUBJECT });
+
+ expect(routeOf()).toBe('/events/subscribe');
+ expect(bodyOf()).toEqual({ subject: SUBJECT });
+ expect(sub).toBe(view);
+ });
+
+ it('carries delivery, targets, handlerName and expiry when given', async () => {
+ await makeModule().onPersistent({
+ subject: SUBJECT,
+ delivery: 'single',
+ targets: ['worker'],
+ handlerName: 'ingestUpload',
+ expiresAt: 4102444800,
+ });
+
+ expect(bodyOf()).toEqual({
+ subject: SUBJECT,
+ delivery: 'single',
+ targets: ['worker'],
+ handlerName: 'ingestUpload',
+ expiresAt: 4102444800,
+ });
+ });
+
+ it('sends an inline handler as a hash, never as source', async () => {
+ await makeModule().onPersistent({
+ subject: SUBJECT,
+ handlerName: 'ingestUpload',
+ handler: HANDLER,
+ });
+
+ expect(bodyOf().handlerHash).toBe(handlerHash);
+ expect(bodyOf().source).toBeUndefined();
+ expect(JSON.stringify(bodyOf())).not.toContain('fetch(');
+ });
+
+ it('refuses an inline handler with no name to publish it under', async () => {
+ const error = await rejects(() =>
+ makeModule().onPersistent({ subject: SUBJECT, handler: HANDLER }),
+ );
+
+ expect(error.code).toBe('events_handler_name_required');
+ expect(mockRequest).not.toHaveBeenCalled();
+ });
+
+ it('rejects an inline handler that closes over something', async () => {
+ const error = await rejects(() =>
+ makeModule().onPersistent({
+ subject: SUBJECT,
+ handlerName: 'ingestUpload',
+ handler: '({ event }) => fetch(endpoint, { body: event.path })',
+ }),
+ );
+
+ expect(error.code).toBe('events_handler_free_variable');
+ expect(error.message).toContain('`endpoint`');
+ expect(mockRequest).not.toHaveBeenCalled();
+ });
+
+ it('reads a `{ file }` handler through the caller`s own filesystem', async () => {
+ const read = vi.fn(async () => ({
+ text: async () => '({ event, ctx }) => console.log(event.uid, ctx.url)',
+ }));
+
+ await makeModule(read).onPersistent({
+ subject: SUBJECT,
+ handlerName: 'ingestUpload',
+ handler: { file: '~/AppData/handler.js' },
+ });
+
+ expect(read).toHaveBeenCalledWith('~/AppData/handler.js');
+ expect(bodyOf().handlerHash).toMatch(/^[0-9a-f]{64}$/);
+ });
+
+ it('refuses a subject that is not a non-empty string', async () => {
+ for ( const subject of [undefined, null, '', ' ', 42] ) {
+ const error = await rejects(() =>
+ makeModule().onPersistent({ subject }),
+ );
+ expect(error.code).toBe('invalid_subject');
+ }
+ expect(mockRequest).not.toHaveBeenCalled();
+ });
+
+ it('refuses a handler that is none of the three accepted forms', async () => {
+ const error = await rejects(() =>
+ makeModule().onPersistent({
+ subject: SUBJECT,
+ handlerName: 'x',
+ handler: 42,
+ }),
+ );
+ expect(error.code).toBe('events_handler_invalid');
+ });
+
+ describe('context', () => {
+ it('sends what was passed, evaluated now', async () => {
+ await makeModule().onPersistent({
+ subject: SUBJECT,
+ context: { url: 'https://ingest.example', retries: 2 },
+ });
+
+ expect(bodyOf().context).toEqual({
+ url: 'https://ingest.example',
+ retries: 2,
+ });
+ });
+
+ it('refuses one over the cap before the network', async () => {
+ const error = await rejects(() =>
+ makeModule().onPersistent({
+ subject: SUBJECT,
+ context: { blob: 'x'.repeat(5000) },
+ }),
+ );
+
+ expect(error.code).toBe('events_context_too_large');
+ expect(mockRequest).not.toHaveBeenCalled();
+ });
+
+ it('refuses one that cannot be serialized', async () => {
+ const cyclic = {};
+ cyclic.self = cyclic;
+
+ const error = await rejects(() =>
+ makeModule().onPersistent({ subject: SUBJECT, context: cyclic }),
+ );
+ expect(error.code).toBe('events_context_invalid');
+ });
+ });
+});
+
+describe('unsubscribe', () => {
+ it('names the subscription to end', async () => {
+ await makeModule().unsubscribe('app-1#a');
+
+ expect(routeOf()).toBe('/events/unsubscribe');
+ expect(bodyOf()).toEqual({ subId: 'app-1#a' });
+ });
+
+ it('answers an empty id the way the server answers one it cannot find', async () => {
+ const error = await rejects(() => makeModule().unsubscribe(''));
+ expect(error.code).toBe('subscription_does_not_exist');
+ expect(mockRequest).not.toHaveBeenCalled();
+ });
+});
+
+describe('list', () => {
+ it('follows the cursor and resolves to one array', async () => {
+ mockRequest
+ .mockResolvedValueOnce({ items: [{ subId: 'a' }], cursor: 'next' })
+ .mockResolvedValueOnce({ items: [{ subId: 'b' }] });
+
+ const rows = await makeModule().list();
+
+ expect(rows.map(row => row.subId)).toEqual(['a', 'b']);
+ expect(mockRequest.mock.calls[1][3]).toMatchObject({ cursor: 'next' });
+ });
+
+ it('returns one page envelope when the caller asks for pagination', async () => {
+ mockRequest.mockResolvedValue({ items: [], cursor: 'c', total: 7 });
+
+ const page = await makeModule().list({ cursor: null, includeTotal: true });
+
+ expect(page).toEqual({ items: [], cursor: 'c', total: 7 });
+ expect(mockRequest.mock.calls[0][3]).toMatchObject({ includeTotal: true });
+ });
+
+ it('streams page envelopes when asked to', async () => {
+ mockRequest
+ .mockResolvedValueOnce({ items: [{ subId: 'a' }], cursor: 'next' })
+ .mockResolvedValueOnce({ items: [{ subId: 'b' }] });
+
+ const pages = [];
+ for await ( const page of makeModule().list({ stream: true }) ) pages.push(page);
+
+ expect(pages.map(page => page.items[0].subId)).toEqual(['a', 'b']);
+ });
+});
+
+describe('handlers', () => {
+ it('publishes the serialized source under a name', async () => {
+ mockRequest.mockResolvedValue({
+ name: 'ingestUpload',
+ hash: handlerHash,
+ outcome: 'created',
+ });
+
+ const published = await makeModule().handlers.publish('ingestUpload', HANDLER);
+
+ expect(routeOf()).toBe('/events/handlers/publish');
+ expect(bodyOf()).toEqual({
+ name: 'ingestUpload',
+ source: Function.prototype.toString.call(HANDLER),
+ });
+ expect(published.outcome).toBe('created');
+ });
+
+ it('names the base it is updating once it knows one', async () => {
+ const module = makeModule();
+ mockRequest.mockResolvedValue({ name: 'ingestUpload', hash: 'hash-1' });
+ await module.handlers.publish('ingestUpload', HANDLER);
+
+ mockRequest.mockResolvedValue({ name: 'ingestUpload', hash: 'hash-2' });
+ await module.handlers.publish('ingestUpload', '({ ctx }) => ctx.url');
+
+ expect(bodyOf(1).ifHash).toBe('hash-1');
+ });
+
+ it('takes the base from a listing too', async () => {
+ const module = makeModule();
+ mockRequest.mockResolvedValue({
+ handlers: [{ name: 'ingestUpload', hash: 'hash-9', subscriptions: 0 }],
+ });
+ await module.handlers.list();
+
+ mockRequest.mockResolvedValue({ name: 'ingestUpload', hash: 'hash-10' });
+ await module.handlers.publish('ingestUpload', HANDLER);
+
+ expect(bodyOf(1).ifHash).toBe('hash-9');
+ });
+
+ it('names no base when the caller means to take the name', async () => {
+ const module = makeModule();
+ mockRequest.mockResolvedValue({ name: 'ingestUpload', hash: 'hash-1' });
+ await module.handlers.publish('ingestUpload', HANDLER);
+
+ await module.handlers.publish('ingestUpload', '({ ctx }) => ctx.url', {
+ replace: true,
+ });
+
+ expect(bodyOf(1)).toMatchObject({ replace: true });
+ expect(bodyOf(1).ifHash).toBeUndefined();
+ });
+
+ it('publishes a whole set in one call', async () => {
+ mockRequest.mockResolvedValue({
+ handlers: [
+ { name: 'a', hash: 'h1' },
+ { name: 'b', hash: 'h2' },
+ ],
+ });
+
+ const published = await makeModule().handlers.publishAll([
+ { name: 'a', handler: HANDLER },
+ { name: 'b', handler: '({ ctx }) => ctx.url' },
+ ]);
+
+ expect(routeOf()).toBe('/events/handlers/publishAll');
+ expect(bodyOf().handlers.map(entry => entry.name)).toEqual(['a', 'b']);
+ expect(published).toHaveLength(2);
+ });
+
+ it('rejects a set item that closes over something, before sending anything', async () => {
+ const error = await rejects(() =>
+ makeModule().handlers.publishAll([
+ { name: 'a', handler: HANDLER },
+ { name: 'b', handler: '({ event }) => publish(event)' },
+ ]),
+ );
+
+ expect(error.code).toBe('events_handler_free_variable');
+ expect(error.message).toContain('`publish`');
+ expect(mockRequest).not.toHaveBeenCalled();
+ });
+
+ it('names an app when the caller is an account session', async () => {
+ mockRequest.mockResolvedValue({ name: 'a', hash: 'h' });
+ await makeModule().handlers.publish('a', HANDLER, { appUid: 'app-7' });
+
+ expect(bodyOf()).toMatchObject({ appUid: 'app-7' });
+ });
+
+ it('does not carry one app`s base into another`s', async () => {
+ const module = makeModule();
+ mockRequest.mockResolvedValue({ name: 'a', hash: 'hash-1' });
+ await module.handlers.publish('a', HANDLER, { appUid: 'app-1' });
+
+ // The same name in another app is different code, and this publish is
+ // not an update to anything.
+ await module.handlers.publish('a', HANDLER, { appUid: 'app-2' });
+ expect(bodyOf(1).ifHash).toBeUndefined();
+ });
+
+ it('lists names and hashes, and forgets a name it removes', async () => {
+ const module = makeModule();
+ mockRequest.mockResolvedValue({
+ handlers: [{ name: 'a', hash: 'h', updatedAt: 1, subscriptions: 3 }],
+ });
+
+ const listed = await module.handlers.list();
+ expect(routeOf()).toBe('/events/handlers/list');
+ expect(listed).toEqual([
+ { name: 'a', hash: 'h', updatedAt: 1, subscriptions: 3 },
+ ]);
+
+ mockRequest.mockResolvedValue({ name: 'a', removed: true, suspended: 3 });
+ await module.handlers.remove('a');
+ expect(routeOf(1)).toBe('/events/handlers/remove');
+ expect(module.handlers.known.size).toBe(0);
+ });
+
+ it('refuses a name that is not a non-empty string', async () => {
+ for ( const name of [undefined, '', ' ', 7] ) {
+ const error = await rejects(() =>
+ makeModule().handlers.publish(name, HANDLER),
+ );
+ expect(error.code).toBe('events_handler_name_invalid');
+ }
+ });
+
+ it('works when destructured off the module', async () => {
+ const { publish } = makeModule().handlers;
+ mockRequest.mockResolvedValue({ name: 'a', hash: 'h' });
+
+ await publish('a', HANDLER);
+ expect(routeOf()).toBe('/events/handlers/publish');
+ });
+});
diff --git a/src/puter-js/src/modules/events/types.js b/src/puter-js/src/modules/events/types.js
index f63ab7bd7..5dae4f9aa 100644
--- a/src/puter-js/src/modules/events/types.js
+++ b/src/puter-js/src/modules/events/types.js
@@ -63,8 +63,9 @@
* @property {'gap'} op Always `'gap'`.
* @property {string} reason Why the delivery was dropped —
* `matched_subscription_limit`, `filter_evaluation_limit`,
- * `delivery_rate_limit`, or `backlog_overflow` when undelivered events were
- * shed to stay inside a backlog cap.
+ * `delivery_rate_limit`, `backlog_overflow` when undelivered events were
+ * shed to stay inside a backlog cap, or `suspended_backlog_expired` when a
+ * suspended subscription held them past its deadline.
* @property {number} ts Milliseconds since the epoch.
*/
@@ -75,6 +76,9 @@
* @typedef {Object} EventDelivery
* @property {PuterEvent | PuterKvEvent | EventGapMarker} event The delivered
* event, or a gap marker in place of events that were dropped.
+ * @property {Readonly>} [ctx] The `context` the
+ * subscription was created with, frozen. Present only for a persistent
+ * subscription; a session subscription carries none.
*/
/**
@@ -96,3 +100,111 @@
* @property {number} [timeout] How long to wait for the server to answer
* `subscribe`, in milliseconds. Default `30000`.
*/
+
+/**
+ * Options for {@link import('./onPersistent.js').onPersistent}.
+ *
+ * @typedef {Object} OnPersistentOptions
+ * @property {string} subject What to watch — the same grammar `onLocal()`
+ * takes, e.g. `fs:~/Documents` or `fs:~/inbox/*.json:add`.
+ * @property {'broadcast' | 'single'} [delivery] `broadcast` (the default)
+ * delivers to everything listening; `single` delivers to exactly one
+ * consumer, which must acknowledge, and requires a `handlerName`.
+ * @property {Array<'socket' | 'worker' | 'push'>} [targets] Transports the
+ * deliveries may take. Defaults to `['socket', 'worker']`. A `single`
+ * subscription may not target `push`.
+ * @property {string} [handlerName] The published handler this subscription
+ * binds to. Required for `single`.
+ * @property {Function | string | { file: string }} [handler] The handler
+ * source this subscription was written against. Sent as a hash, not as
+ * source: the subscription binds only if it matches what is published under
+ * `handlerName`, which is also required when this is given.
+ * @property {Record} [context] Values the handler needs,
+ * evaluated **now** and delivered as a frozen `ctx` on every invocation.
+ * Capped at 4 KB serialized.
+ * @property {number | string} [expiresAt] When the subscription ends by
+ * itself — unix seconds or an ISO-8601 string, and it has to be in the
+ * future.
+ */
+
+/**
+ * A subscription that outlives the connection that made it.
+ *
+ * `context` values are deliberately absent: the column holds whatever secret
+ * the handler needs, and a listing is the one surface an app can call
+ * repeatedly. What comes back is its shape — which keys are set, and a hash
+ * that changes when any value does.
+ *
+ * @typedef {Object} PersistentSubscription
+ * @property {string} subId The subscription's id, and what `unsubscribe()`
+ * names. Stable for the life of the subscription.
+ * @property {string} subject The subject it was created with.
+ * @property {EventAnchor} anchor The node it is keyed to.
+ * @property {string | null} match The pattern events under the anchor are
+ * matched against, or `null` when the subject named the anchor itself.
+ * @property {string | null} op The single operation it is limited to, or
+ * `null` for all of them.
+ * @property {Array<'socket' | 'worker' | 'push'>} targets Transports its
+ * deliveries may take.
+ * @property {'broadcast' | 'single'} delivery Its delivery class.
+ * @property {string | null} handlerName The handler it is bound to.
+ * @property {string | null} appUid The app that created it, or `null` for one
+ * an account session made.
+ * @property {string[] | null} contextKeys Key names of its stored context,
+ * never the values, or `null` when it carries none.
+ * @property {string | null} contextHash Hash of its stored context, so a
+ * change is visible without the values.
+ * @property {number} createdAt Unix seconds.
+ * @property {number | null} expiresAt Unix seconds, or `null` for one with no
+ * end.
+ * @property {number | null} suspendedAt When it stopped delivering without
+ * being removed, or `null` while it is live.
+ * @property {string | null} suspendedReason Why it stopped —
+ * `handler_not_found`, `failures`, `no_credit`, or `permission_revoked`.
+ */
+
+/**
+ * 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.
+ *
+ * @typedef {Object} HandlerOptions
+ * @property {boolean} [replace] Take the name whatever is published under it.
+ * Without this, a publish whose base has moved is refused with
+ * `events_handler_conflict`.
+ * @property {string} [appUid] The app to publish into. Required when the
+ * caller is an account session rather than an app.
+ */
+
+/**
+ * One item of a `publishAll()` set.
+ *
+ * @typedef {Object} HandlerPublication
+ * @property {string} name The name subscriptions bind to.
+ * @property {Function | string | { file: string }} handler The handler: a
+ * function, its source, or a path to read it from.
+ * @property {boolean} [replace] Take the name whatever is published under it.
+ */
+
+/**
+ * What a publish reports back. Never the source.
+ *
+ * @typedef {Object} PublishedHandler
+ * @property {string} name
+ * @property {string} hash SHA-256 of the published source.
+ * @property {number} updatedAt Unix seconds.
+ * @property {'created' | 'updated' | 'unchanged'} outcome What the publish
+ * did. `unchanged` means the same source was already published.
+ * @property {number} resumed Suspended subscriptions this publish brought
+ * back into service.
+ */
+
+/**
+ * One handler as `puter.events.handlers.list()` reports it.
+ *
+ * @typedef {Object} HandlerSummary
+ * @property {string} name
+ * @property {string} hash SHA-256 of the published source.
+ * @property {number} updatedAt Unix seconds.
+ * @property {number} subscriptions How many subscriptions are bound to this
+ * name, suspended ones included.
+ */
diff --git a/src/puter-js/src/modules/events/unsubscribe.js b/src/puter-js/src/modules/events/unsubscribe.js
new file mode 100644
index 000000000..a8e0f6bc5
--- /dev/null
+++ b/src/puter-js/src/modules/events/unsubscribe.js
@@ -0,0 +1,23 @@
+import { PuterJSError } from '../../lib/PuterJSError.js';
+import { request } from './lib/api.js';
+
+/**
+ * Ends a persistent subscription.
+ *
+ * An id this account does not hold — one already ended, or one another app
+ * created — reads as absent rather than refused, so the call cannot be used to
+ * find out which subscriptions exist.
+ *
+ * @this {import('./index.js').EventsModule}
+ * @param {string} subId The `subId` of the subscription to end.
+ * @returns {Promise}
+ */
+export async function unsubscribe (subId) {
+ if ( typeof subId !== 'string' || subId.trim().length === 0 ) {
+ throw new PuterJSError(
+ 'No such subscription',
+ 'subscription_does_not_exist',
+ );
+ }
+ await request(this.puter, '/events/unsubscribe', { subId });
+}
diff --git a/src/puter-js/tests/api/suites/events.suite.ts b/src/puter-js/tests/api/suites/events.suite.ts
index 0ceb2fa87..2db58390c 100644
--- a/src/puter-js/tests/api/suites/events.suite.ts
+++ b/src/puter-js/tests/api/suites/events.suite.ts
@@ -60,12 +60,40 @@ const open = (
timeout: SUBSCRIBE_TIMEOUT_MS,
});
+/** A handler that closes over nothing, so the free-variable scan accepts it. */
+const HANDLER = '({ event, ctx }) => { console.log(event.path, ctx.label); }';
+const OTHER_HANDLER = '({ event, ctx }) => { console.log(ctx.label, event.uid); }';
+
+/** An app of this account's own, so its handlers are the caller's to publish. */
+const makeApp = async (t: TestContext): Promise => {
+ const name = unique('events-handlers');
+ const app = await t.puter.apps.create(name, `https://example.com/${name}`);
+ return (app as unknown as { uid: string }).uid;
+};
+
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');
},
+ 'exposes the persistent surface': async (t) => {
+ for (const method of ['onPersistent', 'unsubscribe', 'list'] as const) {
+ t.assert.equal(
+ typeof t.puter.events[method],
+ 'function',
+ `puter.events.${method} is a function`,
+ );
+ }
+ for (const method of ['publish', 'publishAll', 'list', 'remove'] as const) {
+ t.assert.equal(
+ typeof t.puter.events.handlers[method],
+ 'function',
+ `puter.events.handlers.${method} is a 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(
@@ -319,4 +347,214 @@ export default suite('events', {
);
}
},
+
+ // -- Persistent subscriptions ------------------------------------
+
+ 'refuses an inline handler with no name to publish it under': async (t) => {
+ const error = await t.assert.rejects(() =>
+ t.puter.events.onPersistent({
+ subject: `fs:/${t.env.users.user.username}`,
+ handler: HANDLER,
+ }),
+ );
+ t.assert.equal(codeOf(error), 'events_handler_name_required');
+ },
+
+ 'refuses a handler that closes over something it cannot carry': async (t) => {
+ const error = await t.assert.rejects(() =>
+ t.puter.events.onPersistent({
+ subject: `fs:/${t.env.users.user.username}`,
+ handlerName: 'ingestUpload',
+ handler: '({ event }) => fetch(ingestUrl, { body: event.path })',
+ }),
+ );
+ t.assert.equal(codeOf(error), 'events_handler_free_variable');
+ t.assert.ok(
+ (error as Error).message.includes('ingestUrl'),
+ 'the error names the identifier that could not be resolved',
+ );
+ },
+
+ 'refuses a context over the cap before the round trip': async (t) => {
+ const error = await t.assert.rejects(() =>
+ t.puter.events.onPersistent({
+ subject: `fs:/${t.env.users.user.username}`,
+ context: { blob: 'x'.repeat(5000) },
+ }),
+ );
+ t.assert.equal(codeOf(error), 'events_context_too_large');
+ },
+
+ 'creates, lists and ends a persistent subscription': async (t) => {
+ const dir = await makeDir(t, 'events-persistent');
+
+ const sub = await t.puter.events.onPersistent({
+ subject: `fs:${dir}`,
+ context: { label: 'ingest', token: 'shhh-not-in-a-listing' },
+ });
+
+ try {
+ t.assert.ok(sub.subId, 'the subscription carries a server id');
+ t.assert.equal(sub.subject, `fs:${dir}`);
+ t.assert.equal(sub.delivery, 'broadcast');
+ t.assert.equal(sub.suspendedAt, null);
+
+ const held = await t.puter.events.list();
+ const listed = held.find((row) => row.subId === sub.subId);
+ t.assert.ok(listed, 'the subscription is in the account`s listing');
+ // The context is where an API key lives, so a listing reports its
+ // shape and never its values.
+ t.assert.deepEqual(listed?.contextKeys, ['label', 'token']);
+ t.assert.ok(
+ typeof listed?.contextHash === 'string' &&
+ listed.contextHash.length === 64,
+ 'the listing carries a content hash of the context',
+ );
+ t.assert.ok(
+ ! JSON.stringify(listed).includes('shhh-not-in-a-listing'),
+ 'the listing carries no context values',
+ );
+ } finally {
+ await t.puter.events.unsubscribe(sub.subId);
+ }
+
+ const after = await t.puter.events.list();
+ t.assert.ok(
+ ! after.some((row) => row.subId === sub.subId),
+ 'the subscription is gone once unsubscribed',
+ );
+ },
+
+ 'answers a listing page when asked for one': async (t) => {
+ const page = await t.puter.events.list({ cursor: null, includeTotal: true });
+ t.assert.ok(Array.isArray(page.items), 'a page carries items');
+ t.assert.equal(typeof page.total, 'number', 'a total was requested');
+ },
+
+ 'answers an id it does not hold the way it answers one that is gone': async (t) => {
+ const error = await t.assert.rejects(() =>
+ t.puter.events.unsubscribe(''),
+ );
+ t.assert.equal(codeOf(error), 'subscription_does_not_exist');
+ },
+
+ 'refuses to bind an inline handler nothing is published for': async (t) => {
+ const dir = await makeDir(t, 'events-unbound');
+ const error = await t.assert.rejects(() =>
+ t.puter.events.onPersistent({
+ subject: `fs:${dir}`,
+ handlerName: unique('missing'),
+ handler: HANDLER,
+ }),
+ );
+ t.assert.equal(codeOf(error), 'events_handler_not_found');
+ },
+
+ // -- Handlers ----------------------------------------------------
+
+ 'publishes, lists and removes a named handler': async (t) => {
+ const appUid = await makeApp(t);
+
+ const published = await t.puter.events.handlers.publish(
+ 'ingestUpload',
+ HANDLER,
+ { appUid },
+ );
+ t.assert.equal(published.name, 'ingestUpload');
+ t.assert.equal(published.outcome, 'created');
+ t.assert.ok(
+ typeof published.hash === 'string' && published.hash.length === 64,
+ 'a publish reports the source hash',
+ );
+
+ const listed = await t.puter.events.handlers.list({ appUid });
+ t.assert.deepEqual(
+ listed.map((row) => row.name),
+ ['ingestUpload'],
+ );
+ t.assert.equal(listed[0].subscriptions, 0);
+ t.assert.ok(
+ ! JSON.stringify(listed).includes('console.log'),
+ 'a listing never carries handler source',
+ );
+
+ const removed = await t.puter.events.handlers.remove('ingestUpload', {
+ appUid,
+ });
+ t.assert.equal(removed.removed, true);
+ t.assert.equal(removed.suspended, 0);
+ t.assert.deepEqual(await t.puter.events.handlers.list({ appUid }), []);
+ },
+
+ 'republishing the same source changes nothing': async (t) => {
+ const appUid = await makeApp(t);
+ await t.puter.events.handlers.publish('ingestUpload', HANDLER, { appUid });
+
+ const again = await t.puter.events.handlers.publish(
+ 'ingestUpload',
+ HANDLER,
+ { appUid },
+ );
+ t.assert.equal(again.outcome, 'unchanged');
+ },
+
+ 'updates a name it published, and takes one it means to replace': async (t) => {
+ const appUid = await makeApp(t);
+ await t.puter.events.handlers.publish('ingestUpload', HANDLER, { appUid });
+
+ // Having published it, this client knows the base it is updating, so
+ // the change is accepted rather than read as a racing build step.
+ const updated = await t.puter.events.handlers.publish(
+ 'ingestUpload',
+ OTHER_HANDLER,
+ { appUid },
+ );
+ t.assert.equal(updated.outcome, 'updated');
+
+ const replaced = await t.puter.events.handlers.publish(
+ 'ingestUpload',
+ HANDLER,
+ { appUid, replace: true },
+ );
+ t.assert.equal(replaced.outcome, 'updated');
+ t.assert.equal(
+ (await t.puter.events.handlers.list({ appUid }))[0].hash,
+ replaced.hash,
+ 'the listing reports what the last publish left',
+ );
+ },
+
+ 'takes a whole set in one call': async (t) => {
+ const appUid = await makeApp(t);
+
+ const published = await t.puter.events.handlers.publishAll(
+ [
+ { name: 'ingestUpload', handler: HANDLER },
+ { name: 'indexDocument', handler: OTHER_HANDLER },
+ ],
+ { appUid },
+ );
+
+ t.assert.deepEqual(
+ published.map((row) => row.name),
+ ['ingestUpload', 'indexDocument'],
+ );
+ t.assert.equal((await t.puter.events.handlers.list({ appUid })).length, 2);
+ },
+
+ 'refuses to publish into an app this account does not own': async (t) => {
+ const error = await t.assert.rejects(() =>
+ t.puter.events.handlers.publish('ingestUpload', HANDLER, {
+ appUid: 'app-00000000-0000-4000-8000-000000000099',
+ }),
+ );
+ t.assert.equal(codeOf(error), 'events_handler_forbidden');
+ },
+
+ 'refuses to publish without naming an app at all': async (t) => {
+ const error = await t.assert.rejects(() =>
+ t.puter.events.handlers.publish('ingestUpload', HANDLER),
+ );
+ t.assert.equal(codeOf(error), 'events_handler_app_required');
+ },
});