diff --git a/app/config.ts b/app/config.ts index d5ff02e..6dd74dc 100644 --- a/app/config.ts +++ b/app/config.ts @@ -60,7 +60,6 @@ export const config: ContrailConfig = { }, profiles: [ - "app.bsky.actor.profile", - { collection: "site.standard.publication", rkey: "blento.self" } + "app.bsky.actor.profile" ] }; diff --git a/community-lexicon-templates/whoami.json b/community-lexicon-templates/space/whoami.json similarity index 94% rename from community-lexicon-templates/whoami.json rename to community-lexicon-templates/space/whoami.json index bf5e847..fd0a6fa 100644 --- a/community-lexicon-templates/whoami.json +++ b/community-lexicon-templates/space/whoami.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "tools.atmo.community.whoami", + "id": "tools.atmo.community.space.whoami", "defs": { "main": { "type": "query", diff --git a/docs/community-spec-mapping.md b/docs/community-spec-mapping.md index ba2cc1b..290ee62 100644 --- a/docs/community-spec-mapping.md +++ b/docs/community-spec-mapping.md @@ -21,13 +21,13 @@ permissioned-data spec with no knowledge of access levels or delegation. | ------------------------------------------- | ------------------------------------------------------------------- | --------- | -------------------------------------------------------------------------------------------------------- | | Community as a DID | `communities.did` | ✅ | 1:1 | | Mint a fresh did:plc for a community | `community.mint` → P-256 keypairs + genesis op + plc.directory POST | ✅ | Post uses secp256k1; we use P-256 (both spec-valid, we avoid the dep) | -| Adopt an existing account | `community.adopt` (app password) | ➕ | Not in the post; contrail addition. See [community.md](./community.md) for the rationale | +| Adopt an existing account | `community.adopt` (app password) | ➕ | Not in the post; contrail addition — app passwords avoid OAuth's periodic re-auth, useful for machine-operated accounts | | Creator-held rotation key (recovery) | Returned once by `community.mint` as `recoveryKey` | ✅ | Never stored; caller must save it | | Groups-are-spaces | Community-owned rows in `spaces` | ✅ | No separate groups table; every group is a space | | `$admin` reserved space | Auto-created on community creation | ✅ | Keyed by the literal string `$admin`; cannot be deleted | | `$publishers` reserved space | Auto-created on community creation | ➕ | Not in the post; contrail extension for the "publish public records as the community" capability | | Delegated membership (space → space) | `community_access_levels.subject_kind = 'space'` | ✅ | Recursive resolution with cycle guard + depth cap | -| Access-level ladder | `member` / `manager` / `admin` / `owner` | ⚠️ | 4 levels vs the post's 8. See [community.md § Relationship to the Arbiter post](./community.md) | +| Access-level ladder | `member` / `manager` / `admin` / `owner` | ⚠️ | 4 levels vs the post's 8 — post-levels 2–3 (add/remove members) collapse into `manager`; levels 4–5 collapse into `admin` | | Read-Member-List (pre-member tier) | _none_ | ❌ | Post's level 1; skipped in v1 | | Add-Members vs Remove-Members split | Bundled into `manager` | ⚠️ | Post treats them separately | | Configure-Space | `admin` in that space | ✅ | Post's level 5 | @@ -35,7 +35,7 @@ permissioned-data spec with no knowledge of access levels or delegation. | Remove-Space | `owner` in target space OR `admin` in `$admin` | ✅ | Post's level 7 | | Owner | `owner`; only meaningful in `$admin` for owner-management | ✅ | Post's level 8 | | Push model for membership lists | Reconciler writes `spaces_members` after each change | ➕ | Post doesn't specify a sync direction; push keeps spaces read-path zero-overhead | -| Cross-community / cross-arbiter delegation | _same-contrail only_ | ❌ | v1 constraint. Private-membership federation is deferred — see [community.md § Deferred work] | +| Cross-community / cross-arbiter delegation | _same-contrail only_ | ❌ | v1 constraint. Private-membership federation is deferred until the cross-instance distribution shape firms up | | Invites as a separate service | `community.invite.*` (create/list/revoke/redeem) | ➕ | Built-in — tokens pre-sign a grant, redeemer gets the encoded access level. SHA-256 at rest, atomic redeem. | | Writing records under the arbiter's account | `community.space.putRecord` (in-space) + `community.putRecord` (public) | ✅ | In-space: `admin+`. Public: `member+` in `$publishers` — routes through adopted community's PDS | | Public membership-list flag | _none_ | ❌ | Post allows spaces to expose membership publicly. Ruled out for cross-instance federation (privacy) | @@ -55,7 +55,7 @@ membership is user-controlled or community-controlled. ### Community lifecycle - `community.adopt` · `community.mint` · `community.reauth` · `community.delete` -- `community.list` · `community.getHealth` · `community.whoami` +- `community.list` · `community.getHealth` ### Space (group / role / channel) lifecycle - `community.space.create` · `community.space.delete` @@ -63,6 +63,7 @@ membership is user-controlled or community-controlled. ### Membership - `community.space.grant` · `community.space.revoke` · `community.space.setAccessLevel` - `community.space.listMembers` (`?flatten=true` for the resolved DID list) · `community.space.resync` +- `community.space.whoami` — caller's effective access level in a space ### Invites - `community.invite.create` · `community.invite.list` · `community.invite.revoke` · `community.invite.redeem` diff --git a/docs/spaces-later.md b/docs/spaces-later.md index 972a4ce..270b55c 100644 --- a/docs/spaces-later.md +++ b/docs/spaces-later.md @@ -61,15 +61,11 @@ don't have that yet. Lightweight interim: Server-Sent Events on `space.subscribeRecords?spaceUri=&collection=`. Works for first-party apps right away; swap to the real thing later. -## Namespace split for contrail-specific extras -Right now `space.invite.*`, `space.whoami`, and `space.leaveSpace` all live -alongside spec-adjacent endpoints. If the spec lands with different names or -semantics for some of these, migration cost is "rename everywhere." A second -namespace (`.spaceExt.*` or `.contrail.*`) for clearly-off-spec -features would keep the `space.*` surface close to whatever the spec becomes. - -Decision: split them. Pick a namespace name, move at least `invite.*` and -`whoami`; `leaveSpace` is ambiguous. +## Namespace split for contrail-specific extras — done +`space.invite.*` and `space.whoami` moved to `.spaceExt.*`. See +[spaces-spec-mapping.md § Contrail extras](./spaces-spec-mapping.md#contrail-extras-namespace-nsspaceext). +`leaveSpace` is still in `space.*` — revisit if the spec lands with different +self-remove semantics. ## Ownership transfer Dropped for now. The space URI is `at:////` — owner DID diff --git a/docs/spaces-spec-mapping.md b/docs/spaces-spec-mapping.md index 36c9ccf..3368191 100644 --- a/docs/spaces-spec-mapping.md +++ b/docs/spaces-spec-mapping.md @@ -44,7 +44,6 @@ All endpoints are emitted under `.space.*` from templates in - `space.listMembers` — members for a space (member/owner only) - `space.listRecords` — space-scoped record listing; bearer-read supported - `space.getRecord` — single record; bearer-read supported -- `space.whoami` — caller's relationship to a space (extra; not in spec) ### Write - `space.putRecord` @@ -56,11 +55,16 @@ All endpoints are emitted under `.space.*` from templates in - `space.removeMember` - `space.leaveSpace` — self-remove; owner cannot leave (extra) -### Invites (extra; not in the spec) -- `space.invite.create` — returns raw token once; hash stored -- `space.invite.redeem` -- `space.invite.list` -- `space.invite.revoke` +### Contrail extras (namespace: `.spaceExt.*`) +Clearly-off-spec features live under a separate namespace so the `space.*` +surface stays close to whatever the permissioned-data spec becomes. Moved here +from `space.*` in an earlier refactor. + +- `spaceExt.whoami` — caller's relationship to a space (owner / member flags) +- `spaceExt.invite.create` — returns raw token once; hash stored +- `spaceExt.invite.redeem` +- `spaceExt.invite.list` +- `spaceExt.invite.revoke` Invites have three kinds: `join`, `read`, `read-join`. `read` tokens grant bearer-only anonymous read access; `read-join` does both; `join` requires a diff --git a/docs/todo.md b/docs/todo.md deleted file mode 100644 index 6e2293a..0000000 --- a/docs/todo.md +++ /dev/null @@ -1,4 +0,0 @@ -- permissioned blobs -- websockets/sse/webhooks -- add ingestion logic? -- public spaces? \ No newline at end of file diff --git a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/channel/watchRecords.json b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/channel/watchRecords.json new file mode 100644 index 0000000..c128f52 --- /dev/null +++ b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/channel/watchRecords.json @@ -0,0 +1,114 @@ +{ + "lexicon": 1, + "id": "tools.atmo.chat.channel.watchRecords", + "defs": { + "main": { + "type": "query", + "description": "Subscribe to a live tools.atmo.chat.channel query. Returns Server-Sent Events: `snapshot.start`, `snapshot.record` (N of), `snapshot.end`, then `record.created` / `record.deleted` as records enter/leave the result, plus a final `member.removed` if the caller loses access mid-stream. v1 requires spaceUri; cross-space watch is deferred.", + "parameters": { + "type": "params", + "properties": { + "limit": { + "type": "integer", + "minimum": 1, + "maximum": 200, + "default": 50 + }, + "cursor": { + "type": "string" + }, + "actor": { + "type": "string", + "format": "at-identifier", + "description": "Filter by DID or handle (triggers on-demand backfill)" + }, + "profiles": { + "type": "boolean", + "description": "Include profile + identity info keyed by DID" + }, + "spaceUri": { + "type": "string", + "format": "at-uri", + "description": "If set, query records inside this permissioned space (requires service-auth JWT or a read-grant invite token)." + }, + "byUser": { + "type": "string", + "format": "did", + "description": "Only used with spaceUri — filter to records authored by this DID." + }, + "inviteToken": { + "type": "string", + "description": "Read-grant invite token for anonymous bearer access. Replaces JWT auth when supplied." + }, + "communityDid": { + "type": "string", + "description": "Filter by communityDid" + }, + "name": { + "type": "string", + "description": "Filter by name" + }, + "topic": { + "type": "string", + "description": "Filter by topic" + }, + "visibility": { + "type": "string", + "description": "Filter by visibility" + }, + "createdAtMin": { + "type": "string", + "description": "Minimum value for createdAt" + }, + "createdAtMax": { + "type": "string", + "description": "Maximum value for createdAt" + }, + "sort": { + "type": "string", + "knownValues": [ + "communityDid", + "name", + "topic", + "visibility", + "createdAt" + ], + "description": "Field to sort by (default: time_us)" + }, + "order": { + "type": "string", + "knownValues": [ + "asc", + "desc" + ], + "description": "Sort direction (default: desc for dates/numbers/counts, asc for strings)" + } + } + }, + "output": { + "encoding": "text/event-stream", + "schema": { + "type": "object", + "properties": {} + } + }, + "errors": [ + { + "name": "InvalidRequest" + }, + { + "name": "AuthRequired" + }, + { + "name": "Forbidden" + }, + { + "name": "NotFound" + }, + { + "name": "NotSupported" + } + ] + } + } +} diff --git a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/community/whoami.json b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/community/space/whoami.json similarity index 95% rename from examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/community/whoami.json rename to examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/community/space/whoami.json index 92b1f97..457244d 100644 --- a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/community/whoami.json +++ b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/community/space/whoami.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "tools.atmo.chat.community.whoami", + "id": "tools.atmo.chat.community.space.whoami", "defs": { "main": { "type": "query", diff --git a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/message/watchRecords.json b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/message/watchRecords.json new file mode 100644 index 0000000..b5e607a --- /dev/null +++ b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/message/watchRecords.json @@ -0,0 +1,104 @@ +{ + "lexicon": 1, + "id": "tools.atmo.chat.message.watchRecords", + "defs": { + "main": { + "type": "query", + "description": "Subscribe to a live tools.atmo.chat.message query. Returns Server-Sent Events: `snapshot.start`, `snapshot.record` (N of), `snapshot.end`, then `record.created` / `record.deleted` as records enter/leave the result, plus a final `member.removed` if the caller loses access mid-stream. v1 requires spaceUri; cross-space watch is deferred.", + "parameters": { + "type": "params", + "properties": { + "limit": { + "type": "integer", + "minimum": 1, + "maximum": 200, + "default": 50 + }, + "cursor": { + "type": "string" + }, + "actor": { + "type": "string", + "format": "at-identifier", + "description": "Filter by DID or handle (triggers on-demand backfill)" + }, + "profiles": { + "type": "boolean", + "description": "Include profile + identity info keyed by DID" + }, + "spaceUri": { + "type": "string", + "format": "at-uri", + "description": "If set, query records inside this permissioned space (requires service-auth JWT or a read-grant invite token)." + }, + "byUser": { + "type": "string", + "format": "did", + "description": "Only used with spaceUri — filter to records authored by this DID." + }, + "inviteToken": { + "type": "string", + "description": "Read-grant invite token for anonymous bearer access. Replaces JWT auth when supplied." + }, + "text": { + "type": "string", + "description": "Filter by text" + }, + "createdAtMin": { + "type": "string", + "description": "Minimum value for createdAt" + }, + "createdAtMax": { + "type": "string", + "description": "Maximum value for createdAt" + }, + "replyTo": { + "type": "string", + "description": "Filter by replyTo" + }, + "sort": { + "type": "string", + "knownValues": [ + "text", + "createdAt", + "replyTo" + ], + "description": "Field to sort by (default: time_us)" + }, + "order": { + "type": "string", + "knownValues": [ + "asc", + "desc" + ], + "description": "Sort direction (default: desc for dates/numbers/counts, asc for strings)" + } + } + }, + "output": { + "encoding": "text/event-stream", + "schema": { + "type": "object", + "properties": {} + } + }, + "errors": [ + { + "name": "InvalidRequest" + }, + { + "name": "AuthRequired" + }, + { + "name": "Forbidden" + }, + { + "name": "NotFound" + }, + { + "name": "NotSupported" + } + ] + } + } +} diff --git a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/permissionSet.json b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/permissionSet.json index 1704d8a..99c71cf 100644 --- a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/permissionSet.json +++ b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/permissionSet.json @@ -14,6 +14,7 @@ "lxm": [ "tools.atmo.chat.channel.getRecord", "tools.atmo.chat.channel.listRecords", + "tools.atmo.chat.channel.watchRecords", "tools.atmo.chat.community.adopt", "tools.atmo.chat.community.delete", "tools.atmo.chat.community.deleteRecord", @@ -35,27 +36,25 @@ "tools.atmo.chat.community.space.resync", "tools.atmo.chat.community.space.revoke", "tools.atmo.chat.community.space.setAccessLevel", - "tools.atmo.chat.community.whoami", + "tools.atmo.chat.community.space.whoami", "tools.atmo.chat.getCursor", "tools.atmo.chat.getOverview", "tools.atmo.chat.getProfile", "tools.atmo.chat.message.getRecord", "tools.atmo.chat.message.listRecords", + "tools.atmo.chat.message.watchRecords", "tools.atmo.chat.notifyOfUpdate", "tools.atmo.chat.realtime.subscribe", "tools.atmo.chat.realtime.ticket", "tools.atmo.chat.server.getRecord", "tools.atmo.chat.server.listRecords", + "tools.atmo.chat.server.watchRecords", "tools.atmo.chat.space.addMember", "tools.atmo.chat.space.createSpace", "tools.atmo.chat.space.deleteRecord", "tools.atmo.chat.space.getBlob", "tools.atmo.chat.space.getRecord", "tools.atmo.chat.space.getSpace", - "tools.atmo.chat.space.invite.create", - "tools.atmo.chat.space.invite.list", - "tools.atmo.chat.space.invite.redeem", - "tools.atmo.chat.space.invite.revoke", "tools.atmo.chat.space.leaveSpace", "tools.atmo.chat.space.listBlobs", "tools.atmo.chat.space.listMembers", @@ -64,7 +63,11 @@ "tools.atmo.chat.space.putRecord", "tools.atmo.chat.space.removeMember", "tools.atmo.chat.space.uploadBlob", - "tools.atmo.chat.space.whoami" + "tools.atmo.chat.spaceExt.invite.create", + "tools.atmo.chat.spaceExt.invite.list", + "tools.atmo.chat.spaceExt.invite.redeem", + "tools.atmo.chat.spaceExt.invite.revoke", + "tools.atmo.chat.spaceExt.whoami" ] } ] diff --git a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/server/watchRecords.json b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/server/watchRecords.json new file mode 100644 index 0000000..9a8bbe7 --- /dev/null +++ b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/server/watchRecords.json @@ -0,0 +1,109 @@ +{ + "lexicon": 1, + "id": "tools.atmo.chat.server.watchRecords", + "defs": { + "main": { + "type": "query", + "description": "Subscribe to a live tools.atmo.chat.server query. Returns Server-Sent Events: `snapshot.start`, `snapshot.record` (N of), `snapshot.end`, then `record.created` / `record.deleted` as records enter/leave the result, plus a final `member.removed` if the caller loses access mid-stream. v1 requires spaceUri; cross-space watch is deferred.", + "parameters": { + "type": "params", + "properties": { + "limit": { + "type": "integer", + "minimum": 1, + "maximum": 200, + "default": 50 + }, + "cursor": { + "type": "string" + }, + "actor": { + "type": "string", + "format": "at-identifier", + "description": "Filter by DID or handle (triggers on-demand backfill)" + }, + "profiles": { + "type": "boolean", + "description": "Include profile + identity info keyed by DID" + }, + "spaceUri": { + "type": "string", + "format": "at-uri", + "description": "If set, query records inside this permissioned space (requires service-auth JWT or a read-grant invite token)." + }, + "byUser": { + "type": "string", + "format": "did", + "description": "Only used with spaceUri — filter to records authored by this DID." + }, + "inviteToken": { + "type": "string", + "description": "Read-grant invite token for anonymous bearer access. Replaces JWT auth when supplied." + }, + "communityDid": { + "type": "string", + "description": "Filter by communityDid" + }, + "name": { + "type": "string", + "description": "Filter by name" + }, + "description": { + "type": "string", + "description": "Filter by description" + }, + "createdAtMin": { + "type": "string", + "description": "Minimum value for createdAt" + }, + "createdAtMax": { + "type": "string", + "description": "Maximum value for createdAt" + }, + "sort": { + "type": "string", + "knownValues": [ + "communityDid", + "name", + "description", + "createdAt" + ], + "description": "Field to sort by (default: time_us)" + }, + "order": { + "type": "string", + "knownValues": [ + "asc", + "desc" + ], + "description": "Sort direction (default: desc for dates/numbers/counts, asc for strings)" + } + } + }, + "output": { + "encoding": "text/event-stream", + "schema": { + "type": "object", + "properties": {} + } + }, + "errors": [ + { + "name": "InvalidRequest" + }, + { + "name": "AuthRequired" + }, + { + "name": "Forbidden" + }, + { + "name": "NotFound" + }, + { + "name": "NotSupported" + } + ] + } + } +} diff --git a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/invite/create.json b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/invite/create.json similarity index 97% rename from examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/invite/create.json rename to examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/invite/create.json index dc058ac..f6ed9fe 100644 --- a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/invite/create.json +++ b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/invite/create.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "tools.atmo.chat.space.invite.create", + "id": "tools.atmo.chat.spaceExt.invite.create", "defs": { "main": { "type": "procedure", diff --git a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/invite/list.json b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/invite/list.json similarity index 95% rename from examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/invite/list.json rename to examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/invite/list.json index aa3924f..d5191da 100644 --- a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/invite/list.json +++ b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/invite/list.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "tools.atmo.chat.space.invite.list", + "id": "tools.atmo.chat.spaceExt.invite.list", "defs": { "main": { "type": "query", diff --git a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/invite/redeem.json b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/invite/redeem.json similarity index 94% rename from examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/invite/redeem.json rename to examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/invite/redeem.json index 8398704..9d9e552 100644 --- a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/invite/redeem.json +++ b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/invite/redeem.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "tools.atmo.chat.space.invite.redeem", + "id": "tools.atmo.chat.spaceExt.invite.redeem", "defs": { "main": { "type": "procedure", diff --git a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/invite/revoke.json b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/invite/revoke.json similarity index 95% rename from examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/invite/revoke.json rename to examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/invite/revoke.json index 3ccbace..49d5d2e 100644 --- a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/invite/revoke.json +++ b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/invite/revoke.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "tools.atmo.chat.space.invite.revoke", + "id": "tools.atmo.chat.spaceExt.invite.revoke", "defs": { "main": { "type": "procedure", diff --git a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/whoami.json b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/whoami.json similarity index 95% rename from examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/whoami.json rename to examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/whoami.json index 8e4f95f..83d5849 100644 --- a/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/space/whoami.json +++ b/examples/sveltekit-group-chat/lexicons-generated/tools/atmo/chat/spaceExt/whoami.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "tools.atmo.chat.space.whoami", + "id": "tools.atmo.chat.spaceExt.whoami", "defs": { "main": { "type": "query", diff --git a/examples/sveltekit-group-chat/src/lib/atproto/generated-methods.ts b/examples/sveltekit-group-chat/src/lib/atproto/generated-methods.ts index e5d4872..0cfd48f 100644 --- a/examples/sveltekit-group-chat/src/lib/atproto/generated-methods.ts +++ b/examples/sveltekit-group-chat/src/lib/atproto/generated-methods.ts @@ -4,6 +4,7 @@ export const xrpcMethods = [ "tools.atmo.chat.channel.getRecord", "tools.atmo.chat.channel.listRecords", + "tools.atmo.chat.channel.watchRecords", "tools.atmo.chat.community.adopt", "tools.atmo.chat.community.delete", "tools.atmo.chat.community.deleteRecord", @@ -25,27 +26,25 @@ export const xrpcMethods = [ "tools.atmo.chat.community.space.resync", "tools.atmo.chat.community.space.revoke", "tools.atmo.chat.community.space.setAccessLevel", - "tools.atmo.chat.community.whoami", + "tools.atmo.chat.community.space.whoami", "tools.atmo.chat.getCursor", "tools.atmo.chat.getOverview", "tools.atmo.chat.getProfile", "tools.atmo.chat.message.getRecord", "tools.atmo.chat.message.listRecords", + "tools.atmo.chat.message.watchRecords", "tools.atmo.chat.notifyOfUpdate", "tools.atmo.chat.realtime.subscribe", "tools.atmo.chat.realtime.ticket", "tools.atmo.chat.server.getRecord", "tools.atmo.chat.server.listRecords", + "tools.atmo.chat.server.watchRecords", "tools.atmo.chat.space.addMember", "tools.atmo.chat.space.createSpace", "tools.atmo.chat.space.deleteRecord", "tools.atmo.chat.space.getBlob", "tools.atmo.chat.space.getRecord", "tools.atmo.chat.space.getSpace", - "tools.atmo.chat.space.invite.create", - "tools.atmo.chat.space.invite.list", - "tools.atmo.chat.space.invite.redeem", - "tools.atmo.chat.space.invite.revoke", "tools.atmo.chat.space.leaveSpace", "tools.atmo.chat.space.listBlobs", "tools.atmo.chat.space.listMembers", @@ -54,5 +53,9 @@ export const xrpcMethods = [ "tools.atmo.chat.space.putRecord", "tools.atmo.chat.space.removeMember", "tools.atmo.chat.space.uploadBlob", - "tools.atmo.chat.space.whoami" + "tools.atmo.chat.spaceExt.invite.create", + "tools.atmo.chat.spaceExt.invite.list", + "tools.atmo.chat.spaceExt.invite.redeem", + "tools.atmo.chat.spaceExt.invite.revoke", + "tools.atmo.chat.spaceExt.whoami" ] as const; diff --git a/examples/sveltekit-group-chat/src/lib/contrail/index.ts b/examples/sveltekit-group-chat/src/lib/contrail/index.ts index 0c0b382..13f2473 100644 --- a/examples/sveltekit-group-chat/src/lib/contrail/index.ts +++ b/examples/sveltekit-group-chat/src/lib/contrail/index.ts @@ -52,23 +52,20 @@ function build(env: Env): Bundle { maxSize: 2 * 1024 * 1024, accept: ['image/png', 'image/jpeg', 'image/webp', 'image/gif'] }, - // Dev-only bypass: trust the HMAC-signed `did` cookie the OAuth flow - // sets, in lieu of a service-auth JWT. bsky.social rejects - // `getServiceAuth` for loopback clients, so without this the demo - // would need a tunnel. NEVER set DEV_AUTH=1 in prod. + // Dev-only bypass: trust the HMAC-signed session cookie (or an + // X-Dev-Did header for server-side synthetic requests) as auth, so + // the loopback OAuth client doesn't need to round-trip through the + // user's PDS for getServiceAuth (bsky.social rejects that on + // loopback). In prod, DEV_AUTH is unset and every caller — browser + // or third-party — must present a real atproto service-auth JWT. + // Browsers get one by calling our SvelteKit helper which uses the + // OAuth session to mint via the user's PDS. authOverride: devAuth ? (req: Request) => { - // Accept either the HMAC-signed session cookie (from the - // browser via /xrpc/...) or an X-Dev-Did header (from the - // server-side remote helpers, which construct synthetic - // requests and already trust the session themselves). const headerDid = req.headers.get('x-dev-did'); const did = headerDid ?? getSignedCookieFromRequest(req, 'did'); if (!did) return null; - return { - issuer: did, - audience: env.SERVICE_DID - }; + return { issuer: did, audience: env.SERVICE_DID }; } : undefined }, diff --git a/examples/sveltekit-group-chat/src/lib/rooms/connection.svelte.ts b/examples/sveltekit-group-chat/src/lib/rooms/connection.svelte.ts new file mode 100644 index 0000000..831e4e8 --- /dev/null +++ b/examples/sveltekit-group-chat/src/lib/rooms/connection.svelte.ts @@ -0,0 +1,35 @@ +/** Cross-component connection status for the realtime subscription powering + * the current view. The channel page updates this as its WatchQuery status + * changes; the server navbar renders a tiny colored dot for it. */ + +import type { WatchStoreStatus } from '@atmo-dev/contrail/sync'; + +export const connection = $state<{ status: WatchStoreStatus }>({ status: 'idle' }); + +export function setConnectionStatus(status: WatchStoreStatus): void { + connection.status = status; +} + +export function resetConnectionStatus(): void { + connection.status = 'idle'; +} + +/** Map status → (color, label) for the UI dot. */ +export function connectionIndicator(status: WatchStoreStatus): { + color: 'green' | 'orange' | 'red' | 'gray'; + label: string; +} { + switch (status) { + case 'live': + return { color: 'green', label: 'Connected' }; + case 'connecting': + case 'snapshot': + case 'reconnecting': + return { color: 'orange', label: 'Connecting…' }; + case 'closed': + return { color: 'red', label: 'Disconnected' }; + case 'idle': + default: + return { color: 'gray', label: 'Idle' }; + } +} diff --git a/examples/sveltekit-group-chat/src/lib/rooms/realtime.svelte.ts b/examples/sveltekit-group-chat/src/lib/rooms/realtime.svelte.ts index 3b03d1f..70bd04a 100644 --- a/examples/sveltekit-group-chat/src/lib/rooms/realtime.svelte.ts +++ b/examples/sveltekit-group-chat/src/lib/rooms/realtime.svelte.ts @@ -110,37 +110,42 @@ export function connectCommunityRealtime(communityDid: string): () => void { } catch { return; } + // Payload shape mirrors listRecords output (uri/did/collection/rkey/cid/record/time_us/space?). + // `space` is only set for space records — which is what we filter this community stream to. if (kind === 'record.created') { const p = ev.payload as { - spaceUri: string; + uri: string; + did: string; collection: string; - authorDid: string; rkey: string; record: Record; - createdAt: number | string; + time_us: number; + space?: string; }; + if (!p.space) return; if (p.collection === 'tools.atmo.chat.message') { const rec = p.record as { text?: string; createdAt?: string; replyTo?: string }; if (rec.text && rec.createdAt) { - channelMessages.append(p.spaceUri, { + channelMessages.append(p.space, { rkey: p.rkey, - authorDid: p.authorDid, + authorDid: p.did, text: rec.text, createdAt: rec.createdAt, replyTo: rec.replyTo }); // Bump unread if we're not currently on this channel. - if (!isCurrentChannel(p.spaceUri)) { - bumpUnread(p.spaceUri, rec.createdAt); + if (!isCurrentChannel(p.space)) { + bumpUnread(p.space, rec.createdAt); } } } else if (p.collection === 'tools.atmo.chat.channel') { void invalidateAll(); } } else if (kind === 'record.deleted') { - const p = ev.payload as { spaceUri: string; collection: string; rkey: string }; + const p = ev.payload as { uri: string; did: string; collection: string; rkey: string; space?: string }; + if (!p.space) return; if (p.collection === 'tools.atmo.chat.message') { - channelMessages.remove(p.spaceUri, p.rkey); + channelMessages.remove(p.space, p.rkey); } else if (p.collection === 'tools.atmo.chat.channel') { void invalidateAll(); } diff --git a/examples/sveltekit-group-chat/src/lib/rooms/server.ts b/examples/sveltekit-group-chat/src/lib/rooms/server.ts index a8b429d..c8e0e99 100644 --- a/examples/sveltekit-group-chat/src/lib/rooms/server.ts +++ b/examples/sveltekit-group-chat/src/lib/rooms/server.ts @@ -213,7 +213,7 @@ export async function whoami( ctx: AuthedCallContext, query: { spaceUri: string } ): Promise<{ spaceUri: string; accessLevel: string | null }> { - return callContrail(ctx, 'tools.atmo.chat.community.whoami', { query }); + return callContrail(ctx, 'tools.atmo.chat.community.space.whoami', { query }); } // --- realtime -------------------------------------------------------------- diff --git a/examples/sveltekit-group-chat/src/routes/api/watch-token/+server.ts b/examples/sveltekit-group-chat/src/routes/api/watch-token/+server.ts new file mode 100644 index 0000000..a54999c --- /dev/null +++ b/examples/sveltekit-group-chat/src/routes/api/watch-token/+server.ts @@ -0,0 +1,25 @@ +import { error, json } from '@sveltejs/kit'; +import type { RequestHandler } from './$types'; + +/** POST { lxm } → { token } — mint an atproto service-auth JWT scoped to one + * XRPC method, using the caller's OAuth session to call getServiceAuth on + * their PDS. Used by the browser sync engine to auth the watchRecords + * handshake fetch — it can't mint JWTs itself, but same-origin to us it + * can delegate. + * + * Cross-origin apps skip this and mint their own JWTs server-side. */ +export const POST: RequestHandler = async ({ request, locals, platform }) => { + if (!locals.did || !locals.client) error(401, 'Not authenticated'); + const body = (await request.json().catch(() => null)) as { lxm?: string } | null; + if (!body?.lxm) error(400, 'lxm required'); + + const res = await locals.client.get('com.atproto.server.getServiceAuth', { + params: { + aud: platform!.env.SERVICE_DID as `did:${string}:${string}`, + lxm: body.lxm as `${string}.${string}.${string}`, + exp: Math.floor(Date.now() / 1000) + 120 + } + }); + if (!res.ok) error(502, `getServiceAuth failed: ${JSON.stringify(res.data)}`); + return json({ token: (res.data as { token: string }).token }); +}; diff --git a/examples/sveltekit-group-chat/src/routes/c/[communityDid]/+layout.server.ts b/examples/sveltekit-group-chat/src/routes/c/[communityDid]/+layout.server.ts index f444a2d..d3f554a 100644 --- a/examples/sveltekit-group-chat/src/routes/c/[communityDid]/+layout.server.ts +++ b/examples/sveltekit-group-chat/src/routes/c/[communityDid]/+layout.server.ts @@ -118,7 +118,7 @@ export const load: LayoutServerLoad = async ({ locals, params, platform }) => { const adminUri = buildAdminUri(communityDid); const d = await authedFetch<{ accessLevel: string | null }>( ctx, - 'tools.atmo.chat.community.whoami', + 'tools.atmo.chat.community.space.whoami', { query: { spaceUri: adminUri } } ); isAdmin = d.accessLevel === 'admin' || d.accessLevel === 'owner'; diff --git a/examples/sveltekit-group-chat/src/routes/c/[communityDid]/+layout.svelte b/examples/sveltekit-group-chat/src/routes/c/[communityDid]/+layout.svelte index 34c8e64..27980bc 100644 --- a/examples/sveltekit-group-chat/src/routes/c/[communityDid]/+layout.svelte +++ b/examples/sveltekit-group-chat/src/routes/c/[communityDid]/+layout.svelte @@ -7,6 +7,7 @@ import { createChannel } from '$lib/rooms/rooms.remote'; import { unread, resetUnread } from '$lib/rooms/unread.svelte'; import { connectCommunityRealtime } from '$lib/rooms/realtime.svelte'; + import { connection, connectionIndicator } from '$lib/rooms/connection.svelte'; let { data, children } = $props(); @@ -116,6 +117,24 @@ {data.server?.name ?? 'server'} {/if} + + {@const ind = connectionIndicator(connection.status)} + + + + diff --git a/examples/sveltekit-group-chat/src/routes/c/[communityDid]/[channelKey]/+page.svelte b/examples/sveltekit-group-chat/src/routes/c/[communityDid]/[channelKey]/+page.svelte index fa27463..4814463 100644 --- a/examples/sveltekit-group-chat/src/routes/c/[communityDid]/[channelKey]/+page.svelte +++ b/examples/sveltekit-group-chat/src/routes/c/[communityDid]/[channelKey]/+page.svelte @@ -9,6 +9,7 @@ import { createWatchQuery } from '$lib/rooms/watch.svelte'; import type { WatchRecord } from '@atmo-dev/contrail/sync'; import { dev } from '$app/environment'; + import { setConnectionStatus, resetConnectionStatus } from '$lib/rooms/connection.svelte'; let { data } = $props(); @@ -18,22 +19,46 @@ // Live message feed via contrail's watchRecords subscription. The engine // handles the snapshot + live merge; we just render its `records` array. // Sorted newest-first by default — flip to oldest-first for chat. - let query = $derived.by(() => - createWatchQuery({ - url: `/xrpc/tools.atmo.chat.message.watchRecords?spaceUri=${encodeURIComponent(data.spaceUri)}&limit=50`, - // In prod the server has a DurableObjectPubSub — open a DO-terminated - // WS so idle connections hibernate (near-zero cost at CF scale). - // In dev the server uses InMemoryPubSub (one isolate), which only - // speaks SSE, so downgrade. + // + // The query is keyed strictly on spaceUri — recreating it on every + // `data` identity change would tear down and rebuild the subscription on + // unrelated page-data invalidations (e.g. channel list refresh), flashing + // "Loading…" each time. Using a keyed effect here avoids that. + let query = $state | null>(null); + let currentUri: string | null = null; + + $effect(() => { + const uri = data.spaceUri; + if (uri === currentUri) return; + query?.stop(); + currentUri = uri; + query = createWatchQuery({ + url: `/xrpc/tools.atmo.chat.message.watchRecords?spaceUri=${encodeURIComponent(uri)}&limit=50`, transport: dev ? 'sse' : 'ws', + fetchAuthToken: dev + ? undefined + : async () => { + const res = await fetch('/api/watch-token', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ lxm: 'tools.atmo.chat.message.watchRecords' }) + }); + if (!res.ok) throw new Error(`watch-token mint failed: ${res.status}`); + const d = (await res.json()) as { token: string }; + return d.token; + }, compareRecords: (a: WatchRecord, b: WatchRecord) => (a.time_us ?? 0) - (b.time_us ?? 0) - }) - ); + }); + }); - // Teardown on channel change. + // Final teardown on component unmount. $effect(() => { - return () => query.stop(); + return () => { + query?.stop(); + query = null; + currentUri = null; + }; }); // Unread / current-channel bookkeeping. @@ -43,9 +68,16 @@ return () => setCurrentChannel(null); }); + // Mirror the query's connection status into the shared store so the + // layout's navbar can render a dot for it. + $effect(() => { + if (query) setConnectionStatus(query.status); + return () => resetConnectionStatus(); + }); + // Project records to the shape the existing template expects. let messages = $derived( - query.records.map((r) => { + (query?.records ?? []).map((r) => { const rec = r.record as { text?: string; createdAt?: string; replyTo?: string }; return { rkey: r.rkey, @@ -94,7 +126,7 @@
- {#if query.status === 'connecting' || query.status === 'snapshot'} + {#if messages.length === 0 && (query?.status === 'connecting' || query?.status === 'snapshot' || query?.status === 'idle')}
Loading…
{:else if messages.length === 0}
No messages yet. Say hi.
diff --git a/lexicons-generated/rsvp/atmo/community/defs.json b/lexicons-generated/rsvp/atmo/community/defs.json index 1471660..6f64581 100644 --- a/lexicons-generated/rsvp/atmo/community/defs.json +++ b/lexicons-generated/rsvp/atmo/community/defs.json @@ -130,6 +130,56 @@ "type": "integer" } } + }, + "inviteView": { + "type": "object", + "required": [ + "tokenHash", + "spaceUri", + "accessLevel", + "createdBy", + "createdAt", + "usedCount" + ], + "properties": { + "tokenHash": { + "type": "string", + "description": "SHA-256 of the raw token. Stable id for list/revoke; never grants access on its own." + }, + "spaceUri": { + "type": "string", + "format": "at-uri" + }, + "accessLevel": { + "type": "ref", + "ref": "#accessLevel" + }, + "createdBy": { + "type": "string", + "format": "did" + }, + "createdAt": { + "type": "integer" + }, + "expiresAt": { + "type": "integer", + "description": "Unix ms. Omitted for no expiry." + }, + "maxUses": { + "type": "integer", + "description": "Omitted for unlimited." + }, + "usedCount": { + "type": "integer" + }, + "revokedAt": { + "type": "integer", + "description": "Unix ms. Omitted if not revoked." + }, + "note": { + "type": "string" + } + } } } } diff --git a/lexicons-generated/rsvp/atmo/community/invite/create.json b/lexicons-generated/rsvp/atmo/community/invite/create.json new file mode 100644 index 0000000..9b4d211 --- /dev/null +++ b/lexicons-generated/rsvp/atmo/community/invite/create.json @@ -0,0 +1,93 @@ +{ + "lexicon": 1, + "id": "rsvp.atmo.community.invite.create", + "defs": { + "main": { + "type": "procedure", + "description": "Create an invite for a community-owned space. Caller must have manager+ on the target space and cannot grant a higher level than their own. Returns the raw token once; only its SHA-256 is stored.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": [ + "spaceUri", + "accessLevel" + ], + "properties": { + "spaceUri": { + "type": "string", + "format": "at-uri" + }, + "accessLevel": { + "type": "ref", + "ref": "rsvp.atmo.community.defs#accessLevel" + }, + "expiresAt": { + "type": "integer", + "description": "Unix ms. Omit for no expiry." + }, + "maxUses": { + "type": "integer", + "minimum": 1, + "description": "Omit for unlimited redemptions." + }, + "note": { + "type": "string", + "maxLength": 500 + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": [ + "token", + "tokenHash", + "spaceUri", + "accessLevel", + "createdAt" + ], + "properties": { + "token": { + "type": "string", + "description": "Raw token. Shown once — cannot be retrieved later." + }, + "tokenHash": { + "type": "string" + }, + "spaceUri": { + "type": "string", + "format": "at-uri" + }, + "accessLevel": { + "type": "ref", + "ref": "rsvp.atmo.community.defs#accessLevel" + }, + "expiresAt": { + "type": "integer" + }, + "maxUses": { + "type": "integer" + }, + "createdAt": { + "type": "integer" + } + } + } + }, + "errors": [ + { + "name": "InvalidRequest" + }, + { + "name": "Forbidden" + }, + { + "name": "NotFound" + } + ] + } + } +} diff --git a/lexicons-generated/rsvp/atmo/community/invite/list.json b/lexicons-generated/rsvp/atmo/community/invite/list.json new file mode 100644 index 0000000..7bd0010 --- /dev/null +++ b/lexicons-generated/rsvp/atmo/community/invite/list.json @@ -0,0 +1,52 @@ +{ + "lexicon": 1, + "id": "rsvp.atmo.community.invite.list", + "defs": { + "main": { + "type": "query", + "description": "List invites for a community-owned space. Caller must have manager+ on the space. Raw tokens are never returned — tokenHash is the stable id.", + "parameters": { + "type": "params", + "required": [ + "spaceUri" + ], + "properties": { + "spaceUri": { + "type": "string", + "format": "at-uri" + }, + "includeRevoked": { + "type": "boolean", + "default": false + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": [ + "invites" + ], + "properties": { + "invites": { + "type": "array", + "items": { + "type": "ref", + "ref": "rsvp.atmo.community.defs#inviteView" + } + } + } + } + }, + "errors": [ + { + "name": "Forbidden" + }, + { + "name": "NotFound" + } + ] + } + } +} diff --git a/lexicons-generated/rsvp/atmo/community/invite/redeem.json b/lexicons-generated/rsvp/atmo/community/invite/redeem.json new file mode 100644 index 0000000..0995e9b --- /dev/null +++ b/lexicons-generated/rsvp/atmo/community/invite/redeem.json @@ -0,0 +1,60 @@ +{ + "lexicon": 1, + "id": "rsvp.atmo.community.invite.redeem", + "defs": { + "main": { + "type": "procedure", + "description": "Redeem a community invite. The caller's DID (from service-auth) is granted the access level encoded in the invite. Atomic — a token is consumed only if the invite is currently usable.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": [ + "token" + ], + "properties": { + "token": { + "type": "string" + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": [ + "spaceUri", + "accessLevel", + "communityDid" + ], + "properties": { + "spaceUri": { + "type": "string", + "format": "at-uri" + }, + "accessLevel": { + "type": "ref", + "ref": "rsvp.atmo.community.defs#accessLevel" + }, + "communityDid": { + "type": "string", + "format": "did" + } + } + } + }, + "errors": [ + { + "name": "InvalidRequest" + }, + { + "name": "InvalidToken" + }, + { + "name": "NotFound" + } + ] + } + } +} diff --git a/lexicons-generated/rsvp/atmo/community/invite/revoke.json b/lexicons-generated/rsvp/atmo/community/invite/revoke.json new file mode 100644 index 0000000..c7aa335 --- /dev/null +++ b/lexicons-generated/rsvp/atmo/community/invite/revoke.json @@ -0,0 +1,46 @@ +{ + "lexicon": 1, + "id": "rsvp.atmo.community.invite.revoke", + "defs": { + "main": { + "type": "procedure", + "description": "Revoke an invite. Caller must be the invite's creator or have manager+ on the target space.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": [ + "tokenHash" + ], + "properties": { + "tokenHash": { + "type": "string" + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": [ + "ok" + ], + "properties": { + "ok": { + "type": "boolean" + } + } + } + }, + "errors": [ + { + "name": "NotFound" + }, + { + "name": "Forbidden" + } + ] + } + } +} diff --git a/lexicons-generated/rsvp/atmo/community/whoami.json b/lexicons-generated/rsvp/atmo/community/space/whoami.json similarity index 95% rename from lexicons-generated/rsvp/atmo/community/whoami.json rename to lexicons-generated/rsvp/atmo/community/space/whoami.json index 691d361..524345d 100644 --- a/lexicons-generated/rsvp/atmo/community/whoami.json +++ b/lexicons-generated/rsvp/atmo/community/space/whoami.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "rsvp.atmo.community.whoami", + "id": "rsvp.atmo.community.space.whoami", "defs": { "main": { "type": "query", diff --git a/lexicons-generated/rsvp/atmo/event/getRecord.json b/lexicons-generated/rsvp/atmo/event/getRecord.json index d2dc8a6..ab6b24d 100644 --- a/lexicons-generated/rsvp/atmo/event/getRecord.json +++ b/lexicons-generated/rsvp/atmo/event/getRecord.json @@ -213,11 +213,8 @@ "type": "string" }, "record": { - "type": "union", - "refs": [ - "#appBskyActorProfile", - "#siteStandardPublication" - ] + "type": "ref", + "ref": "#appBskyActorProfile" } } }, @@ -283,57 +280,6 @@ "type": "ref" } } - }, - "siteStandardPublication": { - "type": "object", - "required": [ - "url", - "name" - ], - "properties": { - "url": { - "type": "string", - "format": "uri", - "description": "Base publication url (ex: https://standard.site). The canonical document URL is formed by combining this value with the document path." - }, - "icon": { - "type": "blob", - "accept": [ - "image/*" - ], - "maxSize": 1000000, - "description": "Square image to identify the publication. Should be at least 256x256." - }, - "name": { - "type": "string", - "maxLength": 5000, - "description": "Name of the publication.", - "maxGraphemes": 500 - }, - "labels": { - "refs": [ - "com.atproto.label.defs#selfLabels" - ], - "type": "union", - "description": "Self-label values for this publication. Effectively content warnings." - }, - "basicTheme": { - "ref": "site.standard.theme.basic", - "type": "ref", - "description": "Simplified publication theme for tools and apps to utilize when displaying content." - }, - "description": { - "type": "string", - "maxLength": 30000, - "description": "Brief description of the publication.", - "maxGraphemes": 3000 - }, - "preferences": { - "ref": "#preferences", - "type": "ref", - "description": "Object containing platform specific preferences (with a few shared properties)." - } - } } } } diff --git a/lexicons-generated/rsvp/atmo/event/listRecords.json b/lexicons-generated/rsvp/atmo/event/listRecords.json index 3f938e0..1c68aee 100644 --- a/lexicons-generated/rsvp/atmo/event/listRecords.json +++ b/lexicons-generated/rsvp/atmo/event/listRecords.json @@ -327,11 +327,8 @@ "type": "string" }, "record": { - "type": "union", - "refs": [ - "#appBskyActorProfile", - "#siteStandardPublication" - ] + "type": "ref", + "ref": "#appBskyActorProfile" } } }, @@ -397,57 +394,6 @@ "type": "ref" } } - }, - "siteStandardPublication": { - "type": "object", - "required": [ - "url", - "name" - ], - "properties": { - "url": { - "type": "string", - "format": "uri", - "description": "Base publication url (ex: https://standard.site). The canonical document URL is formed by combining this value with the document path." - }, - "icon": { - "type": "blob", - "accept": [ - "image/*" - ], - "maxSize": 1000000, - "description": "Square image to identify the publication. Should be at least 256x256." - }, - "name": { - "type": "string", - "maxLength": 5000, - "description": "Name of the publication.", - "maxGraphemes": 500 - }, - "labels": { - "refs": [ - "com.atproto.label.defs#selfLabels" - ], - "type": "union", - "description": "Self-label values for this publication. Effectively content warnings." - }, - "basicTheme": { - "ref": "site.standard.theme.basic", - "type": "ref", - "description": "Simplified publication theme for tools and apps to utilize when displaying content." - }, - "description": { - "type": "string", - "maxLength": 30000, - "description": "Brief description of the publication.", - "maxGraphemes": 3000 - }, - "preferences": { - "ref": "#preferences", - "type": "ref", - "description": "Object containing platform specific preferences (with a few shared properties)." - } - } } } } diff --git a/lexicons-generated/rsvp/atmo/getProfile.json b/lexicons-generated/rsvp/atmo/getProfile.json index d6c6191..d6c1fe5 100644 --- a/lexicons-generated/rsvp/atmo/getProfile.json +++ b/lexicons-generated/rsvp/atmo/getProfile.json @@ -65,11 +65,8 @@ "type": "string" }, "record": { - "type": "union", - "refs": [ - "#appBskyActorProfile", - "#siteStandardPublication" - ] + "type": "ref", + "ref": "#appBskyActorProfile" } } }, @@ -135,57 +132,6 @@ "type": "ref" } } - }, - "siteStandardPublication": { - "type": "object", - "required": [ - "url", - "name" - ], - "properties": { - "url": { - "type": "string", - "format": "uri", - "description": "Base publication url (ex: https://standard.site). The canonical document URL is formed by combining this value with the document path." - }, - "icon": { - "type": "blob", - "accept": [ - "image/*" - ], - "maxSize": 1000000, - "description": "Square image to identify the publication. Should be at least 256x256." - }, - "name": { - "type": "string", - "maxLength": 5000, - "description": "Name of the publication.", - "maxGraphemes": 500 - }, - "labels": { - "refs": [ - "com.atproto.label.defs#selfLabels" - ], - "type": "union", - "description": "Self-label values for this publication. Effectively content warnings." - }, - "basicTheme": { - "ref": "site.standard.theme.basic", - "type": "ref", - "description": "Simplified publication theme for tools and apps to utilize when displaying content." - }, - "description": { - "type": "string", - "maxLength": 30000, - "description": "Brief description of the publication.", - "maxGraphemes": 3000 - }, - "preferences": { - "ref": "#preferences", - "type": "ref", - "description": "Object containing platform specific preferences (with a few shared properties)." - } - } } } } diff --git a/lexicons-generated/rsvp/atmo/permissionSet.json b/lexicons-generated/rsvp/atmo/permissionSet.json index 9ac2975..5fae345 100644 --- a/lexicons-generated/rsvp/atmo/permissionSet.json +++ b/lexicons-generated/rsvp/atmo/permissionSet.json @@ -16,6 +16,10 @@ "rsvp.atmo.community.delete", "rsvp.atmo.community.deleteRecord", "rsvp.atmo.community.getHealth", + "rsvp.atmo.community.invite.create", + "rsvp.atmo.community.invite.list", + "rsvp.atmo.community.invite.redeem", + "rsvp.atmo.community.invite.revoke", "rsvp.atmo.community.list", "rsvp.atmo.community.mint", "rsvp.atmo.community.putRecord", @@ -29,7 +33,7 @@ "rsvp.atmo.community.space.resync", "rsvp.atmo.community.space.revoke", "rsvp.atmo.community.space.setAccessLevel", - "rsvp.atmo.community.whoami", + "rsvp.atmo.community.space.whoami", "rsvp.atmo.event.getRecord", "rsvp.atmo.event.listRecords", "rsvp.atmo.getCursor", @@ -41,19 +45,22 @@ "rsvp.atmo.space.addMember", "rsvp.atmo.space.createSpace", "rsvp.atmo.space.deleteRecord", + "rsvp.atmo.space.getBlob", "rsvp.atmo.space.getRecord", "rsvp.atmo.space.getSpace", - "rsvp.atmo.space.invite.create", - "rsvp.atmo.space.invite.list", - "rsvp.atmo.space.invite.redeem", - "rsvp.atmo.space.invite.revoke", "rsvp.atmo.space.leaveSpace", + "rsvp.atmo.space.listBlobs", "rsvp.atmo.space.listMembers", "rsvp.atmo.space.listRecords", "rsvp.atmo.space.listSpaces", "rsvp.atmo.space.putRecord", "rsvp.atmo.space.removeMember", - "rsvp.atmo.space.whoami" + "rsvp.atmo.space.uploadBlob", + "rsvp.atmo.spaceExt.invite.create", + "rsvp.atmo.spaceExt.invite.list", + "rsvp.atmo.spaceExt.invite.redeem", + "rsvp.atmo.spaceExt.invite.revoke", + "rsvp.atmo.spaceExt.whoami" ] } ] diff --git a/lexicons-generated/rsvp/atmo/rsvp/getRecord.json b/lexicons-generated/rsvp/atmo/rsvp/getRecord.json index d63df47..6bb1bcf 100644 --- a/lexicons-generated/rsvp/atmo/rsvp/getRecord.json +++ b/lexicons-generated/rsvp/atmo/rsvp/getRecord.json @@ -162,11 +162,8 @@ "type": "string" }, "record": { - "type": "union", - "refs": [ - "#appBskyActorProfile", - "#siteStandardPublication" - ] + "type": "ref", + "ref": "#appBskyActorProfile" } } }, @@ -232,57 +229,6 @@ "type": "ref" } } - }, - "siteStandardPublication": { - "type": "object", - "required": [ - "url", - "name" - ], - "properties": { - "url": { - "type": "string", - "format": "uri", - "description": "Base publication url (ex: https://standard.site). The canonical document URL is formed by combining this value with the document path." - }, - "icon": { - "type": "blob", - "accept": [ - "image/*" - ], - "maxSize": 1000000, - "description": "Square image to identify the publication. Should be at least 256x256." - }, - "name": { - "type": "string", - "maxLength": 5000, - "description": "Name of the publication.", - "maxGraphemes": 500 - }, - "labels": { - "refs": [ - "com.atproto.label.defs#selfLabels" - ], - "type": "union", - "description": "Self-label values for this publication. Effectively content warnings." - }, - "basicTheme": { - "ref": "site.standard.theme.basic", - "type": "ref", - "description": "Simplified publication theme for tools and apps to utilize when displaying content." - }, - "description": { - "type": "string", - "maxLength": 30000, - "description": "Brief description of the publication.", - "maxGraphemes": 3000 - }, - "preferences": { - "ref": "#preferences", - "type": "ref", - "description": "Object containing platform specific preferences (with a few shared properties)." - } - } } } } diff --git a/lexicons-generated/rsvp/atmo/rsvp/listRecords.json b/lexicons-generated/rsvp/atmo/rsvp/listRecords.json index d216597..3491b78 100644 --- a/lexicons-generated/rsvp/atmo/rsvp/listRecords.json +++ b/lexicons-generated/rsvp/atmo/rsvp/listRecords.json @@ -215,11 +215,8 @@ "type": "string" }, "record": { - "type": "union", - "refs": [ - "#appBskyActorProfile", - "#siteStandardPublication" - ] + "type": "ref", + "ref": "#appBskyActorProfile" } } }, @@ -285,57 +282,6 @@ "type": "ref" } } - }, - "siteStandardPublication": { - "type": "object", - "required": [ - "url", - "name" - ], - "properties": { - "url": { - "type": "string", - "format": "uri", - "description": "Base publication url (ex: https://standard.site). The canonical document URL is formed by combining this value with the document path." - }, - "icon": { - "type": "blob", - "accept": [ - "image/*" - ], - "maxSize": 1000000, - "description": "Square image to identify the publication. Should be at least 256x256." - }, - "name": { - "type": "string", - "maxLength": 5000, - "description": "Name of the publication.", - "maxGraphemes": 500 - }, - "labels": { - "refs": [ - "com.atproto.label.defs#selfLabels" - ], - "type": "union", - "description": "Self-label values for this publication. Effectively content warnings." - }, - "basicTheme": { - "ref": "site.standard.theme.basic", - "type": "ref", - "description": "Simplified publication theme for tools and apps to utilize when displaying content." - }, - "description": { - "type": "string", - "maxLength": 30000, - "description": "Brief description of the publication.", - "maxGraphemes": 3000 - }, - "preferences": { - "ref": "#preferences", - "type": "ref", - "description": "Object containing platform specific preferences (with a few shared properties)." - } - } } } } diff --git a/lexicons-generated/rsvp/atmo/space/invite/create.json b/lexicons-generated/rsvp/atmo/spaceExt/invite/create.json similarity index 97% rename from lexicons-generated/rsvp/atmo/space/invite/create.json rename to lexicons-generated/rsvp/atmo/spaceExt/invite/create.json index c60f793..b604074 100644 --- a/lexicons-generated/rsvp/atmo/space/invite/create.json +++ b/lexicons-generated/rsvp/atmo/spaceExt/invite/create.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "rsvp.atmo.space.invite.create", + "id": "rsvp.atmo.spaceExt.invite.create", "defs": { "main": { "type": "procedure", diff --git a/lexicons-generated/rsvp/atmo/space/invite/list.json b/lexicons-generated/rsvp/atmo/spaceExt/invite/list.json similarity index 96% rename from lexicons-generated/rsvp/atmo/space/invite/list.json rename to lexicons-generated/rsvp/atmo/spaceExt/invite/list.json index 848d4e2..304be52 100644 --- a/lexicons-generated/rsvp/atmo/space/invite/list.json +++ b/lexicons-generated/rsvp/atmo/spaceExt/invite/list.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "rsvp.atmo.space.invite.list", + "id": "rsvp.atmo.spaceExt.invite.list", "defs": { "main": { "type": "query", diff --git a/lexicons-generated/rsvp/atmo/space/invite/redeem.json b/lexicons-generated/rsvp/atmo/spaceExt/invite/redeem.json similarity index 95% rename from lexicons-generated/rsvp/atmo/space/invite/redeem.json rename to lexicons-generated/rsvp/atmo/spaceExt/invite/redeem.json index 06261e9..b0fa07e 100644 --- a/lexicons-generated/rsvp/atmo/space/invite/redeem.json +++ b/lexicons-generated/rsvp/atmo/spaceExt/invite/redeem.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "rsvp.atmo.space.invite.redeem", + "id": "rsvp.atmo.spaceExt.invite.redeem", "defs": { "main": { "type": "procedure", diff --git a/lexicons-generated/rsvp/atmo/space/invite/revoke.json b/lexicons-generated/rsvp/atmo/spaceExt/invite/revoke.json similarity index 95% rename from lexicons-generated/rsvp/atmo/space/invite/revoke.json rename to lexicons-generated/rsvp/atmo/spaceExt/invite/revoke.json index afa34f2..40c4293 100644 --- a/lexicons-generated/rsvp/atmo/space/invite/revoke.json +++ b/lexicons-generated/rsvp/atmo/spaceExt/invite/revoke.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "rsvp.atmo.space.invite.revoke", + "id": "rsvp.atmo.spaceExt.invite.revoke", "defs": { "main": { "type": "procedure", diff --git a/lexicons-generated/rsvp/atmo/space/whoami.json b/lexicons-generated/rsvp/atmo/spaceExt/whoami.json similarity index 96% rename from lexicons-generated/rsvp/atmo/space/whoami.json rename to lexicons-generated/rsvp/atmo/spaceExt/whoami.json index 8eb98a2..0f5d078 100644 --- a/lexicons-generated/rsvp/atmo/space/whoami.json +++ b/lexicons-generated/rsvp/atmo/spaceExt/whoami.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "rsvp.atmo.space.whoami", + "id": "rsvp.atmo.spaceExt.whoami", "defs": { "main": { "type": "query", diff --git a/lexicons-pulled/README.md b/lexicons-pulled/README.md index 69dd796..378ad0f 100644 --- a/lexicons-pulled/README.md +++ b/lexicons-pulled/README.md @@ -2,4 +2,4 @@ this directory contains lexicon documents pulled from the following sources: -- atproto (nsids: app.bsky.actor.profile, community.lexicon.calendar.event, community.lexicon.calendar.rsvp, community.lexicon.location.address, community.lexicon.location.fsq, community.lexicon.location.geo, community.lexicon.location.hthree, site.standard.publication, site.standard.theme.basic, site.standard.theme.color) +- atproto (nsids: app.bsky.actor.profile, community.lexicon.calendar.event, community.lexicon.calendar.rsvp, community.lexicon.location.address, community.lexicon.location.fsq, community.lexicon.location.geo, community.lexicon.location.hthree) diff --git a/lexicons-pulled/site/standard/publication.json b/lexicons-pulled/site/standard/publication.json deleted file mode 100644 index 9656716..0000000 --- a/lexicons-pulled/site/standard/publication.json +++ /dev/null @@ -1,66 +0,0 @@ -{ - "id": "site.standard.publication", - "defs": { - "main": { - "key": "tid", - "type": "record", - "record": { - "type": "object", - "required": ["url", "name"], - "properties": { - "url": { - "type": "string", - "format": "uri", - "description": "Base publication url (ex: https://standard.site). The canonical document URL is formed by combining this value with the document path." - }, - "icon": { - "type": "blob", - "accept": ["image/*"], - "maxSize": 1000000, - "description": "Square image to identify the publication. Should be at least 256x256." - }, - "name": { - "type": "string", - "maxLength": 5000, - "description": "Name of the publication.", - "maxGraphemes": 500 - }, - "labels": { - "refs": ["com.atproto.label.defs#selfLabels"], - "type": "union", - "description": "Self-label values for this publication. Effectively content warnings." - }, - "basicTheme": { - "ref": "site.standard.theme.basic", - "type": "ref", - "description": "Simplified publication theme for tools and apps to utilize when displaying content." - }, - "description": { - "type": "string", - "maxLength": 30000, - "description": "Brief description of the publication.", - "maxGraphemes": 3000 - }, - "preferences": { - "ref": "#preferences", - "type": "ref", - "description": "Object containing platform specific preferences (with a few shared properties)." - } - } - }, - "description": "A publication record representing a blog, website, or content platform. Publications serve as containers for documents and define the overall branding and settings." - }, - "preferences": { - "type": "object", - "properties": { - "showInDiscover": { - "type": "boolean", - "default": true, - "description": "Boolean which decides whether the publication should appear in discovery feeds." - } - } - } - }, - "$type": "com.atproto.lexicon.schema", - "lexicon": 1 -} diff --git a/lexicons-pulled/site/standard/theme/basic.json b/lexicons-pulled/site/standard/theme/basic.json deleted file mode 100644 index 88ff587..0000000 --- a/lexicons-pulled/site/standard/theme/basic.json +++ /dev/null @@ -1,38 +0,0 @@ -{ - "id": "site.standard.theme.basic", - "defs": { - "main": { - "key": "tid", - "type": "record", - "record": { - "type": "object", - "required": ["background", "foreground", "accent", "accentForeground"], - "properties": { - "accent": { - "refs": ["site.standard.theme.color#rgb"], - "type": "union", - "description": "Color used for links and button backgrounds." - }, - "background": { - "refs": ["site.standard.theme.color#rgb"], - "type": "union", - "description": "Color used for content background." - }, - "foreground": { - "refs": ["site.standard.theme.color#rgb"], - "type": "union", - "description": "Color used for content text." - }, - "accentForeground": { - "refs": ["site.standard.theme.color#rgb"], - "type": "union", - "description": "Color used for button text." - } - } - }, - "description": "A simplified theme definition for publications, providing basic color customization for content display across different platforms and applications." - } - }, - "$type": "com.atproto.lexicon.schema", - "lexicon": 1 -} diff --git a/lexicons-pulled/site/standard/theme/color.json b/lexicons-pulled/site/standard/theme/color.json deleted file mode 100644 index 5dde553..0000000 --- a/lexicons-pulled/site/standard/theme/color.json +++ /dev/null @@ -1,54 +0,0 @@ -{ - "id": "site.standard.theme.color", - "defs": { - "rgb": { - "type": "object", - "required": ["r", "g", "b"], - "properties": { - "b": { - "type": "integer", - "maximum": 255, - "minimum": 0 - }, - "g": { - "type": "integer", - "maximum": 255, - "minimum": 0 - }, - "r": { - "type": "integer", - "maximum": 255, - "minimum": 0 - } - } - }, - "rgba": { - "type": "object", - "required": ["r", "g", "b", "a"], - "properties": { - "a": { - "type": "integer", - "maximum": 100, - "minimum": 0 - }, - "b": { - "type": "integer", - "maximum": 255, - "minimum": 0 - }, - "g": { - "type": "integer", - "maximum": 255, - "minimum": 0 - }, - "r": { - "type": "integer", - "maximum": 255, - "minimum": 0 - } - } - } - }, - "$type": "com.atproto.lexicon.schema", - "lexicon": 1 -} diff --git a/spaces-lexicon-templates/invite/create.json b/spaces-lexicon-templates/ext/invite/create.json similarity index 97% rename from spaces-lexicon-templates/invite/create.json rename to spaces-lexicon-templates/ext/invite/create.json index e3cdf75..4450aed 100644 --- a/spaces-lexicon-templates/invite/create.json +++ b/spaces-lexicon-templates/ext/invite/create.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "tools.atmo.space.invite.create", + "id": "tools.atmo.spaceExt.invite.create", "defs": { "main": { "type": "procedure", diff --git a/spaces-lexicon-templates/invite/list.json b/spaces-lexicon-templates/ext/invite/list.json similarity index 95% rename from spaces-lexicon-templates/invite/list.json rename to spaces-lexicon-templates/ext/invite/list.json index 217e822..ed316f9 100644 --- a/spaces-lexicon-templates/invite/list.json +++ b/spaces-lexicon-templates/ext/invite/list.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "tools.atmo.space.invite.list", + "id": "tools.atmo.spaceExt.invite.list", "defs": { "main": { "type": "query", diff --git a/spaces-lexicon-templates/invite/redeem.json b/spaces-lexicon-templates/ext/invite/redeem.json similarity index 94% rename from spaces-lexicon-templates/invite/redeem.json rename to spaces-lexicon-templates/ext/invite/redeem.json index 398127d..d0372c2 100644 --- a/spaces-lexicon-templates/invite/redeem.json +++ b/spaces-lexicon-templates/ext/invite/redeem.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "tools.atmo.space.invite.redeem", + "id": "tools.atmo.spaceExt.invite.redeem", "defs": { "main": { "type": "procedure", diff --git a/spaces-lexicon-templates/invite/revoke.json b/spaces-lexicon-templates/ext/invite/revoke.json similarity index 94% rename from spaces-lexicon-templates/invite/revoke.json rename to spaces-lexicon-templates/ext/invite/revoke.json index e9d068b..13ecf62 100644 --- a/spaces-lexicon-templates/invite/revoke.json +++ b/spaces-lexicon-templates/ext/invite/revoke.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "tools.atmo.space.invite.revoke", + "id": "tools.atmo.spaceExt.invite.revoke", "defs": { "main": { "type": "procedure", diff --git a/spaces-lexicon-templates/whoami.json b/spaces-lexicon-templates/ext/whoami.json similarity index 95% rename from spaces-lexicon-templates/whoami.json rename to spaces-lexicon-templates/ext/whoami.json index c0e8f9d..5736c86 100644 --- a/spaces-lexicon-templates/whoami.json +++ b/spaces-lexicon-templates/ext/whoami.json @@ -1,6 +1,6 @@ { "lexicon": 1, - "id": "tools.atmo.space.whoami", + "id": "tools.atmo.spaceExt.whoami", "defs": { "main": { "type": "query", diff --git a/src/contrail.ts b/src/contrail.ts index 268975e..38f04c1 100644 --- a/src/contrail.ts +++ b/src/contrail.ts @@ -11,6 +11,10 @@ import { processNotifyUris } from "./core/router/notify"; import type { NotifyResult } from "./core/router/notify"; import { runPersistent as runPersistentIngestion } from "./core/persistent"; import type { PersistentIngestOptions } from "./core/persistent"; +import type { PubSub } from "./core/realtime/types"; +import { InMemoryPubSub } from "./core/realtime/in-memory"; +import { createApp, type CreateAppOptions } from "./core/router"; +import type { Hono } from "hono"; export interface ContrailOptions extends ContrailConfig { db?: Database; @@ -23,6 +27,7 @@ export class Contrail { private _db?: Database; private _spacesDb?: Database; private _ingestState: IngestState = createIngestState(); + private _pubsub: PubSub | null = null; constructor(options: ContrailOptions) { const { db, spacesDb, ...configInput } = options; @@ -30,6 +35,18 @@ export class Contrail { validateConfig(this.config); this._db = db; this._spacesDb = spacesDb; + // Build the pubsub instance up-front so ingestion and HTTP routes share + // it. Caller overrides via `config.realtime.pubsub` (e.g. DurableObject). + if (this.config.realtime) { + this._pubsub = + this.config.realtime.pubsub ?? + new InMemoryPubSub({ queueBound: this.config.realtime.queueBound }); + } + } + + /** The shared realtime pubsub, or null when realtime isn't configured. */ + get pubsub(): PubSub | null { + return this._pubsub; } private getDb(db?: Database): Database { @@ -62,12 +79,22 @@ export class Contrail { /** Run one Jetstream ingestion cycle (catches up to present, then stops). */ async ingest(options?: { timeoutMs?: number }, db?: Database): Promise { - await runIngestCycle(this.getDb(db), this.config, options?.timeoutMs, this._ingestState); + await runIngestCycle( + this.getDb(db), + this.config, + options?.timeoutMs, + this._ingestState, + this._pubsub ?? undefined + ); } /** Run persistent Jetstream ingestion (long-lived, stays connected). */ async runPersistent(options?: Omit, db?: Database): Promise { - await runPersistentIngestion(this.getDb(db), this.config, { ...options, logger: this.config.logger }); + await runPersistentIngestion(this.getDb(db), this.config, { + ...options, + logger: this.config.logger, + pubsub: this._pubsub ?? undefined, + }); } /** Discover users from relays. Returns discovered DIDs. */ @@ -109,4 +136,33 @@ export class Contrail { const uriList = Array.isArray(uris) ? uris : [uris]; return processNotifyUris(this.getDb(db), this.config, uriList); } + + /** Build the Hono app for this Contrail instance. All HTTP routes + * (collection / spaces / community / realtime) are registered here. + * The realtime pubsub on this instance is reused, so subscribers see + * events published from `ingest()` / `runPersistent()` on the same instance. */ + app(options: AppOptions = {}): Hono { + const { db, ...appOpts } = options; + const main = this.getDb(db); + const spaces = options.spacesDb ?? this._spacesDb; + return createApp(main, this.config, { + ...appOpts, + spacesDb: spaces, + realtime: { ...appOpts.realtime, pubsub: this._pubsub ?? undefined }, + }); + } + + /** Fetch-style handler built from `app()`. Use this from SvelteKit / Next / + * Workers / Bun — anything that takes `(request) => Response`. */ + handler(options: AppOptions = {}): (request: Request) => Promise { + const app = this.app(options); + return (request: Request) => app.fetch(request) as Promise; + } +} + +/** Overrides accepted by `Contrail.app()` and `Contrail.handler()`. Mirrors + * `CreateAppOptions` but lets the caller also override the DBs (falling back + * to the ones given to the Contrail constructor). */ +export interface AppOptions extends CreateAppOptions { + db?: Database; } diff --git a/src/core/community/router.ts b/src/core/community/router.ts index 0318d7e..b3dfcd2 100644 --- a/src/core/community/router.ts +++ b/src/core/community/router.ts @@ -25,10 +25,7 @@ import { RESERVED_KEYS, } from "./types"; import type { ServiceJwtVerifier } from "@atcute/xrpc-server/auth"; -import { - generateInviteToken, - hashInviteToken, -} from "../spaces/invite-token"; +import { hashInviteToken, mintInviteToken } from "../invite/token"; export interface CommunityRoutesOptions { /** Override auth middleware for tests. */ @@ -257,7 +254,7 @@ export function registerCommunityRoutes( }); }); - app.get(`/xrpc/${NS}.whoami`, auth, async (c) => { + app.get(`/xrpc/${NS}.space.whoami`, auth, async (c) => { const sa = getAuth(c); const spaceUri = c.req.query("spaceUri"); if (!spaceUri) { @@ -1018,8 +1015,7 @@ export function registerCommunityRoutes( return c.json({ error: "Forbidden", reason: "cannot-grant-higher-than-self" }, 403); } - const token = generateInviteToken(); - const tokenHash = await hashInviteToken(token); + const { token, tokenHash } = await mintInviteToken(); const row = await community.createInvite({ spaceUri: body.spaceUri, diff --git a/src/core/db/records.ts b/src/core/db/records.ts index c15f53d..b7ea175 100644 --- a/src/core/db/records.ts +++ b/src/core/db/records.ts @@ -391,6 +391,11 @@ export async function applyEvents( skipFeedFanout?: boolean; /** Pre-fetched existing records — skips the internal lookup when provided */ existing?: Map; + /** When provided, publish `collection:` and `actor:` realtime + * events for each applied event. Space-scoped publishing happens elsewhere + * (see `realtime/publishing-adapter.ts`); public topics carry public + * records only, which is exactly the scope of this function. */ + pubsub?: import("../realtime/types").PubSub; } ): Promise { if (events.length === 0) return; @@ -479,6 +484,47 @@ export async function applyEvents( } await db.batch(batch); + + // Publish realtime events for public records (collection: and actor:). + // Space records publish via the wrapping adapter; this path is public-only. + if (options?.pubsub) { + const pubsub = options.pubsub; + const ts = Date.now(); + for (const e of events) { + if (e.operation === "delete") { + const payload = { + uri: e.uri, + did: e.did, + collection: e.collection, + rkey: e.rkey, + }; + await pubsub.publish({ topic: `collection:${e.collection}`, kind: "record.deleted", payload, ts }); + await pubsub.publish({ topic: `actor:${e.did}`, kind: "record.deleted", payload, ts }); + } else { + const record = e.record ? safeParseJson(e.record) : {}; + const payload = { + uri: e.uri, + did: e.did, + collection: e.collection, + rkey: e.rkey, + cid: e.cid, + record, + time_us: e.time_us, + }; + await pubsub.publish({ topic: `collection:${e.collection}`, kind: "record.created", payload, ts }); + await pubsub.publish({ topic: `actor:${e.did}`, kind: "record.created", payload, ts }); + } + } + } +} + +function safeParseJson(s: string): Record { + try { + const v = JSON.parse(s); + return v && typeof v === "object" && !Array.isArray(v) ? (v as Record) : {}; + } catch { + return {}; + } } // --- Count columns --- diff --git a/src/core/invite/index.ts b/src/core/invite/index.ts new file mode 100644 index 0000000..d79fc3b --- /dev/null +++ b/src/core/invite/index.ts @@ -0,0 +1 @@ +export { generateInviteToken, hashInviteToken, mintInviteToken } from "./token"; diff --git a/src/core/spaces/invite-token.ts b/src/core/invite/token.ts similarity index 78% rename from src/core/spaces/invite-token.ts rename to src/core/invite/token.ts index 1fc1779..76f4880 100644 --- a/src/core/spaces/invite-token.ts +++ b/src/core/invite/token.ts @@ -33,3 +33,11 @@ export async function hashInviteToken(token: string): Promise { const digest = await crypto.subtle.digest("SHA-256", encoded); return bytesToHex(new Uint8Array(digest)); } + +/** Convenience: generate a token and return both the raw form (returned to + * the creator once) and its hash (persisted as the stable ID). */ +export async function mintInviteToken(): Promise<{ token: string; tokenHash: string }> { + const token = generateInviteToken(); + const tokenHash = await hashInviteToken(token); + return { token, tokenHash }; +} diff --git a/src/core/jetstream.ts b/src/core/jetstream.ts index fa74282..d42a5e8 100644 --- a/src/core/jetstream.ts +++ b/src/core/jetstream.ts @@ -180,7 +180,8 @@ export async function runIngestCycle( db: Database, config: ContrailConfig, timeoutMs: number = 25_000, - state?: IngestState + state?: IngestState, + pubsub?: import("./realtime/types").PubSub ): Promise { const log = getLogger(config); const s = state ?? createIngestState(); @@ -241,7 +242,7 @@ export async function runIngestCycle( for (let i = 0; i < events.length; i += BATCH_SIZE) { const batch = events.slice(i, i + BATCH_SIZE); - await applyEvents(db, batch, config); + await applyEvents(db, batch, config, { pubsub }); } // Refresh stale/missing identities for DIDs in this batch diff --git a/src/core/persistent.ts b/src/core/persistent.ts index 0574656..5eb48a2 100644 --- a/src/core/persistent.ts +++ b/src/core/persistent.ts @@ -15,6 +15,9 @@ export interface PersistentIngestOptions { /** Override subscription creation for testing */ createSubscription?: (cursor: number | null) => JetstreamSubscription; logger?: Logger; + /** Publish `collection:` / `actor:` events for each applied + * public record. Usually supplied by the Contrail instance. */ + pubsub?: import("./realtime/types").PubSub; } function getLogger(config: ContrailConfig, options?: PersistentIngestOptions): Logger { @@ -68,6 +71,7 @@ export async function runPersistent( state, log, createSubscription: options?.createSubscription, + pubsub: options?.pubsub, }); reconnectAttempts = 0; } catch (err) { @@ -93,6 +97,7 @@ interface StreamOptions { state: IngestState; log: Logger; createSubscription?: (cursor: number | null) => any; + pubsub?: import("./realtime/types").PubSub; } async function streamAndFlush( @@ -132,7 +137,7 @@ async function streamAndFlush( resetFlushTimer(); try { - await applyEvents(db, batch, config); + await applyEvents(db, batch, config, { pubsub: opts.pubsub }); const lastTimeUs = Math.max(...batch.map((e) => e.time_us)); await saveCursor(db, lastTimeUs); diff --git a/src/core/realtime/publishing-adapter.ts b/src/core/realtime/publishing-adapter.ts index e96a226..07cbc00 100644 --- a/src/core/realtime/publishing-adapter.ts +++ b/src/core/realtime/publishing-adapter.ts @@ -68,7 +68,7 @@ export function wrapWithPublishing( await publishSpaceAndCommunity(spaceUri, owner, (topic) => ({ topic, kind: "member.added", - payload: { spaceUri, did }, + payload: { space: spaceUri, did }, ts: now, })); }, @@ -80,7 +80,7 @@ export function wrapWithPublishing( await publishSpaceAndCommunity(spaceUri, owner, (topic) => ({ topic, kind: "member.removed", - payload: { spaceUri, did }, + payload: { space: spaceUri, did }, ts: now, })); }, @@ -94,7 +94,7 @@ export function wrapWithPublishing( await publishSpaceAndCommunity(spaceUri, owner, (topic) => ({ topic, kind: "member.added", - payload: { spaceUri, did }, + payload: { space: spaceUri, did }, ts: now, })); } @@ -102,7 +102,7 @@ export function wrapWithPublishing( await publishSpaceAndCommunity(spaceUri, owner, (topic) => ({ topic, kind: "member.removed", - payload: { spaceUri, did }, + payload: { space: spaceUri, did }, ts: now, })); } @@ -112,17 +112,22 @@ export function wrapWithPublishing( await inner.putRecord(record); const owner = await ownerOf(record.spaceUri); const now = Date.now(); + // Space records use ms timestamps; listRecords surface uses microseconds + // (time_us). Convert here so subscribers can render a row identically. + const time_us = record.createdAt * 1000; + const uri = `at://${record.authorDid}/${record.collection}/${record.rkey}`; await publishSpaceAndCommunity(record.spaceUri, owner, (topic) => ({ topic, kind: "record.created", payload: { - spaceUri: record.spaceUri, + uri, + did: record.authorDid, collection: record.collection, - authorDid: record.authorDid, rkey: record.rkey, cid: record.cid, record: record.record, - createdAt: record.createdAt, + time_us, + space: record.spaceUri, }, ts: now, })); @@ -132,10 +137,11 @@ export function wrapWithPublishing( await inner.deleteRecord(spaceUri, collection, authorDid, rkey); const owner = await ownerOf(spaceUri); const now = Date.now(); + const uri = `at://${authorDid}/${collection}/${rkey}`; await publishSpaceAndCommunity(spaceUri, owner, (topic) => ({ topic, kind: "record.deleted", - payload: { spaceUri, collection, authorDid, rkey }, + payload: { uri, did: authorDid, collection, rkey, space: spaceUri }, ts: now, })); }, diff --git a/src/core/realtime/query-filter.ts b/src/core/realtime/query-filter.ts index 6806c45..f332097 100644 --- a/src/core/realtime/query-filter.ts +++ b/src/core/realtime/query-filter.ts @@ -70,9 +70,9 @@ export function translateForQuery( const spec = sub.querySpec; if (!spec) return null; if (event.kind !== "record.created" && event.kind !== "record.deleted") return []; - if (event.payload.spaceUri !== spec.spaceUri) return []; + if (event.payload.space !== spec.spaceUri) return []; - const primaryUri = `at://${event.payload.authorDid}/${event.payload.collection}/${event.payload.rkey}`; + const primaryUri = `at://${event.payload.did}/${event.payload.collection}/${event.payload.rkey}`; if (event.payload.collection === spec.collection) { if (event.kind === "record.created") { @@ -82,7 +82,7 @@ export function translateForQuery( data: { record: { uri: primaryUri, - did: event.payload.authorDid, + did: event.payload.did, rkey: event.payload.rkey, collection: event.payload.collection, cid: event.payload.cid, @@ -100,7 +100,7 @@ export function translateForQuery( kind: "record.deleted", data: { uri: primaryUri, - did: event.payload.authorDid, + did: event.payload.did, rkey: event.payload.rkey } } @@ -125,7 +125,7 @@ export function translateForQuery( relation: relName, child: { uri: primaryUri, - did: event.payload.authorDid, + did: event.payload.did, rkey: event.payload.rkey, collection: event.payload.collection, cid: event.payload.cid, @@ -145,7 +145,7 @@ export function translateForQuery( parentUri: info.parentUri, relation: relName, childRkey: event.payload.rkey, - childDid: event.payload.authorDid + childDid: event.payload.did } } ]; diff --git a/src/core/realtime/resolve.ts b/src/core/realtime/resolve.ts index 1499faa..f8a0ab4 100644 --- a/src/core/realtime/resolve.ts +++ b/src/core/realtime/resolve.ts @@ -17,7 +17,10 @@ import { resolveReachableSpaces } from "../community/acl"; import { spaceTopic, parseCommunityTopic, parseSpaceTopic } from "./types"; export interface TopicResolutionContext { - spaces: StorageAdapter; + /** May be null when the deployment has no spaces module — in that case + * `space:` and `community:` topics are NotSupported. Public topics + * (`collection:`, `actor:`) still resolve. */ + spaces: StorageAdapter | null; /** May be null if the community module is not enabled. */ community: CommunityAdapter | null; } @@ -41,6 +44,9 @@ export async function resolveTopicForCaller( // space: const spaceUri = parseSpaceTopic(rawTopic); if (spaceUri) { + if (!ctx.spaces) { + return { ok: false, error: "NotSupported", reason: "spaces-module-disabled" }; + } const space = await ctx.spaces.getSpace(spaceUri); if (!space) return { ok: false, error: "NotFound", reason: "space-not-found" }; if (space.ownerDid === callerDid) return { ok: true, topics: [rawTopic] }; @@ -52,7 +58,7 @@ export async function resolveTopicForCaller( // community: const communityDid = parseCommunityTopic(rawTopic); if (communityDid) { - if (!ctx.community) { + if (!ctx.community || !ctx.spaces) { return { ok: false, error: "NotSupported", reason: "community-module-disabled" }; } const row = await ctx.community.getCommunity(communityDid); @@ -72,18 +78,15 @@ export async function resolveTopicForCaller( return { ok: true, topics }; } - // actor: — self-only in v1. + // actor: — public stream of records authored by this DID. + // Any caller can subscribe (parallels listRecords with an `actor` filter). if (rawTopic.startsWith("actor:")) { - const did = rawTopic.slice("actor:".length); - if (did !== callerDid) { - return { ok: false, error: "Forbidden", reason: "actor-self-only" }; - } return { ok: true, topics: [rawTopic] }; } - // collection: — public firehose, not exposed in v1. + // collection: — public firehose for this collection. if (rawTopic.startsWith("collection:")) { - return { ok: false, error: "NotSupported", reason: "collection-firehose-disabled" }; + return { ok: true, topics: [rawTopic] }; } return { ok: false, error: "InvalidRequest", reason: "unknown-topic" }; diff --git a/src/core/realtime/router.ts b/src/core/realtime/router.ts index a6448ef..2726226 100644 --- a/src/core/realtime/router.ts +++ b/src/core/realtime/router.ts @@ -15,14 +15,21 @@ import type { PubSub, RealtimeEvent } from "./types"; import { DEFAULT_TICKET_TTL_MS, DEFAULT_KEEPALIVE_MS } from "./types"; export interface RealtimeRoutesOptions { - /** Required: auth middleware for `.realtime.ticket` (JWT required). - * The subscribe endpoint uses the *ticket* path primarily, so it does not - * require this middleware — but bots can subscribe with Authorization - * header, in which case this middleware is used as a fallback. */ - authMiddleware: MiddlewareHandler; + /** Auth middleware for `.realtime.ticket` and for JWT-based bot + * subscriptions to private topics. Null when no JWT verifier is available + * (deployments without a spaces config) — in that case, private-topic + * subscribe paths return NotSupported and public topics still work without + * auth. */ + authMiddleware: MiddlewareHandler | null; pubsub?: PubSub; } +/** Public topics: subscribable without any auth. Mirrors listRecords + * semantics — no JWT means "public records only". */ +function isPublicTopic(topic: string): boolean { + return topic.startsWith("collection:") || topic.startsWith("actor:"); +} + /** WebSocketPair exists on Cloudflare Workers; on Node/Bun it's absent. * When absent, a platform-provided WebSocket accept hook is used instead. */ interface WebSocketPairCtor { @@ -32,7 +39,7 @@ interface WebSocketPairCtor { export function registerRealtimeRoutes( app: Hono, config: ContrailConfig, - spaces: StorageAdapter, + spaces: StorageAdapter | null, community: CommunityAdapter | null, options: RealtimeRoutesOptions ): void { @@ -47,36 +54,51 @@ export function registerRealtimeRoutes( const NS = `${config.namespace}.realtime`; // POST /.realtime.ticket — { topic } → { ticket, topics, expiresAt } - app.post(`/xrpc/${NS}.ticket`, options.authMiddleware, async (c) => { - const sa = getAuth(c); - const body = (await c.req.json().catch(() => null)) as { topic?: string } | null; - if (!body?.topic) { - return c.json({ error: "InvalidRequest", message: "topic required" }, 400); - } - const resolved = await resolveTopicForCaller(body.topic, sa.issuer, { spaces, community }); - if (!resolved.ok) { - const status = resolved.error === "NotFound" ? 404 : resolved.error === "Forbidden" ? 403 : 400; - return c.json({ error: resolved.error, reason: resolved.reason }, status); - } - const ticket = await signer.sign({ - topics: resolved.topics, - did: sa.issuer, - ttlMs: ticketTtl, - }); - return c.json({ - ticket, - topics: resolved.topics, - expiresAt: Date.now() + ticketTtl, + // Ticket-minting exists so browsers (which can't set Authorization on + // EventSource) can subscribe to *private* topics. Public topics + // (collection:, actor:) don't need tickets — subscribe with `?topic=` directly. + if (options.authMiddleware) { + const authMw = options.authMiddleware; + app.post(`/xrpc/${NS}.ticket`, authMw, async (c) => { + const sa = getAuth(c); + const body = (await c.req.json().catch(() => null)) as { topic?: string } | null; + if (!body?.topic) { + return c.json({ error: "InvalidRequest", message: "topic required" }, 400); + } + const resolved = await resolveTopicForCaller(body.topic, sa.issuer, { spaces, community }); + if (!resolved.ok) { + const status = resolved.error === "NotFound" ? 404 : resolved.error === "Forbidden" ? 403 : 400; + return c.json({ error: resolved.error, reason: resolved.reason }, status); + } + const ticket = await signer.sign({ + topics: resolved.topics, + did: sa.issuer, + ttlMs: ticketTtl, + }); + return c.json({ + ticket, + topics: resolved.topics, + expiresAt: Date.now() + ticketTtl, + }); }); - }); + } - // GET /.realtime.subscribe — SSE or WS, driven by ?ticket= or JWT. + // GET /.realtime.subscribe — SSE or WS. + // + // Three access paths, all land on the same stream: + // - `?topic=collection:` or `?topic=actor:` — *public*, no auth. + // Mirrors listRecords semantics. + // - `?ticket=` — presented by browsers, minted via `.ticket` after + // a JWT-authenticated call. Only used for private topics. + // - `Authorization: Bearer ` + `?topic=space:` — server-side + // bots can skip the ticket dance and go straight to subscribe. app.get(`/xrpc/${NS}.subscribe`, async (c) => { const url = new URL(c.req.url); const ticketParam = url.searchParams.get("ticket"); const collectionFilter = url.searchParams.get("collection"); + const topicParam = url.searchParams.get("topic"); - let callerDid: string; + let callerDid: string | null = null; let topics: string[]; if (ticketParam) { @@ -87,30 +109,41 @@ export function registerRealtimeRoutes( callerDid = payload.did; topics = payload.topics; // Optional: narrow to topics the query explicitly requests. - const requested = url.searchParams.get("topic"); - if (requested) { - if (!payload.topics.includes(requested)) { + if (topicParam) { + if (!payload.topics.includes(topicParam)) { return c.json({ error: "Forbidden", reason: "topic-not-in-ticket" }, 403); } - topics = [requested]; + topics = [topicParam]; } + } else if (topicParam && isPublicTopic(topicParam)) { + // Public subscribe — no auth required. Jetstream ingestion publishes + // record events to collection:/actor: topics directly. + topics = [topicParam]; } else { - // JWT path for server-side bots. Requires the auth middleware to have - // run and attached `serviceAuth` to the context — so we invoke it - // inline. - const runAuth = options.authMiddleware; + // JWT path for private-topic bots. If no auth middleware is available + // (deployment has no spaces config), private topics aren't offered. + if (!options.authMiddleware) { + return c.json( + { + error: "InvalidRequest", + reason: "private-topic-without-auth", + message: + "Subscribing to space:/community: topics requires a JWT verifier; only public topics (collection:, actor:) are available on this deployment.", + }, + 400 + ); + } let authed = false; - await runAuth(c, async () => { + await options.authMiddleware(c, async () => { authed = true; }); if (!authed) return c.res; // middleware already responded with 401 const sa = getAuth(c); callerDid = sa.issuer; - const topic = url.searchParams.get("topic"); - if (!topic) { + if (!topicParam) { return c.json({ error: "InvalidRequest", message: "topic required" }, 400); } - const resolved = await resolveTopicForCaller(topic, callerDid, { spaces, community }); + const resolved = await resolveTopicForCaller(topicParam, callerDid, { spaces, community }); if (!resolved.ok) { const status = resolved.error === "NotFound" ? 404 : resolved.error === "Forbidden" ? 403 : 400; return c.json({ error: resolved.error, reason: resolved.reason }, status); @@ -191,10 +224,11 @@ function anySignal(signals: AbortSignal[]): AbortSignal { /** Wrap an iterable: drop events that don't pass the collection filter (if * any), and close the outer controller as soon as we see a `member.removed` - * for the caller's own DID. */ + * for the caller's own DID. `callerDid` may be null on public subscriptions + * (anonymous) — in that case self-kick is not applicable. */ function withSelfKickAndFilter( source: AsyncIterable, - callerDid: string, + callerDid: string | null, collectionFilter: string | null, ac: AbortController ): AsyncIterable { diff --git a/src/core/realtime/types.ts b/src/core/realtime/types.ts index 4f72dfd..05a246e 100644 --- a/src/core/realtime/types.ts +++ b/src/core/realtime/types.ts @@ -3,40 +3,59 @@ /** Discriminated union of every event kind that flows through the PubSub. * * `record.created` carries the full record body so a subscriber can apply an - * insert or upsert without a follow-up `getRecord` call. `putRecord` over an - * existing `(authorDid, rkey)` publishes another `record.created` — treat it - * as upsert. */ + * insert or upsert without a follow-up `getRecord` call. Writing a new record + * to the same `(did, collection, rkey)` publishes another `record.created` — + * treat it as upsert. + * + * **Payload shape mirrors `listRecords` output** (`uri`, `did`, `space?`, + * `time_us`), so a subscriber can render a live row the same way it renders + * a fetched row. + * + * **Publisher/topic matrix (intentional trust split):** + * - `collection:` and `actor:` carry *public* record events only + * (from jetstream ingestion) — no `space`. + * - `space:` and `community:` carry *space* events — `space` is + * always set. Never cross-published to public topics (privacy). */ export type RealtimeEvent = | { topic: string; kind: "record.created"; payload: { - spaceUri: string; + uri: string; + did: string; collection: string; - authorDid: string; rkey: string; cid: string | null; record: Record; - createdAt: number; + time_us: number; + /** Present only for space records; absent for public records. */ + space?: string; }; ts: number; } | { topic: string; kind: "record.deleted"; - payload: { spaceUri: string; collection: string; authorDid: string; rkey: string }; + payload: { + uri: string; + did: string; + collection: string; + rkey: string; + /** Present only for space records; absent for public records. */ + space?: string; + }; ts: number; } | { topic: string; kind: "member.added"; - payload: { spaceUri: string; did: string }; + payload: { space: string; did: string }; ts: number; } | { topic: string; kind: "member.removed"; - payload: { spaceUri: string; did: string }; + payload: { space: string; did: string }; ts: number; }; diff --git a/src/core/router/collection.ts b/src/core/router/collection.ts index 5c637c8..d6b4968 100644 --- a/src/core/router/collection.ts +++ b/src/core/router/collection.ts @@ -18,7 +18,7 @@ import type { FormattedRecord } from "./helpers"; import { formatRecord, parseIntParam, fieldToParam } from "./helpers"; import { verifyServiceAuthRequest, extractInviteToken, checkInviteReadGrant } from "../spaces/auth"; import { checkAccess } from "../spaces/acl"; -import { hashInviteToken } from "../spaces/invite-token"; +import { hashInviteToken } from "../invite/token"; import type { SpacesContext } from "."; import type { Nsid } from "@atcute/lexicons"; import type { RealtimeEvent } from "../realtime/types"; @@ -27,8 +27,7 @@ import { spaceTopic } from "../realtime/types"; import type { SubscriberQuerySpec } from "../realtime/durable-object"; import { DurableObjectPubSub } from "../realtime/durable-object"; import { TicketSigner, type TicketQuerySpec } from "../realtime/ticket"; -import { parseHydrateParams } from "./hydrate"; -import { getRelationField, getNestedValue, nsidForShortName } from "../types"; +import { getRelationField, getNestedValue } from "../types"; /** Shared implementation of the watchRecords snapshot+live loop. Called by * both transport branches (SSE and Worker-terminated WS). The caller owns @@ -73,8 +72,7 @@ async function runQueryStream(opts: { const parentDids = new Set(); const childToParent = new Map(); - const primaryUri = (payload: { authorDid: string; collection: string; rkey: string }) => - `at://${payload.authorDid}/${payload.collection}/${payload.rkey}`; + const primaryUri = (payload: { uri: string }) => payload.uri; const handleChildEvent = (event: RealtimeEvent) => { if (!trackHydration) return; @@ -82,7 +80,7 @@ async function runQueryStream(opts: { const meta = childCollectionMap.get(event.payload.collection); if (!meta) return; if (!(hydrateSpec.relations as Record)[meta.relName]) return; - if (event.payload.spaceUri !== spaceUri) return; + if (event.payload.space !== spaceUri) return; if (event.kind === "record.created") { const matched = getNestedValue(event.payload.record, meta.matchField); @@ -105,7 +103,7 @@ async function runQueryStream(opts: { relation: meta.relName, child: { uri: primaryUri(event.payload), - did: event.payload.authorDid, + did: event.payload.did, rkey: event.payload.rkey, collection: event.payload.collection, cid: event.payload.cid, @@ -121,7 +119,7 @@ async function runQueryStream(opts: { parentUri: info.parentUri, relation: info.relName, childRkey: event.payload.rkey, - childDid: event.payload.authorDid + childDid: event.payload.did }); } }; @@ -134,7 +132,7 @@ async function runQueryStream(opts: { return; } if (event.kind !== "record.created" && event.kind !== "record.deleted") return; - if (event.payload.spaceUri !== spaceUri) return; + if (event.payload.space !== spaceUri) return; if (event.payload.collection !== colNsid) { handleChildEvent(event); @@ -145,11 +143,11 @@ async function runQueryStream(opts: { const uri = primaryUri(event.payload); if (event.kind === "record.created") { parentUris.add(uri); - parentDids.add(event.payload.authorDid); + parentDids.add(event.payload.did); send("record.created", { record: { uri, - did: event.payload.authorDid, + did: event.payload.did, rkey: event.payload.rkey, collection: event.payload.collection, cid: event.payload.cid, @@ -163,7 +161,7 @@ async function runQueryStream(opts: { parentUris.delete(uri); send("record.deleted", { uri, - did: event.payload.authorDid, + did: event.payload.did, rkey: event.payload.rkey }); } diff --git a/src/core/router/index.ts b/src/core/router/index.ts index 35ef535..74f0499 100644 --- a/src/core/router/index.ts +++ b/src/core/router/index.ts @@ -89,22 +89,27 @@ export function createApp( } : null; - // Realtime pubsub is built up-front so the publishing decorator can reference - // it. The subscribe endpoint uses the same instance. + // Realtime pubsub is built whenever realtime is configured — independent of + // spaces. With spaces, the spaces adapter is wrapped so private record/member + // events publish to space:/community: topics. Without spaces, only public + // topics (collection:/actor:) see traffic — those are published from + // applyEvents (jetstream ingestion), not from here. let realtimePubsub: PubSub | null = null; - if (config.realtime && spacesCtx) { + if (config.realtime) { realtimePubsub = options.realtime?.pubsub ?? config.realtime.pubsub ?? new InMemoryPubSub({ queueBound: config.realtime.queueBound, }); - const communityAdapter = config.community ? new CommunityAdapter(spacesDb) : null; - const isCommunityDid = communityAdapter - ? cachedIsCommunityDid(communityAdapter) - : undefined; - spacesCtx = { - ...spacesCtx, - adapter: wrapWithPublishing(spacesCtx.adapter, realtimePubsub, { isCommunityDid }), - }; + if (spacesCtx) { + const communityAdapter = config.community ? new CommunityAdapter(spacesDb) : null; + const isCommunityDid = communityAdapter + ? cachedIsCommunityDid(communityAdapter) + : undefined; + spacesCtx = { + ...spacesCtx, + adapter: wrapWithPublishing(spacesCtx.adapter, realtimePubsub, { isCommunityDid }), + }; + } } registerAdminRoutes(app, db, config); @@ -130,13 +135,18 @@ export function createApp( ); } - if (config.realtime && spacesCtx && realtimePubsub) { - const authMiddleware = - options.realtime?.authMiddleware ?? - options.spaces?.authMiddleware ?? - createServiceAuthMiddleware(spacesCtx.verifier, { authOverride }); + if (config.realtime && realtimePubsub) { + // The ticket endpoint still needs a JWT verifier — but that verifier only + // exists when spaces is configured. Without spaces, private-topic ticket + // minting simply isn't offered; public subscriptions (collection:/actor:) + // require no auth and still work. + const authMiddleware = spacesCtx + ? options.realtime?.authMiddleware ?? + options.spaces?.authMiddleware ?? + createServiceAuthMiddleware(spacesCtx.verifier, { authOverride }) + : null; const communityAdapter = config.community ? new CommunityAdapter(spacesDb) : null; - registerRealtimeRoutes(app, config, spacesCtx.adapter, communityAdapter, { + registerRealtimeRoutes(app, config, spacesCtx?.adapter ?? null, communityAdapter, { authMiddleware, pubsub: realtimePubsub, }); diff --git a/src/core/spaces/router.ts b/src/core/spaces/router.ts index 6027d06..4fef67a 100644 --- a/src/core/spaces/router.ts +++ b/src/core/spaces/router.ts @@ -11,7 +11,7 @@ import { verifyServiceAuthRequest, } from "./auth"; import { nextTid } from "./tid"; -import { generateInviteToken, hashInviteToken } from "./invite-token"; +import { hashInviteToken, mintInviteToken } from "../invite/token"; import { buildSpaceUri } from "./uri"; import { DEFAULT_BLOB_MAX_SIZE, @@ -87,9 +87,11 @@ export function registerSpacesRoutes( /** Space endpoints are emitted per-deployment under the configured namespace; * the deployment owns and publishes its own lexicons. The library ships - * templates at `lexicons/tools/atmo/space/*` that the generator instantiates - * under `.space.*`. */ + * templates at `spaces-lexicon-templates/*` that the generator instantiates + * under `.space.*` (spec-aligned) and `.spaceExt.*` (contrail + * extras — invites, whoami — that the permissioned-data spec doesn't cover). */ const SPACE = `${config.namespace}.space`; + const SPACE_EXT = `${config.namespace}.spaceExt`; // Read endpoints app.get(`/xrpc/${SPACE}.listSpaces`, auth, async (c) => { @@ -484,8 +486,8 @@ export function registerSpacesRoutes( return c.json({ space: publicSpaceView(space, true) }); }); - // Invites - app.post(`/xrpc/${SPACE}.invite.create`, auth, async (c) => { + // Invites (contrail extras — emitted under .spaceExt.invite.*) + app.post(`/xrpc/${SPACE_EXT}.invite.create`, auth, async (c) => { const sa = getAuth(c); const body = (await c.req.json().catch(() => null)) as | { @@ -509,8 +511,7 @@ export function registerSpacesRoutes( return c.json({ error: "Forbidden", reason: "not-owner" }, 403); } - const token = generateInviteToken(); - const tokenHash = await hashInviteToken(token); + const { token, tokenHash } = await mintInviteToken(); const invite = await adapter.createInvite({ spaceUri: body.spaceUri, tokenHash, @@ -523,7 +524,7 @@ export function registerSpacesRoutes( return c.json({ token, invite: publicInviteView(invite) }); }); - app.post(`/xrpc/${SPACE}.invite.redeem`, auth, async (c) => { + app.post(`/xrpc/${SPACE_EXT}.invite.redeem`, auth, async (c) => { const sa = getAuth(c); const body = (await c.req.json().catch(() => null)) as { token?: string } | null; if (!body?.token) { @@ -538,7 +539,7 @@ export function registerSpacesRoutes( return c.json({ spaceUri: invite.spaceUri }); }); - app.get(`/xrpc/${SPACE}.invite.list`, auth, async (c) => { + app.get(`/xrpc/${SPACE_EXT}.invite.list`, auth, async (c) => { const sa = getAuth(c); const spaceUri = c.req.query("spaceUri"); if (!spaceUri) return c.json({ error: "InvalidRequest", message: "spaceUri required" }, 400); @@ -552,7 +553,7 @@ export function registerSpacesRoutes( return c.json({ invites: invites.map(publicInviteView) }); }); - app.post(`/xrpc/${SPACE}.invite.revoke`, auth, async (c) => { + app.post(`/xrpc/${SPACE_EXT}.invite.revoke`, auth, async (c) => { const sa = getAuth(c); const body = (await c.req.json().catch(() => null)) as | { spaceUri?: string; tokenHash?: string } @@ -624,7 +625,7 @@ export function registerSpacesRoutes( return c.json({ ok: true }); }); - app.get(`/xrpc/${SPACE}.whoami`, auth, async (c) => { + app.get(`/xrpc/${SPACE_EXT}.whoami`, auth, async (c) => { const sa = getAuth(c); const spaceUri = c.req.query("spaceUri"); if (!spaceUri) return c.json({ error: "InvalidRequest", message: "spaceUri required" }, 400); diff --git a/src/core/types.ts b/src/core/types.ts index 688b89c..e45f31e 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -384,6 +384,12 @@ export function validateConfig(config: ContrailConfig): void { } } } + + if (config.community && !config.spaces) { + throw new Error( + "Invalid config: `community` requires `spaces`. Community-owned spaces reuse the spaces storage adapter." + ); + } } // Helpers diff --git a/src/generate.ts b/src/generate.ts index c68ea87..2b43b2f 100644 --- a/src/generate.ts +++ b/src/generate.ts @@ -769,7 +769,10 @@ export function generateLexicons(options: GenerateOptions): Record.space.* --- + // --- Spaces: instantiate library templates under .space.* and + // .spaceExt.* (contrail-specific extras — invites, whoami). + // Splitting the namespace keeps the `space.*` surface aligned with the + // permissioned-data spec; extras move when the spec does not. --- if (config.spaces) { log("Generating space endpoints..."); @@ -777,17 +780,29 @@ export function generateLexicons(options: GenerateOptions): Record - id.startsWith("tools.atmo.space") ? id.replace(/^tools\.atmo\.space/, `${ns}.space`) : id; + // Order matters: more specific prefix first so `spaceExt` isn't matched + // by the `space` rule. + const prefixes: Array<[string, string]> = [ + ["tools.atmo.spaceExt", `${ns}.spaceExt`], + ["tools.atmo.space", `${ns}.space`], + ]; + const templateIdRe = /^tools\.atmo\.space(Ext)?(\.[A-Za-z0-9.]+)?$/; + const idReplace = (id: string) => { + for (const [from, to] of prefixes) { + if (id === from || id.startsWith(from + ".")) { + return to + id.slice(from.length); + } + } + return id; + }; const rewriteRefs = (obj: any): any => { if (Array.isArray(obj)) return obj.map(rewriteRefs); if (obj && typeof obj === "object") { const out: any = {}; for (const [k, v] of Object.entries(obj)) { - if (k === "ref" && typeof v === "string" && v.startsWith("tools.atmo.space")) { - out[k] = v.replace(/^tools\.atmo\.space/, `${ns}.space`); + if (k === "ref" && typeof v === "string") { + out[k] = idReplace(v); } else if (k === "id" && typeof v === "string" && templateIdRe.test(v)) { out[k] = idReplace(v); } else { diff --git a/src/index.ts b/src/index.ts index 404f3b4..dbd672f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -50,7 +50,7 @@ export type { } from "./core/spaces/types"; export { HostedAdapter } from "./core/spaces/adapter"; export { nextTid } from "./core/spaces/tid"; -export { generateInviteToken, hashInviteToken } from "./core/spaces/invite-token"; +export { generateInviteToken, hashInviteToken, mintInviteToken } from "./core/invite"; export { MemoryBlobAdapter, R2BlobAdapter, diff --git a/src/server.ts b/src/server.ts index b8b03eb..ae4f47f 100644 --- a/src/server.ts +++ b/src/server.ts @@ -1,34 +1,29 @@ import type { Contrail } from "./contrail"; import type { Database } from "./core/types"; -import { createApp } from "./core/router"; /** * Create an HTTP handler from a Contrail instance. - * Returns a standard (Request, db?, spacesDb?) => Promise function. * - * Usage: + * Thin wrapper over `contrail.handler()` that accepts per-request DB overrides + * (useful on Cloudflare Workers where the DB binding lives on the request env). + * * const handle = createHandler(contrail); * // SvelteKit: export const GET = ({ request }) => handle(request); * // Workers: return handle(request, env.DB, env.SPACES_DB); + * + * For most cases, prefer `contrail.handler()` directly when DBs are bound at + * construction time. */ export function createHandler( contrail: Contrail -): (request: Request, db?: Database, spacesDb?: Database) => Promise | Response { - // Cache the Hono app when db is bound at construction - let cachedApp: ReturnType | null = null; - +): (request: Request, db?: Database, spacesDb?: Database) => Promise { + // When no per-request DBs are provided, build the app once and reuse it. + let cached: ((request: Request) => Promise) | null = null; return (request: Request, db?: Database, spacesDb?: Database) => { - const d = db ?? (contrail as any)._db; - if (!d) throw new Error("No database provided. Pass db to Contrail constructor or to handler."); - - const sd = spacesDb ?? (contrail as any)._spacesDb; - - // If db is the same bound instance, reuse the Hono app - if (!db && !cachedApp) { - cachedApp = createApp(d, contrail.config, { spacesDb: sd }); + if (db || spacesDb) { + return contrail.handler({ db, spacesDb })(request); } - - const app = db ? createApp(d, contrail.config, { spacesDb: sd }) : cachedApp!; - return app.fetch(request); + cached ??= contrail.handler(); + return cached(request); }; } diff --git a/src/sync/index.ts b/src/sync/index.ts index 15e823c..2a28a37 100644 --- a/src/sync/index.ts +++ b/src/sync/index.ts @@ -37,6 +37,14 @@ export interface WatchStoreOptions { * When omitted, the connection is opened without an explicit token — * rely on cookies / bearer tokens your fetch policy sends automatically. */ fetchTicket?: () => Promise; + /** Optional: mint a short-lived bearer token (typically an atproto + * service-auth JWT scoped to `.watchRecords`). Called once + * per connect attempt; returned token is sent as `Authorization: Bearer` + * on the handshake fetch (`mode=ws`) or on the SSE connection via a + * `?auth=` query param the server doesn't use — browsers can't set + * headers on EventSource, so for SSE the app should cookie-auth or + * fall back to `fetchTicket`. */ + fetchAuthToken?: () => Promise; /** Custom compare. Default: sort by `time_us` descending (newest first), * tie-breaking by rkey. */ compareRecords?: (a: WatchRecord, b: WatchRecord) => number; @@ -97,6 +105,12 @@ export function createWatchStore(options: WatchStoreOptions): WatchStore { let error: Error | null = null; const listeners = new Set<(state: WatchStoreState) => void>(); + // Per-snapshot reconcile: during a snapshot we track which keys were + // included, and on snapshot.end we evict any stale keys the previous + // snapshot had but this one didn't. Keeps data visible across reconnects + // and prevents stale records accumulating forever. + let snapshotSeen: Set | null = null; + let es: EventSource | null = null; let ws: WebSocket | null = null; let started = false; @@ -127,7 +141,9 @@ export function createWatchStore(options: WatchStoreOptions): WatchStore { }; const applySnapshotRecord = (record: WatchRecord) => { - byKey.set(key(record), record); + const k = key(record); + byKey.set(k, record); + snapshotSeen?.add(k); }; const applyCreated = (record: WatchRecord) => { @@ -199,14 +215,28 @@ export function createWatchStore(options: WatchStoreOptions): WatchStore { case "snapshot.start": setStatus("snapshot"); backoffMs = 1000; + // Begin reconcile pass. Stale records from the previous session + // stay visible until snapshot.end replaces them. + snapshotSeen = new Set(); break; case "snapshot.record": applySnapshotRecord(msg.data.record); break; - case "snapshot.end": + case "snapshot.end": { + // Evict anything we had before this snapshot that the server + // didn't re-send. Preserves continuity across reconnects + // while still dropping records that were deleted while we + // were disconnected. + if (snapshotSeen) { + for (const k of Array.from(byKey.keys())) { + if (!snapshotSeen.has(k)) byKey.delete(k); + } + snapshotSeen = null; + } resort(); setStatus("live"); break; + } case "record.created": applyCreated(msg.data.record); break; @@ -282,7 +312,12 @@ export function createWatchStore(options: WatchStoreOptions): WatchStore { }; const openWs = async () => { - // Step 1: snapshot + handshake. + // Step 1: snapshot + handshake. Authenticated by Bearer token from + // `fetchAuthToken` (an atproto service-auth JWT — browsers get this + // from a same-origin helper that uses the OAuth session). The server + // returns a ticket bound to (did, spaceUri, querySpec); we pass that + // back embedded in the wsUrl on the WS upgrade so the WS itself + // doesn't need a JWT (EventSource/WebSocket can't set headers). let snapshotUrl = options.url; const sep = snapshotUrl.includes("?") ? "&" : "?"; snapshotUrl += `${sep}mode=ws`; @@ -290,7 +325,12 @@ export function createWatchStore(options: WatchStoreOptions): WatchStore { const ticket = await options.fetchTicket(); snapshotUrl += `&ticket=${encodeURIComponent(ticket)}`; } - const res = await fetch(snapshotUrl, { headers: { accept: "application/json" } }); + const headers: Record = { accept: "application/json" }; + if (options.fetchAuthToken) { + const token = await options.fetchAuthToken(); + headers.authorization = `Bearer ${token}`; + } + const res = await fetch(snapshotUrl, { headers }); if (!res.ok) throw new Error(`snapshot fetch failed (${res.status})`); const handshake = (await res.json()) as { transport: "ws"; @@ -311,8 +351,15 @@ export function createWatchStore(options: WatchStoreOptions): WatchStore { resort(); backoffMs = 1000; - // Step 2: open the WS. `wsUrl` is a relative path from the server. - const wsHref = new URL(handshake.wsUrl, options.url); + // Step 2: open the WS. `wsUrl` is a relative path from the server; + // resolve it against the page origin (not against options.url — both + // are relative and `new URL(relative, relative)` throws, which + // previously manifested as the connection indicator being stuck on + // "connecting" while the engine re-fetched snapshots in a tight + // loop). The server embeds the handshake ticket in it when issued. + const g = globalThis as { location?: { origin?: string } }; + const base = g.location?.origin ?? "http://localhost"; + const wsHref = new URL(handshake.wsUrl, base); wsHref.protocol = wsHref.protocol === "https:" ? "wss:" : "ws:"; const socket = new WebSocket(wsHref.toString()); ws = socket; @@ -365,8 +412,8 @@ export function createWatchStore(options: WatchStoreOptions): WatchStore { backoffMs = Math.min(backoffMs * 2, 30_000); setTimeout(() => { if (stopped) return; - byKey.clear(); - sorted = []; + // Keep existing records visible; `snapshot.end` will reconcile + // away anything that disappeared while we were offline. void openOnce(); }, delay); }; diff --git a/tests/community-delegation.test.ts b/tests/community-delegation.test.ts index 7466dac..bee3af4 100644 --- a/tests/community-delegation.test.ts +++ b/tests/community-delegation.test.ts @@ -125,7 +125,7 @@ async function grant(app: Hono, caller: string, spaceUri: string, subject: any, } async function whoamiLevel(app: Hono, caller: string, spaceUri: string): Promise { - const res = await call(app, "GET", `/xrpc/test.comm.community.whoami?spaceUri=${encodeURIComponent(spaceUri)}`, caller); + const res = await call(app, "GET", `/xrpc/test.comm.community.space.whoami?spaceUri=${encodeURIComponent(spaceUri)}`, caller); expect(res.status).toBe(200); return ((await res.json()) as any).accessLevel; } diff --git a/tests/community-e2e.test.ts b/tests/community-e2e.test.ts index 5203871..f9d5338 100644 --- a/tests/community-e2e.test.ts +++ b/tests/community-e2e.test.ts @@ -136,13 +136,13 @@ describe("community e2e — stage 1", () => { // whoami in both reserved spaces → owner for (const uri of [adminUri, publishersUri]) { - const r = await call(app, "GET", `/xrpc/test.comm.community.whoami?spaceUri=${encodeURIComponent(uri)}`, ALICE); + const r = await call(app, "GET", `/xrpc/test.comm.community.space.whoami?spaceUri=${encodeURIComponent(uri)}`, ALICE); expect(r.status).toBe(200); expect(((await r.json()) as any).accessLevel).toBe("owner"); } // Bob isn't in either - const r = await call(app, "GET", `/xrpc/test.comm.community.whoami?spaceUri=${encodeURIComponent(adminUri)}`, BOB); + const r = await call(app, "GET", `/xrpc/test.comm.community.space.whoami?spaceUri=${encodeURIComponent(adminUri)}`, BOB); expect(r.status).toBe(200); expect(((await r.json()) as any).accessLevel).toBe(null); }); @@ -255,7 +255,7 @@ describe("community e2e — stage 1", () => { const level = await call( app, "GET", - `/xrpc/test.comm.community.whoami?spaceUri=${encodeURIComponent(spaceUri)}`, + `/xrpc/test.comm.community.space.whoami?spaceUri=${encodeURIComponent(spaceUri)}`, BOB ); expect(((await level.json()) as any).accessLevel).toBe(null); diff --git a/tests/community-mint.test.ts b/tests/community-mint.test.ts index 96922a6..b573dfc 100644 --- a/tests/community-mint.test.ts +++ b/tests/community-mint.test.ts @@ -150,7 +150,7 @@ describe("community.mint — stage 4", () => { // Reserved spaces exist with the caller as owner. const adminUri = `at://${body.communityDid}/tools.atmo.event.space/$admin`; - const whoami = await call(app, "GET", `/xrpc/test.comm.community.whoami?spaceUri=${encodeURIComponent(adminUri)}`, ALICE); + const whoami = await call(app, "GET", `/xrpc/test.comm.community.space.whoami?spaceUri=${encodeURIComponent(adminUri)}`, ALICE); expect(((await whoami.json()) as any).accessLevel).toBe("owner"); }); diff --git a/tests/contrail-entry.test.ts b/tests/contrail-entry.test.ts new file mode 100644 index 0000000..f447a00 --- /dev/null +++ b/tests/contrail-entry.test.ts @@ -0,0 +1,95 @@ +/** Sanity-checks for the unified Contrail entry point: + * - `contrail.app()` returns a Hono instance with routes wired + * - `contrail.handler()` is a fetch-compatible function + * - the shared pubsub is threaded through so subscribing via the app + * receives events published via ingest-path helpers on the same instance. */ + +import { describe, it, expect } from "vitest"; +import { Contrail } from "../src/contrail"; +import { createSqliteDatabase } from "../src/adapters/sqlite"; +import { applyEvents } from "../src/core/db/records"; +import type { IngestEvent } from "../src/core/types"; + +const MSG_NSID = "app.event.message"; + +describe("Contrail.app() / Contrail.handler()", () => { + it("handler serves the health endpoint", async () => { + const db = createSqliteDatabase(":memory:"); + const contrail = new Contrail({ + namespace: "test.entry", + collections: { message: { collection: MSG_NSID } }, + db, + }); + await contrail.init(); + + const handle = contrail.handler(); + const res = await handle(new Request("http://localhost/health")); + expect(res.status).toBe(200); + expect(((await res.json()) as any).status).toBe("ok"); + }); + + it("shares its pubsub: events from applyEvents reach subscribers of contrail.app()", async () => { + const db = createSqliteDatabase(":memory:"); + const contrail = new Contrail({ + namespace: "test.entry", + collections: { message: { collection: MSG_NSID } }, + realtime: { ticketSecret: new Uint8Array(32).fill(1), keepaliveMs: 60_000 }, + db, + }); + await contrail.init(); + expect(contrail.pubsub).toBeTruthy(); + + const app = contrail.app(); + const ac = new AbortController(); + const res = await app.fetch( + new Request( + `http://localhost/xrpc/test.entry.realtime.subscribe?topic=${encodeURIComponent("collection:" + MSG_NSID)}`, + { signal: ac.signal } + ) + ); + expect(res.status).toBe(200); + const reader = res.body!.getReader(); + + // Use the same pubsub via the exposed getter — this is the promise + // of "one Contrail, one pubsub, one app". + const e: IngestEvent = { + uri: `at://did:plc:a/${MSG_NSID}/1`, + did: "did:plc:a", + time_us: 1, + collection: MSG_NSID, + operation: "create", + rkey: "1", + cid: "bafy", + record: JSON.stringify({ text: "shared" }), + indexed_at: Date.now() * 1000, + }; + await applyEvents(db, [e], contrail.config, { pubsub: contrail.pubsub ?? undefined }); + + // SSE frames separated by "\n\n". Skip any that don't carry a data: line + // (comments / keepalives). + const decoder = new TextDecoder(); + let buf = ""; + let payload: any = null; + while (!payload) { + const sep = buf.indexOf("\n\n"); + if (sep < 0) { + const { value, done } = await reader.read(); + if (done) break; + buf += decoder.decode(value, { stream: true }); + continue; + } + const frame = buf.slice(0, sep); + buf = buf.slice(sep + 2); + const dataLine = frame.split("\n").find((l) => l.startsWith("data:")); + if (dataLine) { + try { payload = JSON.parse(dataLine.slice(5).trim()); } catch {} + } + } + ac.abort(); + reader.cancel().catch(() => {}); + + expect(payload).toBeTruthy(); + expect(payload.kind).toBe("record.created"); + expect(payload.payload.did).toBe("did:plc:a"); + }); +}); diff --git a/tests/realtime-durable-object.test.ts b/tests/realtime-durable-object.test.ts index 0c79e34..3953e52 100644 --- a/tests/realtime-durable-object.test.ts +++ b/tests/realtime-durable-object.test.ts @@ -52,13 +52,14 @@ function mkEvent(overrides: Partial = {}): RealtimeEvent { topic: "space:at://x/y/z", kind: "record.created", payload: { - spaceUri: "at://x/y/z", + uri: "at://did:plc:x/c/r", + did: "did:plc:x", collection: "c", - authorDid: "did:plc:x", rkey: "r", cid: null, record: {}, - createdAt: 1, + time_us: 1, + space: "at://x/y/z", }, ts: 1, ...(overrides as any), @@ -96,7 +97,7 @@ describe("RealtimePubSubDO — publish/fanout", () => { doInstance.publishEvent({ topic: "space:x", kind: "member.removed", - payload: { spaceUri: "at://x/y/z", did: "did:plc:kickme" }, + payload: { space: "at://x/y/z", did: "did:plc:kickme" }, ts: 1, }); @@ -141,7 +142,7 @@ describe("RealtimePubSubDO — publish/fanout", () => { doInstance.publishEvent({ topic: "space:x", kind: "member.removed", - payload: { spaceUri: "at://x/y/z", did: "did:plc:self" }, + payload: { space: "at://x/y/z", did: "did:plc:self" }, ts: 8, }); diff --git a/tests/realtime-e2e.test.ts b/tests/realtime-e2e.test.ts index df11f61..7a3d1ec 100644 --- a/tests/realtime-e2e.test.ts +++ b/tests/realtime-e2e.test.ts @@ -191,7 +191,7 @@ describe("realtime e2e (in-memory pubsub, SSE transport)", () => { // Fat payload: subscribers can render the new record without a follow-up fetch. expect(event.payload.record).toEqual({ text: "hello" }); expect(event.payload.collection).toBe("app.event.message"); - expect(event.payload.authorDid).toBe(ALICE); + expect(event.payload.did).toBe(ALICE); close(); }); diff --git a/tests/realtime-foundations.test.ts b/tests/realtime-foundations.test.ts index 9d814d8..c6987d6 100644 --- a/tests/realtime-foundations.test.ts +++ b/tests/realtime-foundations.test.ts @@ -11,13 +11,14 @@ function mk(topic: string, n: number): RealtimeEvent { topic, kind: "record.created", payload: { - spaceUri: "at://x/y/z", + uri: `at://did:plc:x/c/${n}`, + did: "did:plc:x", collection: "c", - authorDid: "did:plc:x", rkey: String(n), cid: null, record: {}, - createdAt: n, + time_us: n, + space: "at://x/y/z", }, ts: n, }; diff --git a/tests/realtime-public.test.ts b/tests/realtime-public.test.ts new file mode 100644 index 0000000..34b13f3 --- /dev/null +++ b/tests/realtime-public.test.ts @@ -0,0 +1,161 @@ +/** Realtime — public topics (`collection:`, `actor:`): + * - Subscribe requires no auth. + * - Events are published by `applyEvents` (the jetstream/public-record path), + * not by the spaces adapter. Spaces can be entirely absent from config. */ + +import { describe, it, expect } from "vitest"; +import { Hono } from "hono"; +import { createSqliteDatabase } from "../src/adapters/sqlite"; +import { initSchema } from "../src/core/db/schema"; +import { applyEvents } from "../src/core/db/records"; +import { createApp } from "../src/core/router"; +import { resolveConfig } from "../src/core/types"; +import type { ContrailConfig, IngestEvent } from "../src/core/types"; +import { InMemoryPubSub } from "../src/core/realtime/in-memory"; +import type { RealtimeEvent } from "../src/core/realtime/types"; + +const REALTIME_SECRET = new Uint8Array(32).fill(9); + +const ALICE = "did:plc:alice"; +const MSG_NSID = "app.event.message"; + +function baseConfig(pubsub: InMemoryPubSub): ContrailConfig { + return { + namespace: "test.pub", + collections: { message: { collection: MSG_NSID } }, + realtime: { + ticketSecret: REALTIME_SECRET, + pubsub, + keepaliveMs: 60_000, + }, + }; +} + +/** Open SSE and return an async iterator over decoded events. */ +async function openSse(app: Hono, path: string): Promise<{ + events: AsyncIterator; + close: () => void; +}> { + const ac = new AbortController(); + const res = await app.fetch( + new Request(`http://localhost${path}`, { method: "GET", signal: ac.signal }) + ); + if (!res.ok) throw new Error(`SSE open failed: ${res.status} ${await res.text()}`); + const reader = res.body!.getReader(); + const decoder = new TextDecoder(); + let buf = ""; + const iter: AsyncIterator = { + async next() { + while (true) { + const sep = buf.indexOf("\n\n"); + if (sep >= 0) { + const frame = buf.slice(0, sep); + buf = buf.slice(sep + 2); + for (const line of frame.split("\n")) { + if (line.startsWith("data:")) { + try { return { done: false, value: JSON.parse(line.slice(5).trim()) }; } catch {} + } + } + continue; + } + const { value, done } = await reader.read(); + if (done) return { done: true, value: undefined as any }; + buf += decoder.decode(value, { stream: true }); + } + }, + }; + return { + events: iter, + close: () => { ac.abort(); reader.cancel().catch(() => {}); }, + }; +} + +describe("realtime — public topics work without spaces and without auth", () => { + it("subscribes to collection: without auth and receives jetstream events", async () => { + const db = createSqliteDatabase(":memory:"); + const pubsub = new InMemoryPubSub(); + const config = resolveConfig(baseConfig(pubsub)); + await initSchema(db, config); + const app = createApp(db, config); + + const { events, close } = await openSse( + app, + `/xrpc/test.pub.realtime.subscribe?topic=${encodeURIComponent("collection:" + MSG_NSID)}` + ); + + // Simulate a public record landing via jetstream. + const e: IngestEvent = { + uri: `at://${ALICE}/${MSG_NSID}/abc`, + did: ALICE, + time_us: 1_700_000_000_000_000, + collection: MSG_NSID, + operation: "create", + rkey: "abc", + cid: "bafytest", + record: JSON.stringify({ text: "hi" }), + indexed_at: Date.now() * 1000, + }; + await applyEvents(db, [e], config, { pubsub }); + + const next = await events.next(); + expect(next.done).toBe(false); + const event = next.value as RealtimeEvent & { kind: "record.created" }; + expect(event.kind).toBe("record.created"); + expect(event.payload.uri).toBe(e.uri); + expect(event.payload.did).toBe(ALICE); + expect(event.payload.collection).toBe(MSG_NSID); + expect(event.payload.record).toEqual({ text: "hi" }); + expect((event.payload as any).space).toBeUndefined(); + close(); + }); + + it("actor: is subscribable without auth and receives events for that DID", async () => { + const db = createSqliteDatabase(":memory:"); + const pubsub = new InMemoryPubSub(); + const config = resolveConfig(baseConfig(pubsub)); + await initSchema(db, config); + const app = createApp(db, config); + + const { events, close } = await openSse( + app, + `/xrpc/test.pub.realtime.subscribe?topic=${encodeURIComponent("actor:" + ALICE)}` + ); + + const e: IngestEvent = { + uri: `at://${ALICE}/${MSG_NSID}/xyz`, + did: ALICE, + time_us: 1_700_000_000_000_000, + collection: MSG_NSID, + operation: "create", + rkey: "xyz", + cid: "bafy", + record: JSON.stringify({ text: "from actor feed" }), + indexed_at: Date.now() * 1000, + }; + await applyEvents(db, [e], config, { pubsub }); + + const next = await events.next(); + expect(next.done).toBe(false); + const event = next.value as RealtimeEvent & { kind: "record.created" }; + expect(event.topic).toBe(`actor:${ALICE}`); + expect(event.payload.did).toBe(ALICE); + close(); + }); + + it("private topic without auth middleware returns 400 with a helpful error", async () => { + const db = createSqliteDatabase(":memory:"); + const pubsub = new InMemoryPubSub(); + const config = resolveConfig(baseConfig(pubsub)); + await initSchema(db, config); + const app = createApp(db, config); + + const res = await app.fetch( + new Request( + `http://localhost/xrpc/test.pub.realtime.subscribe?topic=${encodeURIComponent("space:at://x/y/z")}` + ) + ); + expect(res.status).toBe(400); + const body = (await res.json()) as any; + expect(body.reason).toBe("private-topic-without-auth"); + }); +}); diff --git a/tests/spaces-invites.test.ts b/tests/spaces-invites.test.ts index 22219b3..82f58f7 100644 --- a/tests/spaces-invites.test.ts +++ b/tests/spaces-invites.test.ts @@ -6,7 +6,7 @@ import { initSchema } from "../src/core/db/schema"; import { createApp } from "../src/core/router"; import { resolveConfig } from "../src/core/types"; import type { ContrailConfig } from "../src/core/types"; -import { generateInviteToken, hashInviteToken } from "../src/core/spaces/invite-token"; +import { generateInviteToken, hashInviteToken } from "../src/core/invite/token"; const ALICE = "did:plc:alice"; const BOB = "did:plc:bob"; @@ -84,14 +84,14 @@ describe("invite e2e", () => { }); it("non-owner cannot create an invite", async () => { - const res = await call(app, "POST", "/xrpc/test.spaces.space.invite.create", BOB, { + const res = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.create", BOB, { spaceUri, }); expect(res.status).toBe(403); }); it("owner creates an invite and Bob redeems it to become a member", async () => { - const create = await call(app, "POST", "/xrpc/test.spaces.space.invite.create", ALICE, { + const create = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.create", ALICE, { spaceUri, }); expect(create.status).toBe(200); @@ -101,7 +101,7 @@ describe("invite e2e", () => { expect(invite.spaceUri).toBe(spaceUri); expect(invite.usedCount).toBe(0); - const redeem = await call(app, "POST", "/xrpc/test.spaces.space.invite.redeem", BOB, { token }); + const redeem = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.redeem", BOB, { token }); expect(redeem.status).toBe(200); const body = (await redeem.json()) as any; expect(body.spaceUri).toBe(spaceUri); @@ -116,58 +116,58 @@ describe("invite e2e", () => { }); it("single-use invite rejects second redemption", async () => { - const create = await call(app, "POST", "/xrpc/test.spaces.space.invite.create", ALICE, { + const create = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.create", ALICE, { spaceUri, maxUses: 1, }); const { token } = (await create.json()) as any; - const first = await call(app, "POST", "/xrpc/test.spaces.space.invite.redeem", BOB, { token }); + const first = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.redeem", BOB, { token }); expect(first.status).toBe(200); - const second = await call(app, "POST", "/xrpc/test.spaces.space.invite.redeem", CHARLIE, { token }); + const second = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.redeem", CHARLIE, { token }); expect(second.status).toBe(400); const body = (await second.json()) as any; expect(body.reason).toBe("expired-revoked-or-exhausted"); }); it("expired invite rejects redemption", async () => { - const create = await call(app, "POST", "/xrpc/test.spaces.space.invite.create", ALICE, { + const create = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.create", ALICE, { spaceUri, expiresAt: Date.now() - 1000, }); const { token } = (await create.json()) as any; - const res = await call(app, "POST", "/xrpc/test.spaces.space.invite.redeem", CHARLIE, { token }); + const res = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.redeem", CHARLIE, { token }); expect(res.status).toBe(400); }); it("revoked invite rejects redemption and list filters it by default", async () => { - const create = await call(app, "POST", "/xrpc/test.spaces.space.invite.create", ALICE, { + const create = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.create", ALICE, { spaceUri, }); const { token, invite } = (await create.json()) as any; - const revoke = await call(app, "POST", "/xrpc/test.spaces.space.invite.revoke", ALICE, { + const revoke = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.revoke", ALICE, { spaceUri, tokenHash: invite.tokenHash, }); expect(revoke.status).toBe(200); expect(((await revoke.json()) as any).ok).toBe(true); - const tryRedeem = await call(app, "POST", "/xrpc/test.spaces.space.invite.redeem", CHARLIE, { token }); + const tryRedeem = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.redeem", CHARLIE, { token }); expect(tryRedeem.status).toBe(400); - const listActive = await call(app, "GET", `/xrpc/test.spaces.space.invite.list?spaceUri=${encodeURIComponent(spaceUri)}`, ALICE); + const listActive = await call(app, "GET", `/xrpc/test.spaces.spaceExt.invite.list?spaceUri=${encodeURIComponent(spaceUri)}`, ALICE); const activeHashes = ((await listActive.json()) as any).invites.map((i: any) => i.tokenHash); expect(activeHashes).not.toContain(invite.tokenHash); - const listAll = await call(app, "GET", `/xrpc/test.spaces.space.invite.list?spaceUri=${encodeURIComponent(spaceUri)}&includeRevoked=true`, ALICE); + const listAll = await call(app, "GET", `/xrpc/test.spaces.spaceExt.invite.list?spaceUri=${encodeURIComponent(spaceUri)}&includeRevoked=true`, ALICE); const allHashes = ((await listAll.json()) as any).invites.map((i: any) => i.tokenHash); expect(allHashes).toContain(invite.tokenHash); }); it("non-owner cannot list or revoke invites", async () => { - const listRes = await call(app, "GET", `/xrpc/test.spaces.space.invite.list?spaceUri=${encodeURIComponent(spaceUri)}`, BOB); + const listRes = await call(app, "GET", `/xrpc/test.spaces.spaceExt.invite.list?spaceUri=${encodeURIComponent(spaceUri)}`, BOB); expect(listRes.status).toBe(403); - const revokeRes = await call(app, "POST", "/xrpc/test.spaces.space.invite.revoke", BOB, { + const revokeRes = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.revoke", BOB, { spaceUri, tokenHash: "nonexistent", }); expect(revokeRes.status).toBe(403); @@ -179,7 +179,7 @@ describe("invite e2e", () => { spaceUri, collection: "app.event.message", record: { text: "hello world" }, }); - const create = await call(app, "POST", "/xrpc/test.spaces.space.invite.create", ALICE, { + const create = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.create", ALICE, { spaceUri, kind: "read", }); const { token, invite } = (await create.json()) as any; @@ -193,12 +193,12 @@ describe("invite e2e", () => { expect(body.records.length).toBeGreaterThan(0); // Read token cannot be redeemed for membership. - const redeem = await call(app, "POST", "/xrpc/test.spaces.space.invite.redeem", CHARLIE, { token }); + const redeem = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.redeem", CHARLIE, { token }); expect(redeem.status).toBe(400); }); it("read-join token grants anonymous read AND can be redeemed for membership", async () => { - const create = await call(app, "POST", "/xrpc/test.spaces.space.invite.create", ALICE, { + const create = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.create", ALICE, { spaceUri, kind: "read-join", }); const { token, invite } = (await create.json()) as any; @@ -210,16 +210,16 @@ describe("invite e2e", () => { expect(anon.status).toBe(200); // Same token can also be redeemed by a signed-in user. - const redeem = await call(app, "POST", "/xrpc/test.spaces.space.invite.redeem", CHARLIE, { token }); + const redeem = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.redeem", CHARLIE, { token }); expect(redeem.status).toBe(200); }); it("revoked read-token rejects anonymous read", async () => { - const create = await call(app, "POST", "/xrpc/test.spaces.space.invite.create", ALICE, { + const create = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.create", ALICE, { spaceUri, kind: "read", }); const { token, invite } = (await create.json()) as any; - await call(app, "POST", "/xrpc/test.spaces.space.invite.revoke", ALICE, { + await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.revoke", ALICE, { spaceUri, tokenHash: invite.tokenHash, }); const url = `/xrpc/test.spaces.space.listRecords?spaceUri=${encodeURIComponent(spaceUri)}&collection=app.event.message&inviteToken=${token}`; @@ -228,7 +228,7 @@ describe("invite e2e", () => { }); it("read-token does not grant write", async () => { - const create = await call(app, "POST", "/xrpc/test.spaces.space.invite.create", ALICE, { + const create = await call(app, "POST", "/xrpc/test.spaces.spaceExt.invite.create", ALICE, { spaceUri, kind: "read", }); const { token } = (await create.json()) as any;