···3636| Owner | `owner`; only meaningful in `$admin` for owner-management | ✅ | Post's level 8 |
3737| 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 |
3838| Cross-community / cross-arbiter delegation | _same-contrail only_ | ❌ | v1 constraint. Private-membership federation is deferred — see [community.md § Deferred work] |
3939-| Invites as a separate service | _not implemented_ | ❌ | Post proposes an optional add-on; contrail has spaces invites to draw from when this lands |
3939+| 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. |
4040| 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 |
4141| Public membership-list flag | _none_ | ❌ | Post allows spaces to expose membership publicly. Ruled out for cross-instance federation (privacy) |
4242| Space credential (from arbiter) | _none_ | ⚠️ | Covered by the spaces module's service-auth story, not the community module |
···6363### Membership
6464- `community.space.grant` · `community.space.revoke` · `community.space.setAccessLevel`
6565- `community.space.listMembers` (`?flatten=true` for the resolved DID list) · `community.space.resync`
6666+6767+### Invites
6868+- `community.invite.create` · `community.invite.list` · `community.invite.revoke` · `community.invite.redeem`
66696770### Publishing
6871- `community.putRecord` · `community.deleteRecord` (public records via the community's PDS; adopted only)
+34
src/generate.ts
···690690 ...hydrateDefs, ...refDefs, ...profileDefs(),
691691 },
692692 });
693693+694694+ // watchRecords — realtime subscription over the same query shape.
695695+ // Only emitted when realtime is configured. SSE-only (subscription
696696+ // lexicon doesn't cleanly model custom event framing, so we describe it
697697+ // as a query with a streaming output and document the event kinds.)
698698+ if (config.realtime) {
699699+ writeLexicon(`${ns}.${shortName}.watchRecords`, {
700700+ lexicon: 1,
701701+ id: `${ns}.${shortName}.watchRecords`,
702702+ defs: {
703703+ main: {
704704+ type: "query",
705705+ description:
706706+ `Subscribe to a live ${collection} query. Returns Server-Sent Events: ` +
707707+ `\`snapshot.start\`, \`snapshot.record\` (N of), \`snapshot.end\`, ` +
708708+ `then \`record.created\` / \`record.deleted\` as records enter/leave the result, ` +
709709+ `plus a final \`member.removed\` if the caller loses access mid-stream. ` +
710710+ `v1 requires spaceUri; cross-space watch is deferred.`,
711711+ parameters: { type: "params", properties: listParams },
712712+ output: {
713713+ encoding: "text/event-stream",
714714+ schema: { type: "object", properties: {} },
715715+ },
716716+ errors: [
717717+ { name: "InvalidRequest" },
718718+ { name: "AuthRequired" },
719719+ { name: "Forbidden" },
720720+ { name: "NotFound" },
721721+ { name: "NotSupported" },
722722+ ],
723723+ },
724724+ },
725725+ });
726726+ }
693727 }
694728695729 // --- getRecord ---
+41
src/index.ts
···3535 SpacesConfig,
3636 AppPolicy,
3737 AppPolicyMode,
3838+ AuthOverrideResult,
3839 SpaceRow,
3940 SpaceMemberRow,
4041 StoredRecord,
···5051export { HostedAdapter } from "./core/spaces/adapter";
5152export { nextTid } from "./core/spaces/tid";
5253export { generateInviteToken, hashInviteToken } from "./core/spaces/invite-token";
5454+export {
5555+ MemoryBlobAdapter,
5656+ R2BlobAdapter,
5757+ blobKey,
5858+} from "./core/spaces/blob-adapter";
5959+export type { BlobAdapter, BlobUploadMeta, R2BucketLike } from "./core/spaces/blob-adapter";
6060+export type { SpacesBlobsConfig, BlobMetaRow } from "./core/spaces/types";
53615462// Realtime
5563export type {
···7280 collectionTopic,
7381 communityTopic,
7482 spaceTopic,
8383+ registerRealtimeRoutes,
7584} from "./core/realtime";
8585+export type {
8686+ DurableObjectId,
8787+ DurableObjectNamespace,
8888+ DurableObjectStub,
8989+ DurableObjectState,
9090+} from "./core/realtime/durable-object";
9191+9292+// Community
9393+export {
9494+ CommunityAdapter,
9595+ CredentialCipher,
9696+ registerCommunityRoutes,
9797+ ACCESS_LEVELS,
9898+ RESERVED_KEYS,
9999+ isAccessLevel,
100100+ isReservedKey,
101101+ rankOf,
102102+ resolveEffectiveLevel,
103103+ flattenEffectiveMembers,
104104+ wouldCycle,
105105+ reconcile,
106106+} from "./core/community";
107107+export type {
108108+ CommunityConfig,
109109+ CommunityMode,
110110+ CommunityRow,
111111+ CommunityInviteRow,
112112+ CreateCommunityInviteInput,
113113+ AccessLevel,
114114+ AccessLevelRow,
115115+ ReservedKey,
116116+} from "./core/community";
···4455# Set to your tunnel URL to use a confidential client in dev
66# OAUTH_PUBLIC_URL=https://your-tunnel.trycloudflare.com
77+88+# 32-byte base64 secret used to envelope-encrypt community signing keys.
99+# Generate with: openssl rand -base64 32
1010+COMMUNITY_MASTER_KEY=
1111+1212+# 32-byte base64 secret used to HMAC-sign realtime subscription tickets.
1313+# Generate with: openssl rand -base64 32
1414+REALTIME_TICKET_SECRET=
1515+1616+# Service DID for JWT verification (aud claim). did:web:<your-host> for prod.
1717+SERVICE_DID=did:web:localhost
1818+1919+# Dev-only: trust the HMAC-signed session cookie instead of atproto service-auth
2020+# JWTs, so the loopback OAuth client can reach contrail without a tunnel. NEVER
2121+# set this in production — it turns the cookie into the only auth signal.
2222+DEV_AUTH=1
+84-44
examples/sveltekit-group-chat/README.md
···11-# sveltekit + contrail + atproto OAuth
11+# sveltekit-group-chat
2233-A sveltekit example app on cloudflare workers with self-hosted atproto record indexing via [contrail](https://github.com/flo-bit/contrail), fully typed queries, and OAuth authentication.
33+A Discord-like demo built on top of [contrail](https://github.com/flo-bit/contrail) — SvelteKit + Cloudflare Workers + D1, OAuth, spaces, communities, and realtime (SSE).
44+55+## What this exercises
66+77+- **`community`** — mint a fresh `did:plc` per server, tiered access levels (member / manager / admin / owner), space → space delegation.
88+- **`spaces`** — permissioned per-channel record tables, membership pushed in by the community reconciler.
99+- **`realtime`** — single SSE connection subscribed to `community:<did>`, fanning out per-channel events with unread dots derived client-side.
1010+- Three record collections — `tools.atmo.chat.server`, `tools.atmo.chat.channel`, `tools.atmo.chat.message` — with unified `listRecords` for "every server / channel I can see" in one call.
411512## Setup
613714```sh
815pnpm install
1616+cp .env.example .env
1717+# fill in COMMUNITY_MASTER_KEY and REALTIME_TICKET_SECRET with
1818+# openssl rand -base64 32
1919+pnpm generate:pull # emits lexicons + src/lib/atproto/generated-methods.ts
920pnpm dev
1021```
11221212-Dev mode uses a loopback OAuth client — no keys or Cloudflare setup needed.
2323+Re-run `pnpm generate` whenever you add/remove collections or toggle contrail
2424+modules in `src/lib/contrail/config.ts`: the OAuth scope list is derived from
2525+it, so the consent screen only asks for permissions you actually use.
13261414-## Config
2727+Dev mode uses a loopback OAuth client — no Cloudflare setup needed. The realtime Durable Object runs locally via miniflare.
15281616-Define which AT Protocol collections to index in `src/lib/contrail/config.ts`:
2929+### Auth in dev vs prod
17301818-```ts
1919-import type { ContrailConfig } from '@atmo-dev/contrail';
3131+Contrail normally auths every write with an atproto **service-auth JWT**. The browser delegates minting those JWTs to the SvelteKit worker, which calls `com.atproto.server.getServiceAuth` on the user's PDS.
20322121-export const config: ContrailConfig = {
2222- namespace: 'statusphere.app',
2323- collections: {
2424- status: { // short name → URL path segment
2525- collection: 'xyz.statusphere.status', // full NSID of the record type
2626- queryable: {
2727- status: {}, // equality filter (?status=...)
2828- createdAt: { type: 'range' } // range filter (?createdAtMin=...&createdAtMax=...)
2929- }
3030- }
3131- }
3232-};
3333+bsky.social **refuses** that call for loopback OAuth clients, so out of the box a plain `pnpm dev` couldn't talk to contrail without a [cloudflared tunnel](#with-a-tunnel).
3434+3535+To skip the tunnel this example ships a **dev auth bypass** gated on `DEV_AUTH=1` in `.env`:
3636+3737+- Contrail's auth middleware trusts the HMAC-signed `did` session cookie the OAuth flow already sets, and uses its DID as the authenticated caller.
3838+- The cookie is HMAC-signed with `COOKIE_SECRET`, so only this worker can mint a valid one.
3939+- All access-level checks (admin+, manager+, etc.) still apply — the bypass only replaces the JWT step.
4040+4141+**Never set `DEV_AUTH=1` in production** — that would make the cookie the sole auth signal for third-party clients.
4242+4343+### With a tunnel
4444+4545+If you want to exercise the real JWT path in dev, run a tunnel and drop `DEV_AUTH`:
4646+4747+```sh
4848+pnpm tunnel # prints https://xxx.trycloudflare.com
4949+# in .env: set OAUTH_PUBLIC_URL to the tunnel URL and remove DEV_AUTH
5050+pnpm dev
5151+# open the tunnel URL (not localhost)
3352```
34533535-After changing the config, run `pnpm generate:pull` to regenerate lexicons and types.
5454+## How the pieces connect
36553737-Run `pnpm sync` to backfill existing records from the network.
5656+```
5757+┌────────────┐ service-auth JWT ┌─────────────────┐ contrail XRPCs ┌───────────┐
5858+│ browser │ ─────────────────► │ SvelteKit worker │ ─────────────────► │ contrail │
5959+└────────────┘ └─────────────────┘ └───────────┘
6060+ │ │ │
6161+ │ EventSource /xrpc/.../realtime.subscribe?ticket=... ◄──────── realtime DO │
6262+ └───────────────────────────────────────────────────────────────────────────┘
6363+```
38643939-Wrangler bindings (`wrangler.jsonc`):
6565+- The browser can't mint service-auth JWTs itself — it delegates to the SvelteKit worker, which uses the user's OAuth session to call `com.atproto.server.getServiceAuth` on their PDS. The resulting JWT is scoped to one `lxm`.
6666+- Realtime tickets are minted the same way, then the browser holds an `EventSource` open against the subscribe endpoint.
40674141-- **D1** (`DB`) — contrail's database
4242-- **KV** (`OAUTH_SESSIONS`, `OAUTH_STATES`) — OAuth session storage
4343-- **Vars** — `CRON_SECRET`, `OAUTH_PUBLIC_URL`, `CLIENT_ASSERTION_KEY`, `COOKIE_SECRET`
6868+## Data model
6969+7070+| Record | Lives in | Author |
7171+|---|---|---|
7272+| `tools.atmo.chat.server` | community's `members` space, rkey `self` | community DID (via `community.space.putRecord`) |
7373+| `tools.atmo.chat.channel` | channel's own space, rkey `self` | community DID |
7474+| `tools.atmo.chat.message` | channel's own space, rkey = TID | user DID |
7575+7676+Only admins (`admin+` in the target space) can write the server and channel records, and the community DID is the author. Reads filter by `actor=<communityDid>` so spoof records from random members are silently ignored.
7777+7878+## Feature map
7979+8080+| Route | What it does |
8181+|---|---|
8282+| `/` | "My servers" — unified `server.listRecords` over every space I'm in |
8383+| `/new` | Mint community, bootstrap `members` role-space, write server record, show recovery key once |
8484+| `/c/[communityDid]` | Redirects to the first channel you can see |
8585+| `/c/[communityDid]/[channelKey]` | Chat view — messages via `space.listRecords` + realtime |
8686+| `/c/[communityDid]/settings/members` | Add, promote, demote, revoke members |
8787+8888+## Private channels
8989+9090+New-channel modal offers a visibility toggle:
9191+9292+- **Public** — grants the `members` role-space `member` access on the new channel. Reconciler fans out to everyone.
9393+- **Private** — grants each picked DID directly. Also grants `members` (so they can see the server header) but not automatic access to other channels.
44944595## Deploy
46964797```sh
4848-npx wrangler d1 create statusphere
4949-# Add database_id to wrangler.jsonc
9898+npx wrangler d1 create group-chat
9999+# add the database_id to wrangler.jsonc
100100+npx wrangler secret put COMMUNITY_MASTER_KEY
101101+npx wrangler secret put REALTIME_TICKET_SECRET
5010251103pnpm build
52104npx wrangler deploy
53105```
541065555-## How it works
107107+## Known gaps
561085757-**Contrail** indexes AT Protocol records into D1 via Jetstream (cron, every minute). When a user posts, `contrail.notify()` indexes it immediately.
109109+This is a demo. It deliberately skips:
581105959-**Typed queries** use `@atcute/client` with an in-process handler — full type safety, zero HTTP overhead:
6060-6161-```ts
6262-const client = getClient(platform!.env.DB);
6363-const res = await client.get('statusphere.app.status.listRecords', {
6464- params: { limit: 50, profiles: true } // typed params
6565-});
6666-res.data.records // typed response
6767-```
6868-6969-Types are generated from contrail's config via `pnpm generate:pull`, which produces lexicon JSON and TypeScript types that register with `@atcute/client`.
7070-7171-**Scheduled ingestion** works around SvelteKit's lack of `scheduled` export support ([sveltejs/kit#4841](https://github.com/sveltejs/kit/issues/4841)) by appending a handler post-build that self-calls `/api/cron`.
7272-7373-**OAuth** uses `@atcute/oauth-node-client` with KV-backed sessions and HMAC-signed cookies.
111111+- **`adopt` mode** — only `mint` is wired up.
112112+- **`$publishers`** / community-authored public posts.
113113+- Threads, reactions, image uploads, rich text, typing indicators.
7411475115## License
76116
···11/**
22- * Post-build script: appends a `scheduled` handler to the SvelteKit worker output.
33- *
44- * SvelteKit's adapter-cloudflare doesn't support the `scheduled` export natively
55- * (see https://github.com/sveltejs/kit/issues/4841). This script patches the
66- * generated _worker.js to add one that self-calls the /api/cron endpoint.
22+ * Post-build script:
33+ * - appends a `scheduled` handler to the SvelteKit worker output
44+ * (adapter-cloudflare doesn't support scheduled exports natively:
55+ * https://github.com/sveltejs/kit/issues/4841)
66+ * - re-exports the realtime Durable Object class so wrangler's class_name binding resolves.
77 */
88import { readFileSync, writeFileSync } from 'fs';
99import { join, dirname } from 'path';
···16161717code += `
1818// --- Appended by scripts/append-scheduled.ts ---
1919+import { RealtimePubSubDO as __RealtimePubSubDO } from "@atmo-dev/contrail";
2020+export { __RealtimePubSubDO as RealtimePubSubDO };
2121+1922worker_default.scheduled = async function (event, env, ctx) {
2023 const req = new Request('http://localhost/api/cron', {
2124 method: 'POST',
···2629`;
27302831writeFileSync(workerPath, code);
2929-console.log('Appended scheduled handler to _worker.js');
3232+console.log('Appended scheduled handler + RealtimePubSubDO re-export to _worker.js');
+27-4
examples/sveltekit-group-chat/scripts/generate.ts
···11import { join, dirname } from 'path';
22import { fileURLToPath } from 'url';
33-import { config } from '../src/lib/contrail/config';
44-import { generateLexicons } from '@atmo-dev/contrail/generate';
33+import { writeFileSync } from 'fs';
44+import { baseConfig } from '../src/lib/contrail/config';
55+import { generateLexicons, extractXrpcMethods } from '@atmo-dev/contrail/generate';
5667const ROOT_DIR = join(dirname(fileURLToPath(import.meta.url)), '..');
7888-generateLexicons({
99- config,
99+// Use a config that has spaces/community/realtime declared so the generator
1010+// emits the full surface (space.*, community.*, realtime.*) alongside the
1111+// per-collection endpoints. Values are stubs — only the shape matters for codegen.
1212+const configForGen = {
1313+ ...baseConfig,
1414+ spaces: { type: 'tools.atmo.chat.space', serviceDid: 'did:web:localhost' },
1515+ community: { masterKey: new Uint8Array(32), serviceDid: 'did:web:localhost' },
1616+ realtime: { ticketSecret: new Uint8Array(32) }
1717+};
1818+1919+const generated = generateLexicons({
2020+ config: configForGen,
1021 rootDir: ROOT_DIR,
1122 outputDir: join(ROOT_DIR, 'lexicons-generated'),
1223 writeRuntimeFiles: true
1324});
2525+2626+// Emit the sorted XRPC method list as a TS module. settings.ts imports it and
2727+// feeds it into `scope.rpc({ lxm: [...], aud })` so the OAuth consent screen
2828+// requests exactly the methods contrail exposes (no more, no less).
2929+const methods = extractXrpcMethods(generated);
3030+const out = `// Auto-generated by scripts/generate.ts. Do not edit by hand.
3131+// Source of truth: the set of XRPC methods contrail emits for this namespace.
3232+3333+export const xrpcMethods = ${JSON.stringify(methods, null, '\t')} as const;
3434+`;
3535+writeFileSync(join(ROOT_DIR, 'src/lib/atproto/generated-methods.ts'), out);
3636+console.log(`wrote generated-methods.ts (${methods.length} methods)`);
+7
examples/sveltekit-group-chat/src/app.d.ts
···33import type { OAuthSession } from '@atcute/oauth-node-client';
44import type { Client } from '@atcute/client';
55import type { Did } from '@atcute/lexicons';
66+import type { DurableObjectNamespace } from '@atmo-dev/contrail';
6778declare global {
89 namespace App {
···2425 PROFILE_CACHE?: KVNamespace;
2526 DB: D1Database;
2627 CRON_SECRET: string;
2828+ COMMUNITY_MASTER_KEY: string;
2929+ REALTIME_TICKET_SECRET: string;
3030+ SERVICE_DID: string;
3131+ DEV_AUTH?: string;
3232+ REALTIME: DurableObjectNamespace;
3333+ BLOBS?: R2Bucket;
2734 };
2835 }
2936 }
···4848 return "did:key:z" + base58btcEncode(combined);
4949}
50505151+/** Order of the P-256 curve (n). */
5252+const P256_N = BigInt(
5353+ "0xFFFFFFFF00000000FFFFFFFFFFFFFFFFBCE6FAADA7179E84F3B9CAC2FC632551"
5454+);
5555+const P256_N_HALF = P256_N >> 1n;
5656+5157async function signBytes(privateJwk: JsonWebKey, bytes: Uint8Array): Promise<Uint8Array> {
5258 const key = await crypto.subtle.importKey(
5359 "jwk",
···5662 false,
5763 ["sign"]
5864 );
5959- const sig = await crypto.subtle.sign(
6060- { name: "ECDSA", hash: "SHA-256" },
6161- key,
6262- bytes as BufferSource
6565+ const sig = new Uint8Array(
6666+ await crypto.subtle.sign({ name: "ECDSA", hash: "SHA-256" }, key, bytes as BufferSource)
6367 );
6464- return new Uint8Array(sig);
6868+ // Web Crypto returns IEEE P1363 form: r || s (32 bytes each for P-256).
6969+ // atproto/PLC (and most modern ECDSA consumers) require low-S — i.e.
7070+ // s must be in the lower half of the curve order. Flip high-S signatures
7171+ // by replacing s with n - s. This yields an equivalent valid signature.
7272+ return normalizeLowS(sig);
7373+}
7474+7575+function normalizeLowS(sig: Uint8Array): Uint8Array {
7676+ if (sig.length !== 64) return sig; // not a P-256 P1363 sig — don't touch
7777+ const r = sig.slice(0, 32);
7878+ const sBytes = sig.slice(32);
7979+ const s = bytesToBigInt(sBytes);
8080+ if (s <= P256_N_HALF) return sig;
8181+ const sLow = P256_N - s;
8282+ const sLowBytes = bigIntToBytes(sLow, 32);
8383+ const out = new Uint8Array(64);
8484+ out.set(r, 0);
8585+ out.set(sLowBytes, 32);
8686+ return out;
8787+}
8888+8989+function bytesToBigInt(b: Uint8Array): bigint {
9090+ let v = 0n;
9191+ for (const byte of b) v = (v << 8n) | BigInt(byte);
9292+ return v;
9393+}
9494+9595+function bigIntToBytes(v: bigint, length: number): Uint8Array {
9696+ const out = new Uint8Array(length);
9797+ let x = v;
9898+ for (let i = length - 1; i >= 0; i--) {
9999+ out[i] = Number(x & 0xffn);
100100+ x >>= 8n;
101101+ }
102102+ return out;
65103}
6610467105// ============================================================================
+164
src/core/community/router.ts
···2525 RESERVED_KEYS,
2626} from "./types";
2727import type { ServiceJwtVerifier } from "@atcute/xrpc-server/auth";
2828+import {
2929+ generateInviteToken,
3030+ hashInviteToken,
3131+} from "../spaces/invite-token";
28322933export interface CommunityRoutesOptions {
3034 /** Override auth middleware for tests. */
···970974 identifier,
971975 });
972976 return c.json({ ok: true });
977977+ });
978978+979979+ // ==========================================================================
980980+ // Invites — pre-signed grants for community-owned spaces. Admin/manager
981981+ // creates; any authenticated user with the raw token redeems once.
982982+ // ==========================================================================
983983+984984+ app.post(`/xrpc/${NS}.invite.create`, auth, async (c) => {
985985+ const sa = getAuth(c);
986986+ const body = (await c.req.json().catch(() => null)) as
987987+ | {
988988+ spaceUri?: string;
989989+ accessLevel?: string;
990990+ expiresAt?: number;
991991+ maxUses?: number;
992992+ note?: string;
993993+ }
994994+ | null;
995995+ if (!body?.spaceUri || !body.accessLevel) {
996996+ return c.json(
997997+ { error: "InvalidRequest", message: "spaceUri and accessLevel required" },
998998+ 400
999999+ );
10001000+ }
10011001+ if (!isAccessLevel(body.accessLevel)) {
10021002+ return c.json({ error: "InvalidRequest", message: "invalid accessLevel" }, 400);
10031003+ }
10041004+ const space = await spaces.getSpace(body.spaceUri);
10051005+ if (!space) return c.json({ error: "NotFound" }, 404);
10061006+ const communityRow = await community.getCommunity(space.ownerDid);
10071007+ if (!communityRow) {
10081008+ return c.json({ error: "InvalidRequest", reason: "not-community-owned" }, 400);
10091009+ }
10101010+10111011+ // Caller must have at least manager on the target space, and cannot create
10121012+ // an invite that confers a higher level than their own.
10131013+ const callerLevel = await resolveEffectiveLevel(community, body.spaceUri, sa.issuer);
10141014+ if (!callerLevel || rankOf(callerLevel) < rankOf("manager")) {
10151015+ return c.json({ error: "Forbidden", reason: "manager-required" }, 403);
10161016+ }
10171017+ if (rankOf(body.accessLevel) > rankOf(callerLevel)) {
10181018+ return c.json({ error: "Forbidden", reason: "cannot-grant-higher-than-self" }, 403);
10191019+ }
10201020+10211021+ const token = generateInviteToken();
10221022+ const tokenHash = await hashInviteToken(token);
10231023+10241024+ const row = await community.createInvite({
10251025+ spaceUri: body.spaceUri,
10261026+ tokenHash,
10271027+ accessLevel: body.accessLevel,
10281028+ createdBy: sa.issuer,
10291029+ expiresAt: body.expiresAt ?? null,
10301030+ maxUses: body.maxUses ?? null,
10311031+ note: body.note ?? null,
10321032+ });
10331033+10341034+ // Raw token only returned once. tokenHash is the stable ID for list/revoke.
10351035+ return c.json({
10361036+ token,
10371037+ tokenHash,
10381038+ spaceUri: row.spaceUri,
10391039+ accessLevel: row.accessLevel,
10401040+ expiresAt: row.expiresAt,
10411041+ maxUses: row.maxUses,
10421042+ createdAt: row.createdAt,
10431043+ });
10441044+ });
10451045+10461046+ app.get(`/xrpc/${NS}.invite.list`, auth, async (c) => {
10471047+ const sa = getAuth(c);
10481048+ const spaceUri = c.req.query("spaceUri");
10491049+ const includeRevoked = c.req.query("includeRevoked") === "true";
10501050+ if (!spaceUri) {
10511051+ return c.json({ error: "InvalidRequest", message: "spaceUri required" }, 400);
10521052+ }
10531053+ const space = await spaces.getSpace(spaceUri);
10541054+ if (!space) return c.json({ error: "NotFound" }, 404);
10551055+10561056+ const callerLevel = await resolveEffectiveLevel(community, spaceUri, sa.issuer);
10571057+ if (!callerLevel || rankOf(callerLevel) < rankOf("manager")) {
10581058+ return c.json({ error: "Forbidden", reason: "manager-required" }, 403);
10591059+ }
10601060+10611061+ const rows = await community.listInvites(spaceUri, { includeRevoked });
10621062+ return c.json({
10631063+ invites: rows.map((r) => ({
10641064+ tokenHash: r.tokenHash,
10651065+ spaceUri: r.spaceUri,
10661066+ accessLevel: r.accessLevel,
10671067+ createdBy: r.createdBy,
10681068+ createdAt: r.createdAt,
10691069+ expiresAt: r.expiresAt,
10701070+ maxUses: r.maxUses,
10711071+ usedCount: r.usedCount,
10721072+ revokedAt: r.revokedAt,
10731073+ note: r.note,
10741074+ })),
10751075+ });
10761076+ });
10771077+10781078+ app.post(`/xrpc/${NS}.invite.revoke`, auth, async (c) => {
10791079+ const sa = getAuth(c);
10801080+ const body = (await c.req.json().catch(() => null)) as
10811081+ | { tokenHash?: string }
10821082+ | null;
10831083+ if (!body?.tokenHash) {
10841084+ return c.json({ error: "InvalidRequest", message: "tokenHash required" }, 400);
10851085+ }
10861086+ const invite = await community.getInvite(body.tokenHash);
10871087+ if (!invite) return c.json({ error: "NotFound" }, 404);
10881088+10891089+ // Revoker: either the invite's creator, or manager+ on the target space.
10901090+ let allowed = invite.createdBy === sa.issuer;
10911091+ if (!allowed) {
10921092+ const level = await resolveEffectiveLevel(community, invite.spaceUri, sa.issuer);
10931093+ allowed = !!level && rankOf(level) >= rankOf("manager");
10941094+ }
10951095+ if (!allowed) {
10961096+ return c.json({ error: "Forbidden", reason: "creator-or-manager-required" }, 403);
10971097+ }
10981098+ await community.revokeInvite(body.tokenHash);
10991099+ return c.json({ ok: true });
11001100+ });
11011101+11021102+ app.post(`/xrpc/${NS}.invite.redeem`, auth, async (c) => {
11031103+ const sa = getAuth(c);
11041104+ const body = (await c.req.json().catch(() => null)) as { token?: string } | null;
11051105+ if (!body?.token) {
11061106+ return c.json({ error: "InvalidRequest", message: "token required" }, 400);
11071107+ }
11081108+ const tokenHash = await hashInviteToken(body.token);
11091109+11101110+ // Atomic consume — returns null if the token is expired/revoked/exhausted.
11111111+ const invite = await community.redeemInvite(tokenHash, Date.now());
11121112+ if (!invite) {
11131113+ return c.json({ error: "InvalidToken", reason: "token-invalid-or-exhausted" }, 410);
11141114+ }
11151115+11161116+ const space = await spaces.getSpace(invite.spaceUri);
11171117+ if (!space) {
11181118+ return c.json({ error: "NotFound", reason: "space-not-found" }, 404);
11191119+ }
11201120+11211121+ // The token itself is the authorization: the creator (who had manager+)
11221122+ // pre-signed "anyone with this token gets level X on this space". Grant
11231123+ // directly, attributing to the creator so audit trails make sense.
11241124+ await community.grant({
11251125+ spaceUri: invite.spaceUri,
11261126+ subjectDid: sa.issuer,
11271127+ accessLevel: invite.accessLevel,
11281128+ grantedBy: invite.createdBy,
11291129+ });
11301130+ await reconcile(community, spaces, invite.spaceUri, invite.createdBy);
11311131+11321132+ return c.json({
11331133+ spaceUri: invite.spaceUri,
11341134+ accessLevel: invite.accessLevel,
11351135+ communityDid: space.ownerDid,
11361136+ });
9731137 });
9741138}
9751139
+14
src/core/community/schema.ts
···3030 `CREATE INDEX IF NOT EXISTS idx_cal_subject ON community_access_levels(subject)`,
3131 `CREATE INDEX IF NOT EXISTS idx_cal_subject_space ON community_access_levels(subject)
3232 WHERE subject_kind = 'space'`,
3333+3434+ `CREATE TABLE IF NOT EXISTS community_invites (
3535+ token_hash TEXT PRIMARY KEY NOT NULL,
3636+ space_uri TEXT NOT NULL,
3737+ access_level TEXT NOT NULL,
3838+ created_by TEXT NOT NULL,
3939+ created_at ${dialect.bigintType} NOT NULL,
4040+ expires_at ${dialect.bigintType},
4141+ max_uses INTEGER,
4242+ used_count INTEGER NOT NULL DEFAULT 0,
4343+ revoked_at ${dialect.bigintType},
4444+ note TEXT
4545+ )`,
4646+ `CREATE INDEX IF NOT EXISTS idx_community_invites_space ON community_invites(space_uri, created_at DESC)`,
3347 ];
3448}
3549
+26
src/core/community/types.ts
···5959 grantedAt: number;
6060}
61616262+/** A pre-signed grant for a community-owned space. An admin/manager creates
6363+ * one; anyone with the raw token can redeem it once to get the specified
6464+ * access level. Storage keeps only the SHA-256 of the token. */
6565+export interface CommunityInviteRow {
6666+ tokenHash: string;
6767+ spaceUri: string;
6868+ accessLevel: AccessLevel;
6969+ createdBy: string;
7070+ createdAt: number;
7171+ expiresAt: number | null;
7272+ maxUses: number | null;
7373+ usedCount: number;
7474+ revokedAt: number | null;
7575+ note: string | null;
7676+}
7777+7878+export interface CreateCommunityInviteInput {
7979+ spaceUri: string;
8080+ tokenHash: string;
8181+ accessLevel: AccessLevel;
8282+ createdBy: string;
8383+ expiresAt: number | null;
8484+ maxUses: number | null;
8585+ note: string | null;
8686+}
8787+6288/** Key prefix for reserved community-owned spaces. */
6389export const RESERVED_KEYS = ["$admin", "$publishers"] as const;
6490export type ReservedKey = (typeof RESERVED_KEYS)[number];
+294-21
src/core/realtime/durable-object.ts
···1515 * the DO trusts anything that reaches it. */
16161717import type { PubSub, RealtimeEvent } from "./types";
1818+import { translateForQuery, type TranslatedEnvelope } from "./query-filter";
1919+type TranslatedEvent = TranslatedEnvelope;
2020+2121+/** Query spec attached to a WS subscriber, used to filter events before
2222+ * delivery. Shape matches what the Worker's `watchRecords` handler builds;
2323+ * forwarded to the DO via trusted internal headers on the WS upgrade. */
2424+export interface SubscriberQuerySpec {
2525+ /** NSID of the primary collection the client is watching. */
2626+ collection: string;
2727+ /** Space URI this subscription is scoped to. Events outside are dropped. */
2828+ spaceUri: string;
2929+ /** Hydrated relations. Keyed by relName — value is the child collection
3030+ * NSID and the field on the child record that references the parent. */
3131+ hydrate?: Record<string, { childCollection: string; matchField: string }>;
3232+}
18331934// ---- Minimal structural typings so we don't depend on @cloudflare/workers-types
2035// at the library level. Callers on Workers will have proper types.
···7287 return pullIterator(stub, signal);
7388 },
7489 };
9090+ }
9191+9292+ /** Forward an incoming browser WS upgrade (or SSE GET) through to the DO
9393+ * that owns this topic, attaching a query-filter spec that the DO will use
9494+ * to decide what to deliver. The Worker must verify auth + spec validity
9595+ * before calling this — the DO trusts the headers. */
9696+ async forwardSubscribe(
9797+ topic: string,
9898+ request: Request,
9999+ opts: {
100100+ did?: string;
101101+ querySpec?: SubscriberQuerySpec;
102102+ /** Unix ms. DO replays any buffered event with ts > sinceTs before
103103+ * going live — closes the snapshot→WS race window on the client. */
104104+ sinceTs?: number;
105105+ } = {}
106106+ ): Promise<Response> {
107107+ const headers = new Headers(request.headers);
108108+ if (opts.querySpec) {
109109+ headers.set("X-Contrail-Query-Spec", JSON.stringify(opts.querySpec));
110110+ }
111111+ const url = new URL("https://do/subscribe");
112112+ if (opts.did) url.searchParams.set("did", opts.did);
113113+ if (opts.sinceTs && opts.sinceTs > 0) {
114114+ url.searchParams.set("sinceTs", String(opts.sinceTs));
115115+ }
116116+ return this.stub(topic).fetch(url.toString(), {
117117+ method: "GET",
118118+ headers
119119+ });
75120 }
76121}
77122···150195 * This class intentionally avoids the `DurableObject` base class so we don't
151196 * have to depend on @cloudflare/workers-types at the library level — users
152197 * wire it up directly in their Worker entry. */
198198+/** Rolling buffer of recent events, used to close the snapshot→WS race:
199199+ * when a new subscriber connects with `?sinceTs=X`, replay any buffered
200200+ * event with `event.ts > X` before going live. Bounded by count + age so
201201+ * memory stays small. */
202202+const RECENT_BUFFER_MS = 15_000;
203203+const RECENT_BUFFER_MAX = 500;
204204+153205export class RealtimePubSubDO {
206206+ private readonly recentEvents: RealtimeEvent[] = [];
207207+154208 constructor(
155209 protected readonly state: DurableObjectState,
156210 _env?: unknown
157211 ) {}
212212+213213+ private pushRecent(event: RealtimeEvent): void {
214214+ this.recentEvents.push(event);
215215+ const cutoff = Date.now() - RECENT_BUFFER_MS;
216216+ while (
217217+ this.recentEvents.length > RECENT_BUFFER_MAX ||
218218+ (this.recentEvents.length > 0 && this.recentEvents[0]!.ts < cutoff)
219219+ ) {
220220+ this.recentEvents.shift();
221221+ }
222222+ }
158223159224 /** Worker entry delegates `fetch` to this method. */
160225 async fetch(request: Request): Promise<Response> {
···171236 }
172237 if (request.method === "GET" && url.pathname === "/subscribe") {
173238 const did = url.searchParams.get("did") ?? undefined;
239239+ const sinceTsRaw = url.searchParams.get("sinceTs");
240240+ const sinceTs = sinceTsRaw ? Number(sinceTsRaw) : 0;
241241+ // Optional query-filter spec, forwarded by the Worker after it has
242242+ // verified the caller's auth + access. Parsed once here; the parsed
243243+ // object is serialized into the WS attachment so the DO can filter
244244+ // events on publish without re-parsing.
245245+ let querySpec: SubscriberQuerySpec | undefined;
246246+ const rawSpec = request.headers.get("X-Contrail-Query-Spec");
247247+ if (rawSpec) {
248248+ try {
249249+ querySpec = JSON.parse(rawSpec) as SubscriberQuerySpec;
250250+ } catch {
251251+ return new Response(
252252+ JSON.stringify({ error: "InvalidRequest", message: "bad X-Contrail-Query-Spec" }),
253253+ { status: 400 }
254254+ );
255255+ }
256256+ }
257257+174258 if (request.headers.get("Upgrade")?.toLowerCase() === "websocket") {
175259 const Pair = (globalThis as unknown as { WebSocketPair?: any }).WebSocketPair;
176260 if (!Pair) return new Response("websockets require Workers", { status: 426 });
177261 const pair = new Pair();
178178- this.acceptWebSocketSubscriber(pair[1], did);
262262+ this.acceptWebSocketSubscriber(pair[1], did, querySpec);
263263+ if (sinceTs > 0) this.replayRecentTo(pair[1], sinceTs);
179264 return new Response(null, {
180265 status: 101,
181181- // @ts-expect-error Workers-specific init field
266266+ // Workers-specific init field
182267 webSocket: pair[0],
183183- });
268268+ } as ResponseInit & { webSocket: unknown });
184269 }
185185- return this.openSseResponse(did);
270270+ return this.openSseResponse(did, querySpec, sinceTs);
186271 }
187272 return new Response("not found", { status: 404 });
188273 }
189274190275 /** Fan-out an event to every connected subscriber (WS + SSE).
191191- * Public so tests + advanced callers can skip the HTTP layer. */
276276+ * Public so tests + advanced callers can skip the HTTP layer.
277277+ *
278278+ * If a subscriber has attached a `querySpec`, we translate the raw event
279279+ * into 0–1 watchRecords-shaped events (record.created, record.deleted,
280280+ * hydration.added, hydration.removed) and deliver only those. Otherwise
281281+ * the raw event is delivered as-is (topic-firehose behaviour for the
282282+ * `realtime.subscribe` endpoint). */
192283 publishEvent(event: RealtimeEvent): void {
193193- const payload = JSON.stringify(event);
194194- const frame = `event: ${event.kind}\ndata: ${payload}\n\n`;
284284+ // Buffer first so a subscriber connecting mid-publish (race-window
285285+ // replay) can pick up this event too once they provide their sinceTs.
286286+ this.pushRecent(event);
287287+288288+ const rawPayload = JSON.stringify(event);
289289+ const rawFrame = `event: ${event.kind}\ndata: ${rawPayload}\n\n`;
195290196291 for (const ws of this.state.getWebSockets()) {
197197- try {
198198- ws.send(payload);
199199- } catch {
200200- /* ignore — socket may be closing */
201201- }
202292 const attachment = getAttachment(ws);
293293+294294+ if (attachment?.querySpec) {
295295+ const translated = translateForQuery(event, attachment);
296296+ if (translated) this.writeSubscriberState(ws, attachment, translated);
297297+ for (const msg of translated ?? []) {
298298+ try {
299299+ ws.send(JSON.stringify(msg));
300300+ } catch {
301301+ /* ignore */
302302+ }
303303+ }
304304+ } else {
305305+ try {
306306+ ws.send(rawPayload);
307307+ } catch {
308308+ /* ignore */
309309+ }
310310+ }
311311+203312 if (
204313 event.kind === "member.removed" &&
205314 attachment?.did &&
···212321 }
213322 }
214323 }
324324+215325 for (const entry of this.sseControllers) {
216216- try {
217217- entry.controller.enqueue(this.encoder.encode(frame));
218218- } catch {
219219- /* drop; cleanup happens on the subscribe-side */
326326+ if (entry.querySpec) {
327327+ const translated = translateForQuery(event, entry);
328328+ if (translated) this.writeSubscriberStateForSse(entry, translated);
329329+ for (const msg of translated ?? []) {
330330+ try {
331331+ entry.controller.enqueue(
332332+ this.encoder.encode(`event: ${msg.kind}\ndata: ${JSON.stringify(msg.data)}\n\n`)
333333+ );
334334+ } catch {
335335+ /* drop */
336336+ }
337337+ }
338338+ } else {
339339+ try {
340340+ entry.controller.enqueue(this.encoder.encode(rawFrame));
341341+ } catch {
342342+ /* drop; cleanup happens on the subscribe-side */
343343+ }
220344 }
345345+221346 if (
222347 event.kind === "member.removed" &&
223348 entry.did &&
···232357 }
233358 }
234359235235- /** Register a server-side WebSocket as a subscriber. Wires the DID attachment. */
236236- acceptWebSocketSubscriber(serverWs: any, did?: string): void {
360360+ /** Register a server-side WebSocket as a subscriber. Wires the DID +
361361+ * optional query spec into the hibernation attachment so this DO can
362362+ * filter and route events after going to sleep. */
363363+ acceptWebSocketSubscriber(
364364+ serverWs: any,
365365+ did?: string,
366366+ querySpec?: SubscriberQuerySpec
367367+ ): void {
237368 this.state.acceptWebSocket(serverWs, did ? [did] : undefined);
238238- if (did) setAttachment(serverWs, { did });
369369+ if (did || querySpec) {
370370+ setAttachment(serverWs, {
371371+ did,
372372+ querySpec,
373373+ parentUris: [],
374374+ childToParent: {}
375375+ });
376376+ }
239377 }
240378241379 /** Open an SSE subscriber; returns the streaming Response. */
242242- openSseResponse(did?: string): Response {
380380+ openSseResponse(
381381+ did?: string,
382382+ querySpec?: SubscriberQuerySpec,
383383+ sinceTs = 0
384384+ ): Response {
243385 let entry: SseEntry;
244386 const stream = new ReadableStream<Uint8Array>({
245387 start: (controller) => {
246246- entry = { controller, did };
388388+ entry = {
389389+ controller,
390390+ did,
391391+ querySpec,
392392+ parentUris: new Set(),
393393+ childToParent: new Map()
394394+ };
247395 this.sseControllers.add(entry);
248396 controller.enqueue(this.encoder.encode(`: open\n\n`));
397397+ if (sinceTs > 0) this.replayRecentToSse(entry, sinceTs);
249398 },
250399 cancel: () => {
251400 this.sseControllers.delete(entry);
···261410 });
262411 }
263412413413+ /** Replay buffered events with ts > sinceTs through this subscriber's
414414+ * query-spec filter. Called once, synchronously, on WS connect. */
415415+ private replayRecentTo(ws: any, sinceTs: number): void {
416416+ const attachment = getAttachment(ws);
417417+ for (const event of this.recentEvents) {
418418+ if (event.ts <= sinceTs) continue;
419419+ if (attachment?.querySpec) {
420420+ const translated = translateForQuery(event, attachment);
421421+ if (translated) this.writeSubscriberState(ws, attachment, translated);
422422+ for (const msg of translated ?? []) {
423423+ try {
424424+ ws.send(JSON.stringify(msg));
425425+ } catch {
426426+ /* ignore */
427427+ }
428428+ }
429429+ } else {
430430+ try {
431431+ ws.send(JSON.stringify(event));
432432+ } catch {
433433+ /* ignore */
434434+ }
435435+ }
436436+ }
437437+ }
438438+439439+ private replayRecentToSse(entry: SseEntry, sinceTs: number): void {
440440+ for (const event of this.recentEvents) {
441441+ if (event.ts <= sinceTs) continue;
442442+ if (entry.querySpec) {
443443+ const translated = translateForQuery(event, entry);
444444+ if (translated) this.writeSubscriberStateForSse(entry, translated);
445445+ for (const msg of translated ?? []) {
446446+ try {
447447+ entry.controller.enqueue(
448448+ this.encoder.encode(`event: ${msg.kind}\ndata: ${JSON.stringify(msg.data)}\n\n`)
449449+ );
450450+ } catch {
451451+ /* drop */
452452+ }
453453+ }
454454+ } else {
455455+ try {
456456+ entry.controller.enqueue(
457457+ this.encoder.encode(`event: ${event.kind}\ndata: ${JSON.stringify(event)}\n\n`)
458458+ );
459459+ } catch {
460460+ /* drop */
461461+ }
462462+ }
463463+ }
464464+ }
465465+466466+ /** Update the persisted WS attachment state after we've decided which
467467+ * events to forward. Keeps the parent/child tracking tables warm across
468468+ * hibernation. */
469469+ private writeSubscriberState(
470470+ ws: any,
471471+ attachment: WsAttachment,
472472+ translated: TranslatedEvent[]
473473+ ): void {
474474+ let dirty = false;
475475+ for (const msg of translated) {
476476+ if (msg.kind === "record.created" && msg.data.record?.uri) {
477477+ attachment.parentUris = Array.from(
478478+ new Set([...(attachment.parentUris ?? []), msg.data.record.uri])
479479+ );
480480+ dirty = true;
481481+ } else if (msg.kind === "record.deleted" && msg.data.uri) {
482482+ const before = attachment.parentUris ?? [];
483483+ attachment.parentUris = before.filter((u) => u !== msg.data.uri);
484484+ if (attachment.parentUris.length !== before.length) dirty = true;
485485+ } else if (msg.kind === "hydration.added" && msg.data.child?.rkey) {
486486+ attachment.childToParent = {
487487+ ...(attachment.childToParent ?? {}),
488488+ [msg.data.child.rkey]: {
489489+ parentUri: msg.data.parentUri,
490490+ relName: msg.data.relation
491491+ }
492492+ };
493493+ dirty = true;
494494+ } else if (msg.kind === "hydration.removed" && msg.data.childRkey) {
495495+ const next = { ...(attachment.childToParent ?? {}) };
496496+ if (next[msg.data.childRkey]) {
497497+ delete next[msg.data.childRkey];
498498+ attachment.childToParent = next;
499499+ dirty = true;
500500+ }
501501+ }
502502+ }
503503+ if (dirty) setAttachment(ws, attachment);
504504+ }
505505+506506+ private writeSubscriberStateForSse(
507507+ entry: SseEntry,
508508+ translated: TranslatedEvent[]
509509+ ): void {
510510+ for (const msg of translated) {
511511+ if (msg.kind === "record.created" && msg.data.record?.uri) {
512512+ entry.parentUris?.add(msg.data.record.uri);
513513+ } else if (msg.kind === "record.deleted" && msg.data.uri) {
514514+ entry.parentUris?.delete(msg.data.uri);
515515+ } else if (msg.kind === "hydration.added" && msg.data.child?.rkey) {
516516+ entry.childToParent?.set(msg.data.child.rkey, {
517517+ parentUri: msg.data.parentUri,
518518+ relName: msg.data.relation
519519+ });
520520+ } else if (msg.kind === "hydration.removed" && msg.data.childRkey) {
521521+ entry.childToParent?.delete(msg.data.childRkey);
522522+ }
523523+ }
524524+ }
525525+264526 private readonly sseControllers = new Set<SseEntry>();
265527 private readonly encoder = new TextEncoder();
266528}
···268530interface SseEntry {
269531 controller: ReadableStreamDefaultController<Uint8Array>;
270532 did: string | undefined;
533533+ querySpec?: SubscriberQuerySpec;
534534+ parentUris?: Set<string>;
535535+ childToParent?: Map<string, { parentUri: string; relName: string }>;
271536}
272537273538interface WsAttachment {
274539 did?: string;
540540+ querySpec?: SubscriberQuerySpec;
541541+ /** URIs of primary records currently in this subscriber's result set. */
542542+ parentUris?: string[];
543543+ /** childRkey → parent info, for routing child delete events. */
544544+ childToParent?: Record<string, { parentUri: string; relName: string }>;
275545}
276546277547function setAttachment(ws: any, attachment: WsAttachment): void {
···292562 }
293563 return (ws.__attachment as WsAttachment | undefined) ?? null;
294564}
565565+566566+// Query-spec filtering lives in ./query-filter so the Worker can reuse it for
567567+// non-DO (InMemoryPubSub) watchRecords paths without bundling the whole DO.
···43434444/** Hono middleware that verifies the Authorization: Bearer <JWT> as an atproto
4545 * service-auth token. On success, attaches the decoded claims to c.var.serviceAuth.
4646- * Expected Nsid method is taken from the route pattern (last segment after /xrpc/). */
4646+ * Expected Nsid method is taken from the route pattern (last segment after /xrpc/).
4747+ *
4848+ * If `authOverride` is provided and returns non-null claims, the JWT check is
4949+ * skipped and the override's claims are attached instead — used to paper over
5050+ * bsky.social's loopback-client restriction on `getServiceAuth` during dev.
5151+ * The override must be gated in the caller (e.g. an env flag) since it
5252+ * bypasses cryptographic verification. */
4753export function createServiceAuthMiddleware(
4848- verifier: ServiceJwtVerifier
5454+ verifier: ServiceJwtVerifier,
5555+ options: {
5656+ authOverride?: (
5757+ req: Request
5858+ ) => Promise<import("./types").AuthOverrideResult | null> | import("./types").AuthOverrideResult | null;
5959+ } = {}
4960): MiddlewareHandler {
5061 return async (c, next) => {
6262+ const lxm = extractLxmFromPath(c);
6363+6464+ if (options.authOverride) {
6565+ const override = await options.authOverride(c.req.raw);
6666+ if (override) {
6767+ c.set("serviceAuth", {
6868+ issuer: override.issuer,
6969+ audience: override.audience ?? "",
7070+ lxm: override.lxm ?? lxm ?? undefined,
7171+ clientId: override.clientId,
7272+ } satisfies ServiceAuth);
7373+ await next();
7474+ return;
7575+ }
7676+ }
7777+5178 const header = c.req.header("Authorization");
5279 if (!header || !header.startsWith("Bearer ")) {
5380 return c.json({ error: "AuthRequired", message: "Missing bearer token" }, 401);
5481 }
5582 const token = header.slice(7).trim();
5656- const lxm = extractLxmFromPath(c);
57835884 const result = await verifier.verify(token, { lxm });
5985 if (!result.ok) {
···8511186112/** Verify a request's Authorization: Bearer token out-of-band (e.g. from a
87113 * route handler that doesn't always require auth). Returns the claims on
8888- * success, or null if missing/invalid. */
114114+ * success, or null if missing/invalid.
115115+ *
116116+ * If `authOverride` is provided and returns claims, the JWT path is skipped
117117+ * and those claims are returned — same bypass as the middleware-level auth. */
89118export async function verifyServiceAuthRequest(
90119 verifier: ServiceJwtVerifier,
91120 request: Request,
9292- lxm?: Nsid | null
121121+ lxm?: Nsid | null,
122122+ options: {
123123+ authOverride?: (
124124+ req: Request
125125+ ) => Promise<import("./types").AuthOverrideResult | null> | import("./types").AuthOverrideResult | null;
126126+ } = {}
93127): Promise<ServiceAuth | null> {
128128+ if (options.authOverride) {
129129+ const override = await options.authOverride(request);
130130+ if (override) {
131131+ return {
132132+ issuer: override.issuer,
133133+ audience: override.audience ?? "",
134134+ lxm: override.lxm ?? lxm ?? undefined,
135135+ clientId: override.clientId,
136136+ };
137137+ }
138138+ }
139139+94140 const header = request.headers.get("Authorization");
95141 if (!header || !header.startsWith("Bearer ")) return null;
96142 const token = header.slice(7).trim();
+4-1
src/core/spaces/router.ts
···4646 const adapter = options.adapter ?? ctx?.adapter ?? new HostedAdapter(db, config);
4747 const verifier = ctx?.verifier ?? buildVerifier(spacesConfig);
4848 const auth =
4949- options.authMiddleware ?? createServiceAuthMiddleware(verifier);
4949+ options.authMiddleware ??
5050+ createServiceAuthMiddleware(verifier, {
5151+ authOverride: spacesConfig.authOverride,
5252+ });
50535154 /** Read-route auth: skip the JWT middleware when an `?inviteToken=` is
5255 * present so anonymous bearer reads don't 401 before the route handler can
+18
src/core/spaces/types.ts
···3737 resolver?: DidDocumentResolver;
3838 /** Blob-upload backend. When omitted, blob XRPCs are not exposed. */
3939 blobs?: SpacesBlobsConfig;
4040+ /** Dev-only auth bypass. Runs BEFORE JWT verification. If it returns
4141+ * non-null claims, those are used as the authenticated caller and the
4242+ * JWT check is skipped. Intended for local development where bsky.social
4343+ * rejects `getServiceAuth` for loopback OAuth clients; pair with a
4444+ * signed-cookie session check so only your own browser can trigger it.
4545+ * Never set in production. */
4646+ authOverride?: (
4747+ req: Request
4848+ ) => Promise<AuthOverrideResult | null> | AuthOverrideResult | null;
4949+}
5050+5151+/** Claims returned by an `authOverride`. Matches the ServiceAuth shape the
5252+ * normal JWT path attaches to the request. */
5353+export interface AuthOverrideResult {
5454+ issuer: string;
5555+ audience?: string;
5656+ lxm?: string;
5757+ clientId?: string;
4058}
41594260export interface SpaceRow {
···11import { dev } from '$app/environment';
22import { scope } from '@atcute/oauth-node-client';
33+import { xrpcMethods } from './generated-methods';
3444-// writable collections
55-export const collections = ['xyz.statusphere.status', 'social.atmo.test.blob'] as const;
55+// Writable collections on the USER's PDS. This app writes nothing to the user's
66+// PDS — every record lives in contrail spaces, authored by the user DID via
77+// space.putRecord or by the community DID via community.space.putRecord.
88+// Keep the list empty; repo scope is omitted below.
99+export const collections = [] as const;
610711export type AllowedCollection = (typeof collections)[number];
81299-// OAuth scope — add scope.blob(), scope.rpc(), etc. as needed
1313+// OAuth scope.
1414+//
1515+// We ask for:
1616+// - `atproto` — baseline identity scope.
1717+// - `rpc` for every XRPC method contrail emits in this namespace. The list
1818+// is auto-generated by `pnpm generate` (source: contrail's permissionSet),
1919+// so adding/removing collections or toggling modules automatically updates
2020+// the consent prompt.
2121+// - `aud: '*'` — we don't couple the scope to a specific service DID so one
2222+// OAuth client metadata works in dev and prod. The PDS still only mints
2323+// JWTs for the exact `aud` the app asks for at getServiceAuth time.
1024export const scopes = [
1125 'atproto',
1212- scope.repo({ collection: [...collections] }),
1313- scope.blob({ accept: ['image/*'] })
2626+ scope.rpc({
2727+ aud: '*',
2828+ lxm: [...xrpcMethods] as `${string}.${string}.${string}`[]
2929+ })
1430];
15311632// set to false to disable signup
···11+/** Per-space unread tracker.
22+ * - `counts[spaceUri]` is the number of record.created events received since
33+ * the space was last opened.
44+ * - `lastRead[spaceUri]` is the iso timestamp of the last "open" event,
55+ * persisted to localStorage so cold loads can compare against existing
66+ * messages' createdAt to derive initial dot state.
77+ *
88+ * Reset counts call on channel open via `markLastRead()` / `resetUnread()`. */
99+1010+import { untrack } from 'svelte';
1111+1212+const LS_KEY = 'rooms:lastRead:v1';
1313+1414+function loadLastRead(): Record<string, string> {
1515+ if (typeof localStorage === 'undefined') return {};
1616+ try {
1717+ const raw = localStorage.getItem(LS_KEY);
1818+ if (!raw) return {};
1919+ const parsed = JSON.parse(raw);
2020+ return typeof parsed === 'object' && parsed ? parsed : {};
2121+ } catch {
2222+ return {};
2323+ }
2424+}
2525+2626+function persistLastRead(v: Record<string, string>) {
2727+ if (typeof localStorage === 'undefined') return;
2828+ try {
2929+ localStorage.setItem(LS_KEY, JSON.stringify(v));
3030+ } catch {
3131+ // ignore quota errors
3232+ }
3333+}
3434+3535+export const unread = $state<{
3636+ counts: Record<string, number>;
3737+ lastRead: Record<string, string>;
3838+}>({
3939+ counts: {},
4040+ lastRead: loadLastRead()
4141+});
4242+4343+/** Called when a record.created event for a space arrives while the space is
4444+ * not the active one. Increments the unread counter.
4545+ *
4646+ * Whole body is wrapped in `untrack` so that a caller running inside a
4747+ * `$effect` does NOT subscribe to any of the `unread.*` state we read here.
4848+ * Writes still notify real dependents — we just stop the current effect from
4949+ * subscribing to state it's about to mutate and re-reading it afterwards. */
5050+export function bumpUnread(spaceUri: string, createdAt?: string) {
5151+ untrack(() => {
5252+ const last = unread.lastRead[spaceUri];
5353+ if (createdAt && last && createdAt <= last) return;
5454+ unread.counts[spaceUri] = (unread.counts[spaceUri] ?? 0) + 1;
5555+ });
5656+}
5757+5858+/** Called when a channel view mounts or receives new messages while focused. */
5959+export function markLastRead(spaceUri: string) {
6060+ untrack(() => {
6161+ const next = { ...unread.lastRead, [spaceUri]: new Date().toISOString() };
6262+ unread.counts[spaceUri] = 0;
6363+ unread.lastRead = next;
6464+ persistLastRead(next);
6565+ });
6666+}
6767+6868+export function resetUnread(spaceUri: string) {
6969+ untrack(() => {
7070+ if (unread.counts[spaceUri]) unread.counts[spaceUri] = 0;
7171+ });
7272+}