[READ-ONLY] Mirror of https://github.com/flo-bit/contrail. atproto backend in a bottle
flo-bit.dev/contrail/
sync stuff
+839
-153
+5
-1
examples/sveltekit-group-chat/src/lib/rooms/rooms.remote.ts
+5
-1
examples/sveltekit-group-chat/src/lib/rooms/rooms.remote.ts
···
162
162
const PostMessageInput = v.object({
163
163
spaceUri: v.pipe(v.string(), v.minLength(1)),
164
164
text: v.pipe(v.string(), v.minLength(1), v.maxLength(4000)),
165
-
replyTo: v.optional(v.pipe(v.string(), v.minLength(1)))
165
+
replyTo: v.optional(v.pipe(v.string(), v.minLength(1))),
166
+
/** Optional client-generated rkey — lets the browser match an optimistic
167
+
* record to the stream's `record.created` event by identity. */
168
+
rkey: v.optional(v.pipe(v.string(), v.minLength(1), v.maxLength(64)))
166
169
});
167
170
168
171
export const postMessage = command(PostMessageInput, async (input) => {
···
170
173
const res = await spacePutRecord(ctx, {
171
174
spaceUri: input.spaceUri,
172
175
collection: 'tools.atmo.chat.message',
176
+
...(input.rkey ? { rkey: input.rkey } : {}),
173
177
record: {
174
178
text: input.text,
175
179
createdAt: new Date().toISOString(),
+11
-5
examples/sveltekit-group-chat/src/lib/rooms/uri.ts
+11
-5
examples/sveltekit-group-chat/src/lib/rooms/uri.ts
···
1
-
/** Helpers for assembling space URIs in the `tools.atmo.chat` namespace. */
1
+
/** Helpers for assembling space URIs in the `tools.atmo.chat` namespace.
2
+
*
3
+
* The URIs are returned as `ResourceUri` (branded) so they flow into
4
+
* lexicon-typed XRPC params without per-call casts. We apply the brand
5
+
* once here — structurally these are just `at://…` strings. */
6
+
7
+
import type { ResourceUri } from '@atcute/lexicons';
2
8
3
9
const SPACE_TYPE = 'tools.atmo.chat.space';
4
10
5
-
export function buildSpaceUri(communityDid: string, key: string): string {
6
-
return `at://${communityDid}/${SPACE_TYPE}/${key}`;
11
+
export function buildSpaceUri(communityDid: string, key: string): ResourceUri {
12
+
return `at://${communityDid}/${SPACE_TYPE}/${key}` as ResourceUri;
7
13
}
8
14
9
-
export function buildMembersUri(communityDid: string): string {
15
+
export function buildMembersUri(communityDid: string): ResourceUri {
10
16
return buildSpaceUri(communityDid, 'members');
11
17
}
12
18
13
-
export function buildAdminUri(communityDid: string): string {
19
+
export function buildAdminUri(communityDid: string): ResourceUri {
14
20
return buildSpaceUri(communityDid, '$admin');
15
21
}
16
22
+274
-44
examples/sveltekit-group-chat/src/lib/rooms/watch.svelte.ts
+274
-44
examples/sveltekit-group-chat/src/lib/rooms/watch.svelte.ts
···
1
-
/** Svelte 5 binding over `@atmo-dev/contrail/sync`. Turns the framework-agnostic
2
-
* subscribable store into a reactive `$state` object that Svelte effects can
3
-
* observe in the usual way.
1
+
/** Svelte 5 binding over `@atmo-dev/contrail/sync`.
4
2
*
5
3
* Usage:
6
-
* const query = createWatchQuery({
7
-
* url: `/xrpc/tools.atmo.chat.message.watchRecords?spaceUri=${encodeURIComponent(spaceUri)}`,
8
-
* });
9
-
* // template: {#each query.records as r (r.uri)} ...
10
-
* // teardown: $effect(() => () => query.stop());
4
+
* const messagesQuery = $derived(
5
+
* createWatchQuery({
6
+
* endpoint: 'tools.atmo.chat.message',
7
+
* params: { spaceUri, limit: 50 }
8
+
* })
9
+
* );
10
+
* // template: {#each messagesQuery.records as r (r.rkey)} {r.record.text}
11
+
*
12
+
* Records and params are typed via the app's generated `XRPCQueries` /
13
+
* `Records` ambient declarations (from `src/lexicon-types/`). Reading
14
+
* `.records` auto-subscribes; when no component is reading anymore
15
+
* (e.g. a $derived query instance is replaced on prop change), the
16
+
* underlying stream is torn down via `createSubscriber`.
17
+
*
18
+
* Authentication: tickets are minted on demand. By default the wrapper
19
+
* calls the `mintWatchTicketCmd` remote function; override with
20
+
* `setContrailTicketMinter(fn)` if your app uses a different route.
11
21
*/
12
22
13
-
import { createWatchStore, type WatchRecord, type WatchStoreOptions, type WatchStoreStatus } from '@atmo-dev/contrail/sync';
23
+
import { createSubscriber } from 'svelte/reactivity';
24
+
import { browser } from '$app/environment';
25
+
import {
26
+
createWatchStore,
27
+
type WatchCache,
28
+
type WatchRecord,
29
+
type WatchStore,
30
+
type WatchStoreStatus
31
+
} from '@atmo-dev/contrail/sync';
32
+
import { createIndexedDBCache } from '@atmo-dev/contrail/sync/cache-idb';
33
+
import type { InferInput } from '@atcute/lexicons';
34
+
import type { BaseSchema } from '@atcute/lexicons/validations';
35
+
import type { Records, XRPCQueries } from '@atcute/lexicons/ambient';
36
+
import { dev } from '$app/environment';
37
+
import { mintWatchTicketCmd } from './rooms.remote';
38
+
39
+
// ---------------------------------------------------------------------------
40
+
// Type machinery — derives params + record types from the app's lexicons.
41
+
// ---------------------------------------------------------------------------
42
+
43
+
type WatchRecordsOf<K extends string> = `${K}.watchRecords`;
44
+
45
+
/** Params type for `<endpoint>.watchRecords`, from ambient XRPCQueries. */
46
+
type WatchParamsOf<K extends string> =
47
+
WatchRecordsOf<K> extends keyof XRPCQueries
48
+
? XRPCQueries[WatchRecordsOf<K>] extends { params: infer P }
49
+
? P extends BaseSchema<unknown, unknown>
50
+
? InferInput<P>
51
+
: Record<string, unknown>
52
+
: Record<string, unknown>
53
+
: Record<string, unknown>;
54
+
55
+
/** Record value-shape for a collection, from ambient Records. */
56
+
export type RecordShapeOf<K extends string> = K extends keyof Records
57
+
? Records[K] extends BaseSchema<unknown, unknown>
58
+
? InferInput<Records[K]>
59
+
: Record<string, unknown>
60
+
: Record<string, unknown>;
61
+
62
+
/** WatchRecord with a typed `record` payload. Mirrors WatchRecord's explicit
63
+
* fields but without the `[k: string]: unknown` catchall (which would
64
+
* poison property access under `Omit`). */
65
+
export interface TypedWatchRecord<R> {
66
+
uri: string;
67
+
did: string;
68
+
rkey: string;
69
+
collection: string;
70
+
record: R;
71
+
time_us?: number;
72
+
indexed_at?: number;
73
+
cid?: string | null;
74
+
_space?: string;
75
+
/** Present on optimistic entries added via `addOptimistic`. */
76
+
optimistic?: 'pending' | 'failed';
77
+
/** Error attached via `markFailed`. */
78
+
optimisticError?: Error;
79
+
}
14
80
15
-
export interface WatchQuery {
16
-
readonly records: ReadonlyArray<WatchRecord>;
17
-
readonly status: WatchStoreStatus;
18
-
readonly error: Error | null;
19
-
stop(): void;
81
+
// ---------------------------------------------------------------------------
82
+
// Configurable ticket minter — called when the wrapper needs a fresh ticket.
83
+
// Default: uses `mintWatchTicketCmd`. Override via `setContrailTicketMinter`.
84
+
// ---------------------------------------------------------------------------
85
+
86
+
export interface TicketMintContext {
87
+
endpoint: string;
88
+
params: Record<string, unknown>;
20
89
}
21
90
22
-
export function createWatchQuery(options: WatchStoreOptions): WatchQuery {
23
-
const state = $state<{
24
-
records: ReadonlyArray<WatchRecord>;
25
-
status: WatchStoreStatus;
26
-
error: Error | null;
27
-
}>({
28
-
records: [],
29
-
status: 'idle',
30
-
error: null
91
+
export type TicketMinter = (ctx: TicketMintContext) => Promise<string>;
92
+
93
+
let ticketMinter: TicketMinter | null = async ({ endpoint, params }) => {
94
+
const res = await mintWatchTicketCmd({
95
+
watchRecordsNsid: `${endpoint}.watchRecords`,
96
+
spaceUri: String(params.spaceUri ?? ''),
97
+
limit: typeof params.limit === 'number' ? params.limit : 50
31
98
});
99
+
return res.ticket;
100
+
};
32
101
33
-
const store = createWatchStore(options);
34
-
const unsub = store.subscribe((s) => {
35
-
state.records = s.records;
36
-
state.status = s.status;
37
-
state.error = s.error;
38
-
});
39
-
store.start();
102
+
export function setContrailTicketMinter(fn: TicketMinter | null): void {
103
+
ticketMinter = fn;
104
+
}
105
+
106
+
// ---------------------------------------------------------------------------
107
+
// Configurable cache — persists last-seen records for instant first paint.
108
+
// Default: IndexedDB in the browser, no-op during SSR. Override via
109
+
// `setContrailCache(cache)` or opt out with `setContrailCache(null)`.
110
+
// ---------------------------------------------------------------------------
111
+
112
+
let cache: WatchCache | null = browser ? createIndexedDBCache() : null;
113
+
114
+
export function setContrailCache(next: WatchCache | null): void {
115
+
cache = next;
116
+
}
117
+
118
+
// ---------------------------------------------------------------------------
119
+
// createWatchQuery
120
+
// ---------------------------------------------------------------------------
40
121
41
-
return {
42
-
get records() {
43
-
return state.records;
44
-
},
45
-
get status() {
46
-
return state.status;
47
-
},
48
-
get error() {
49
-
return state.error;
50
-
},
51
-
stop() {
52
-
unsub();
53
-
store.stop();
122
+
export interface WatchQueryOptions<K extends string> {
123
+
/** Collection NSID. The library appends `.watchRecords` to build the URL. */
124
+
endpoint: K;
125
+
/** Query params for the watchRecords endpoint. Typed per-endpoint. */
126
+
params?: WatchParamsOf<K>;
127
+
/** Optional pre-minted ticket (typically from SSR `+page.server.ts`).
128
+
* Used once on the first connect; reconnects fall through to the
129
+
* configured ticket minter. */
130
+
initialTicket?: string;
131
+
/** Per-query override for ticket minting. If omitted, uses the minter
132
+
* configured via `setContrailTicketMinter` (or the default). */
133
+
mintTicket?: string | (() => Promise<string>);
134
+
/** Transport. Default: 'ws' in prod, 'sse' in dev (Vite dev's DO wiring
135
+
* is limited — see `todo/realtime-do-dev-wiring.md`). */
136
+
transport?: 'sse' | 'ws';
137
+
/** Sort comparator. Default: newest first by `time_us`. */
138
+
compare?: (
139
+
a: TypedWatchRecord<RecordShapeOf<K>>,
140
+
b: TypedWatchRecord<RecordShapeOf<K>>
141
+
) => number;
142
+
/** Auto-reconnect on error. Default: true. */
143
+
reconnect?: boolean;
144
+
/** Per-query cache override. If omitted, uses the wrapper's configured
145
+
* cache (see `setContrailCache`). Pass `null` to disable caching for
146
+
* this query specifically. */
147
+
cache?: WatchCache | null;
148
+
/** Custom cache key. Defaults to the built watchRecords URL. */
149
+
cacheKey?: string;
150
+
/** Max records retained in cache. Default: 200. */
151
+
cacheMaxRecords?: number;
152
+
}
153
+
154
+
export class WatchQuery<K extends string> {
155
+
#records = $state<readonly TypedWatchRecord<RecordShapeOf<K>>[]>([]);
156
+
#status = $state<WatchStoreStatus>('idle');
157
+
#error = $state<Error | null>(null);
158
+
#subscribe: () => void;
159
+
#store: WatchStore;
160
+
#collection: K;
161
+
162
+
constructor(opts: WatchQueryOptions<K>) {
163
+
const endpoint = opts.endpoint;
164
+
const params = (opts.params ?? {}) as Record<string, unknown>;
165
+
const url = buildWatchUrl(endpoint, params);
166
+
this.#collection = endpoint;
167
+
168
+
// Ticket source resolution: initialTicket (one-shot) → per-call mintTicket →
169
+
// module-level default. Undefined if none available (public endpoints).
170
+
let pendingInitial = opts.initialTicket;
171
+
const hasSource = !!(opts.initialTicket || opts.mintTicket || ticketMinter);
172
+
173
+
const mintTicket: string | (() => Promise<string>) | undefined = hasSource
174
+
? async () => {
175
+
if (pendingInitial) {
176
+
const t = pendingInitial;
177
+
pendingInitial = undefined;
178
+
return t;
179
+
}
180
+
if (typeof opts.mintTicket === 'string') return opts.mintTicket;
181
+
if (typeof opts.mintTicket === 'function') return opts.mintTicket();
182
+
if (ticketMinter) return ticketMinter({ endpoint, params });
183
+
throw new Error('createWatchQuery: no ticket source available');
184
+
}
185
+
: undefined;
186
+
187
+
// Store is created eagerly so optimistic mutations (add/markFailed/remove)
188
+
// can target it from non-reactive contexts (e.g. a submit handler) even
189
+
// before any template reader has activated the subscriber. The actual
190
+
// network connection only opens when `createSubscriber` fires `start()`
191
+
// on first read, and closes when no readers remain.
192
+
// Cache: explicit override (including `null`) wins over the module-level
193
+
// default. `opts.cache === undefined` means "use whatever's configured."
194
+
const effectiveCache = opts.cache === undefined ? cache : opts.cache;
195
+
196
+
this.#store = createWatchStore({
197
+
url,
198
+
transport: opts.transport ?? (dev ? 'sse' : 'ws'),
199
+
reconnect: opts.reconnect,
200
+
mintTicket,
201
+
compareRecords: opts.compare as
202
+
| ((a: WatchRecord, b: WatchRecord) => number)
203
+
| undefined,
204
+
cache: effectiveCache ?? undefined,
205
+
cacheKey: opts.cacheKey,
206
+
cacheMaxRecords: opts.cacheMaxRecords
207
+
});
208
+
209
+
this.#subscribe = createSubscriber((update) => {
210
+
const unsub = this.#store.subscribe((s) => {
211
+
this.#records = s.records as unknown as readonly TypedWatchRecord<RecordShapeOf<K>>[];
212
+
this.#status = s.status;
213
+
this.#error = s.error;
214
+
update();
215
+
});
216
+
this.#store.start();
217
+
return () => {
218
+
unsub();
219
+
this.#store.stop();
220
+
};
221
+
});
222
+
}
223
+
224
+
get records(): readonly TypedWatchRecord<RecordShapeOf<K>>[] {
225
+
this.#subscribe();
226
+
return this.#records;
227
+
}
228
+
get status(): WatchStoreStatus {
229
+
this.#subscribe();
230
+
return this.#status;
231
+
}
232
+
get error(): Error | null {
233
+
this.#subscribe();
234
+
return this.#error;
235
+
}
236
+
237
+
/** Insert an optimistic record into the query. It shows in `.records`
238
+
* immediately with `optimistic: 'pending'` and is auto-dropped when a
239
+
* server-confirmed record with the same rkey arrives via the stream. */
240
+
addOptimistic(input: {
241
+
rkey: string;
242
+
did: string;
243
+
record: RecordShapeOf<K>;
244
+
time_us?: number;
245
+
}): void {
246
+
this.#store.addOptimistic({
247
+
rkey: input.rkey,
248
+
did: input.did,
249
+
collection: this.#collection,
250
+
record: input.record as Record<string, unknown>,
251
+
time_us: input.time_us
252
+
});
253
+
}
254
+
255
+
/** Flip an optimistic entry's state to `'failed'` and attach an error. */
256
+
markFailed(rkey: string, err: Error): void {
257
+
this.#store.markFailed(rkey, err);
258
+
}
259
+
260
+
/** Remove an optimistic entry (explicit rollback). */
261
+
removeOptimistic(rkey: string): void {
262
+
this.#store.removeOptimistic(rkey);
263
+
}
264
+
}
265
+
266
+
export function createWatchQuery<K extends string>(
267
+
opts: WatchQueryOptions<K>
268
+
): WatchQuery<K> {
269
+
return new WatchQuery(opts);
270
+
}
271
+
272
+
// ---------------------------------------------------------------------------
273
+
274
+
function buildWatchUrl(endpoint: string, params: Record<string, unknown>): string {
275
+
const qs = new URLSearchParams();
276
+
for (const [k, v] of Object.entries(params)) {
277
+
if (v == null) continue;
278
+
if (Array.isArray(v)) {
279
+
for (const item of v) qs.append(k, String(item));
280
+
} else {
281
+
qs.set(k, String(v));
54
282
}
55
-
};
283
+
}
284
+
const query = qs.toString();
285
+
return `/xrpc/${endpoint}.watchRecords${query ? `?${query}` : ''}`;
56
286
}
+2
-20
examples/sveltekit-group-chat/src/routes/c/[communityDid]/[channelKey]/+page.server.ts
+2
-20
examples/sveltekit-group-chat/src/routes/c/[communityDid]/[channelKey]/+page.server.ts
···
1
1
import type { PageServerLoad } from './$types';
2
2
import { error } from '@sveltejs/kit';
3
3
import { buildSpaceUri } from '$lib/rooms/uri';
4
-
import { mintWatchTicket } from '$lib/rooms/server';
5
-
6
-
const WATCH_RECORDS_NSID = 'tools.atmo.chat.message.watchRecords';
7
4
8
-
export const load: PageServerLoad = async ({ locals, params, platform }) => {
5
+
export const load: PageServerLoad = async ({ locals, params }) => {
9
6
if (!locals.did) error(401, 'Not authenticated');
10
7
const communityDid = decodeURIComponent(params.communityDid);
11
8
const spaceUri = buildSpaceUri(communityDid, params.channelKey);
12
9
13
-
// Mint the first watch-scoped ticket alongside the page data so the browser
14
-
// can open the stream on first paint without a second roundtrip. Reconnects
15
-
// call the `mintWatchTicket` remote function.
16
-
let initialTicket: string | null = null;
17
-
try {
18
-
const t = await mintWatchTicket(
19
-
{ env: platform!.env, did: locals.did as string },
20
-
{ watchRecordsNsid: WATCH_RECORDS_NSID, spaceUri, limit: 50 }
21
-
);
22
-
initialTicket = t.ticket;
23
-
} catch {
24
-
// Non-fatal: the browser will mint on demand via the remote function.
25
-
}
26
-
27
10
return {
28
11
spaceUri,
29
12
channelKey: params.channelKey,
30
-
myDid: locals.did,
31
-
initialTicket
13
+
myDid: locals.did
32
14
};
33
15
};
+63
-71
examples/sveltekit-group-chat/src/routes/c/[communityDid]/[channelKey]/+page.svelte
+63
-71
examples/sveltekit-group-chat/src/routes/c/[communityDid]/[channelKey]/+page.svelte
···
2
2
import { tick } from 'svelte';
3
3
import { Button, Input, Navbar } from '@foxui/core';
4
4
import { RelativeTime } from '@foxui/time';
5
-
import { postMessage, mintWatchTicketCmd } from '$lib/rooms/rooms.remote';
5
+
import { postMessage } from '$lib/rooms/rooms.remote';
6
6
import { setCurrentChannel } from '$lib/rooms/realtime.svelte';
7
7
import { markLastRead } from '$lib/rooms/unread.svelte';
8
8
import { displayName } from '$lib/rooms/profiles.svelte';
9
9
import { createWatchQuery } from '$lib/rooms/watch.svelte';
10
-
import type { WatchRecord } from '@atmo-dev/contrail/sync';
11
-
import { dev } from '$app/environment';
12
10
import { setConnectionStatus, resetConnectionStatus } from '$lib/rooms/connection.svelte';
13
-
14
-
const WATCH_RECORDS_NSID = 'tools.atmo.chat.message.watchRecords';
11
+
import { nextTid } from '@atmo-dev/contrail';
15
12
16
13
let { data } = $props();
17
14
18
15
let channel = $derived(data.channels.find((c) => c.key === data.channelKey));
19
16
let channelName = $derived(channel?.name ?? data.channelKey);
20
17
21
-
// Live message feed via contrail's watchRecords subscription. The engine
22
-
// handles the snapshot + live merge; we just render its `records` array.
23
-
// Sorted newest-first by default — flip to oldest-first for chat.
24
-
//
25
-
// The query is keyed strictly on spaceUri — recreating it on every
26
-
// `data` identity change would tear down and rebuild the subscription on
27
-
// unrelated page-data invalidations (e.g. channel list refresh), flashing
28
-
// "Loading…" each time. Using a keyed effect here avoids that.
29
-
let query = $state<ReturnType<typeof createWatchQuery> | null>(null);
30
-
let currentUri: string | null = null;
31
-
32
-
$effect(() => {
33
-
const uri = data.spaceUri;
34
-
if (uri === currentUri) return;
35
-
query?.stop();
36
-
currentUri = uri;
37
-
38
-
// First connect reuses the ticket minted during SSR (landed on page
39
-
// data). Reconnects mint a fresh one via the remote function.
40
-
let pending: string | null = data.initialTicket;
41
-
query = createWatchQuery({
42
-
url: `/xrpc/${WATCH_RECORDS_NSID}?spaceUri=${encodeURIComponent(uri)}&limit=50`,
43
-
transport: dev ? 'sse' : 'ws',
44
-
fetchTicket: async () => {
45
-
if (pending) {
46
-
const t = pending;
47
-
pending = null;
48
-
return t;
49
-
}
50
-
const res = await mintWatchTicketCmd({
51
-
spaceUri: uri,
52
-
watchRecordsNsid: WATCH_RECORDS_NSID,
53
-
limit: 50
54
-
});
55
-
return res.ticket;
56
-
},
57
-
compareRecords: (a: WatchRecord, b: WatchRecord) =>
58
-
(a.time_us ?? 0) - (b.time_us ?? 0)
59
-
});
60
-
});
61
-
62
-
// Final teardown on component unmount.
63
-
$effect(() => {
64
-
return () => {
65
-
query?.stop();
66
-
query = null;
67
-
currentUri = null;
68
-
};
69
-
});
18
+
// Live message feed. `$derived` recreates the query when spaceUri changes;
19
+
// the old instance is auto-torn down via `createSubscriber` once no
20
+
// component reads it. Initial ticket is pre-minted during SSR to save one
21
+
// roundtrip; reconnects call the configured ticket minter transparently.
22
+
let messagesQuery = $derived(
23
+
createWatchQuery({
24
+
endpoint: 'tools.atmo.chat.message',
25
+
params: { spaceUri: data.spaceUri, limit: 50 },
26
+
compare: (a, b) => (a.time_us ?? 0) - (b.time_us ?? 0)
27
+
})
28
+
);
70
29
71
30
// Unread / current-channel bookkeeping.
72
31
$effect(() => {
···
78
37
// Mirror the query's connection status into the shared store so the
79
38
// layout's navbar can render a dot for it.
80
39
$effect(() => {
81
-
if (query) setConnectionStatus(query.status);
40
+
setConnectionStatus(messagesQuery.status);
82
41
return () => resetConnectionStatus();
83
42
});
84
43
85
-
// Project records to the shape the existing template expects.
86
44
let messages = $derived(
87
-
(query?.records ?? []).map((r) => {
88
-
const rec = r.record as { text?: string; createdAt?: string; replyTo?: string };
89
-
return {
90
-
rkey: r.rkey,
91
-
authorDid: r.did,
92
-
text: rec.text ?? '',
93
-
createdAt: rec.createdAt ?? '',
94
-
replyTo: rec.replyTo
95
-
};
96
-
})
45
+
messagesQuery.records.map((r) => ({
46
+
rkey: r.rkey,
47
+
authorDid: r.did,
48
+
text: r.record.text ?? '',
49
+
createdAt: r.record.createdAt ?? '',
50
+
replyTo: r.record.replyTo,
51
+
pending: r.optimistic === 'pending',
52
+
failed: r.optimistic === 'failed',
53
+
error: r.optimisticError
54
+
}))
97
55
);
98
56
99
57
let scrollEl: HTMLDivElement | null = $state(null);
···
114
72
if (!t) return;
115
73
sending = true;
116
74
sendErr = null;
75
+
// Client-generated rkey lets the optimistic entry reconcile with the
76
+
// stream's record.created event by identity.
77
+
const rkey = nextTid();
78
+
messagesQuery.addOptimistic({
79
+
rkey,
80
+
did: data.myDid,
81
+
record: {
82
+
$type: 'tools.atmo.chat.message',
83
+
text: t,
84
+
createdAt: new Date().toISOString()
85
+
}
86
+
});
87
+
text = '';
117
88
try {
118
-
await postMessage({ spaceUri: data.spaceUri, text: t });
119
-
text = '';
89
+
await postMessage({ spaceUri: data.spaceUri, rkey, text: t });
90
+
// Success: leave the optimistic entry — the stream will replace it.
120
91
} catch (err) {
121
-
sendErr = err instanceof Error ? err.message : String(err);
92
+
const e = err instanceof Error ? err : new Error(String(err));
93
+
messagesQuery.markFailed(rkey, e);
94
+
sendErr = e.message;
122
95
} finally {
123
96
sending = false;
124
97
}
···
132
105
</script>
133
106
134
107
<div class="flex min-h-0 flex-1 flex-col pb-20">
108
+
{#if messages.length > 0 && (messagesQuery.status === 'connecting' || messagesQuery.status === 'snapshot')}
109
+
<div
110
+
class="text-base-500 absolute top-[4.5rem] left-1/2 z-10 flex -translate-x-1/2 items-center gap-1.5 rounded-full bg-white/80 px-3 py-1 text-xs shadow-sm backdrop-blur dark:bg-black/80"
111
+
aria-label="updating"
112
+
>
113
+
<svg class="size-3 animate-spin" viewBox="0 0 24 24" fill="none">
114
+
<circle cx="12" cy="12" r="10" stroke="currentColor" stroke-opacity="0.25" stroke-width="4" />
115
+
<path d="M12 2a10 10 0 0 1 10 10" stroke="currentColor" stroke-width="4" stroke-linecap="round" />
116
+
</svg>
117
+
updating
118
+
</div>
119
+
{/if}
135
120
<div bind:this={scrollEl} class="flex-1 overflow-y-auto px-4 py-6">
136
-
{#if messages.length === 0 && (query?.status === 'connecting' || query?.status === 'snapshot' || query?.status === 'idle')}
121
+
{#if messages.length === 0 && (messagesQuery.status === 'connecting' || messagesQuery.status === 'snapshot' || messagesQuery.status === 'idle')}
137
122
<div class="text-base-500 text-center text-sm">Loading…</div>
138
123
{:else if messages.length === 0}
139
124
<div class="text-base-500 text-center text-sm">No messages yet. Say hi.</div>
···
147
132
<span class="font-medium">{displayName(m.authorDid)}</span>
148
133
·
149
134
<RelativeTime date={new Date(m.createdAt)} locale="en-US" />
135
+
{#if m.pending}
136
+
· <span class="text-base-400">sending…</span>
137
+
{:else if m.failed}
138
+
· <span class="text-red-500">failed{m.error ? ` (${m.error.message})` : ''}</span>
139
+
{/if}
150
140
</div>
151
141
<div
152
142
class="{m.authorDid === data.myDid
153
143
? 'bg-accent-500 text-white'
154
-
: 'bg-base-200 dark:bg-base-800'} mt-1 inline-block rounded-2xl px-3 py-2 text-sm whitespace-pre-wrap"
144
+
: 'bg-base-200 dark:bg-base-800'} mt-1 inline-block rounded-2xl px-3 py-2 text-sm whitespace-pre-wrap {m.pending
145
+
? 'opacity-60'
146
+
: ''} {m.failed ? 'opacity-60 ring-1 ring-red-500' : ''}"
155
147
>
156
148
{m.text}
157
149
</div>
+4
package.json
+4
package.json
···
37
37
"types": "./dist/sync/index.d.ts",
38
38
"import": "./dist/sync/index.js"
39
39
},
40
+
"./sync/cache-idb": {
41
+
"types": "./dist/sync/cache-idb.d.ts",
42
+
"import": "./dist/sync/cache-idb.js"
43
+
},
40
44
"./lexicon-templates/*.json": "./lexicon-templates/*.json"
41
45
},
42
46
"repository": {
+91
src/sync/cache-idb.ts
+91
src/sync/cache-idb.ts
···
1
+
/// <reference lib="dom" />
2
+
/** IndexedDB-backed implementation of `WatchCache`. Browser-only.
3
+
*
4
+
* Single object store keyed by the watch query's `cacheKey` (defaults to
5
+
* the watch URL). Values are `{records, updatedAt}`. Persists across page
6
+
* reloads; cleared by the user clearing site data. */
7
+
8
+
import type { WatchCache, WatchRecord } from "./index.js";
9
+
10
+
interface StoredEntry {
11
+
records: WatchRecord[];
12
+
updatedAt: number;
13
+
}
14
+
15
+
export interface IndexedDBCacheOptions {
16
+
/** Database name. Default: 'contrail-watch-cache'. Bump this (or delete
17
+
* via DevTools) if you need to invalidate all cached watch data. */
18
+
dbName?: string;
19
+
/** Object store name within the DB. Default: 'watches'. */
20
+
storeName?: string;
21
+
/** IDB schema version. Default: 1. Only bump when changing
22
+
* storeName / schema. */
23
+
version?: number;
24
+
}
25
+
26
+
export function createIndexedDBCache(
27
+
options: IndexedDBCacheOptions = {}
28
+
): WatchCache {
29
+
const dbName = options.dbName ?? "contrail-watch-cache";
30
+
const storeName = options.storeName ?? "watches";
31
+
const version = options.version ?? 1;
32
+
33
+
let dbPromise: Promise<IDBDatabase> | null = null;
34
+
const openDb = (): Promise<IDBDatabase> => {
35
+
if (dbPromise) return dbPromise;
36
+
dbPromise = new Promise<IDBDatabase>((resolve, reject) => {
37
+
if (typeof indexedDB === "undefined") {
38
+
reject(new Error("IndexedDB not available in this environment"));
39
+
return;
40
+
}
41
+
const req = indexedDB.open(dbName, version);
42
+
req.onupgradeneeded = () => {
43
+
const db = req.result;
44
+
if (!db.objectStoreNames.contains(storeName)) {
45
+
db.createObjectStore(storeName);
46
+
}
47
+
};
48
+
req.onsuccess = () => resolve(req.result);
49
+
req.onerror = () => reject(req.error);
50
+
req.onblocked = () =>
51
+
reject(new Error("IDB open blocked (another tab holds an older version)"));
52
+
});
53
+
return dbPromise;
54
+
};
55
+
56
+
return {
57
+
async read(key) {
58
+
try {
59
+
const db = await openDb();
60
+
return await new Promise<WatchRecord[] | null>((resolve, reject) => {
61
+
const tx = db.transaction(storeName, "readonly");
62
+
const store = tx.objectStore(storeName);
63
+
const req = store.get(key);
64
+
req.onsuccess = () => {
65
+
const entry = req.result as StoredEntry | undefined;
66
+
resolve(entry?.records ?? null);
67
+
};
68
+
req.onerror = () => reject(req.error);
69
+
});
70
+
} catch {
71
+
return null;
72
+
}
73
+
},
74
+
async write(key, records) {
75
+
try {
76
+
const db = await openDb();
77
+
await new Promise<void>((resolve, reject) => {
78
+
const tx = db.transaction(storeName, "readwrite");
79
+
const store = tx.objectStore(storeName);
80
+
const entry: StoredEntry = { records, updatedAt: Date.now() };
81
+
store.put(entry, key);
82
+
tx.oncomplete = () => resolve();
83
+
tx.onerror = () => reject(tx.error);
84
+
tx.onabort = () => reject(tx.error ?? new Error("tx aborted"));
85
+
});
86
+
} catch {
87
+
// Intentional: caching is best-effort; swallow.
88
+
}
89
+
},
90
+
};
91
+
}
+164
-12
src/sync/index.ts
+164
-12
src/sync/index.ts
···
23
23
cid?: string | null;
24
24
/** Set when the record originates from a per-space table. */
25
25
_space?: string;
26
+
/** Present on optimistic entries added via `addOptimistic` — not set by
27
+
* records arriving from the stream. Auto-dropped when a real record
28
+
* with the same rkey arrives via `record.created`. */
29
+
optimistic?: "pending" | "failed";
30
+
/** Error attached via `markFailed` after a failed mutation. */
31
+
optimisticError?: Error;
26
32
/** Additional hydrated relations / references populated server-side. */
27
33
[k: string]: unknown;
28
34
}
···
33
39
/** Transport. Default: 'sse'. Use 'ws' on Cloudflare for DO-terminated
34
40
* long-lived subscriptions that hibernate while idle. */
35
41
transport?: "sse" | "ws";
36
-
/** Mint a watch-scoped ticket to auth the connection. Called once per
37
-
* connect attempt; the returned ticket is sent as `?ticket=...` on the
38
-
* SSE connection or on the `mode=ws` handshake fetch. Omit for public
39
-
* (no-auth) endpoints.
42
+
/** Watch-scoped ticket auth for the connection. Sent as `?ticket=...`
43
+
* on the SSE connection or on the `mode=ws` handshake fetch.
44
+
*
45
+
* - `string` — used as-is on every connect (good for one-shot SSR-minted
46
+
* tokens; reconnects after expiry will surface a 401).
47
+
* - `() => Promise<string>` — called once per connect attempt; a fresh
48
+
* ticket is minted for every (re)connect.
49
+
* - omitted — public (no-auth) endpoints.
40
50
*
41
51
* For atproto-based services, tickets are typically minted server-side
42
52
* via `<ns>.realtime.ticket` (or via an app-specific route that runs
43
53
* the `mode=ws` handshake in-process and returns the signed ticket). */
44
-
fetchTicket?: () => Promise<string>;
54
+
mintTicket?: string | (() => Promise<string>);
45
55
/** Custom compare. Default: sort by `time_us` descending (newest first),
46
56
* tie-breaking by rkey. */
47
57
compareRecords?: (a: WatchRecord, b: WatchRecord) => number;
48
58
/** Reconnect on error with backoff. Default: true, 1s → 30s exponential. */
49
59
reconnect?: boolean;
60
+
/** Optional persistent cache. When provided, the store loads cached
61
+
* records at `start()` for instant first paint, then reconciles against
62
+
* the live snapshot as usual. Writes are debounced and exclude
63
+
* optimistic entries. */
64
+
cache?: WatchCache;
65
+
/** Key for this query in the cache. Defaults to `url`. */
66
+
cacheKey?: string;
67
+
/** Max records retained in the cache per key. Excess records (by sort
68
+
* order) are dropped on write. Default: 200. */
69
+
cacheMaxRecords?: number;
50
70
/** Optional logger for debug output. */
51
71
logger?: {
52
72
log?: (...args: unknown[]) => void;
···
55
75
};
56
76
}
57
77
78
+
/** Adapter interface for persisting the last-seen records of a watch query.
79
+
* Implementations can back this with IndexedDB, localStorage, fs, etc.
80
+
* Errors should be caught internally — caching is a best-effort optimization
81
+
* and must never break the live stream. */
82
+
export interface WatchCache {
83
+
read(key: string): Promise<WatchRecord[] | null>;
84
+
write(key: string, records: WatchRecord[]): Promise<void>;
85
+
}
86
+
58
87
export type WatchStoreStatus =
59
88
| "idle"
60
89
| "connecting"
···
73
102
start(): void;
74
103
/** Close the connection and clear the store. */
75
104
stop(): void;
105
+
106
+
/** Insert an optimistic record immediately, before the server confirms.
107
+
* The entry is merged into `.records` with `optimistic: 'pending'`. When
108
+
* a real record arrives via the stream with the same `rkey`, the
109
+
* optimistic entry is dropped automatically. */
110
+
addOptimistic(input: {
111
+
rkey: string;
112
+
did: string;
113
+
collection?: string;
114
+
record: Record<string, unknown>;
115
+
time_us?: number;
116
+
uri?: string;
117
+
}): void;
118
+
/** Flip an optimistic entry to `optimistic: 'failed'` and attach an error.
119
+
* No-op if no optimistic entry matches. */
120
+
markFailed(rkey: string, err: Error): void;
121
+
/** Remove an optimistic entry (explicit rollback). */
122
+
removeOptimistic(rkey: string): void;
76
123
}
77
124
78
125
export interface WatchStoreState {
···
95
142
const reconnect = options.reconnect !== false;
96
143
const transport = options.transport ?? "sse";
97
144
const log = options.logger ?? {};
145
+
const cache = options.cache ?? null;
146
+
const cacheKey = options.cacheKey ?? options.url;
147
+
const cacheMax = options.cacheMaxRecords ?? 200;
98
148
99
149
const byKey = new Map<string, WatchRecord>();
150
+
/** Optimistic entries keyed by rkey. Merged into `.records`; dropped when
151
+
* a real record with the same rkey arrives via `record.created`. */
152
+
const optimisticByRkey = new Map<string, WatchRecord>();
100
153
let sorted: WatchRecord[] = [];
101
154
let status: WatchStoreStatus = "idle";
102
155
let error: Error | null = null;
103
156
const listeners = new Set<(state: WatchStoreState) => void>();
104
157
158
+
// Debounced cache writer. Schedules one write after the current burst of
159
+
// state changes settles, to avoid writing on every incoming event.
160
+
let cacheWriteTimer: ReturnType<typeof setTimeout> | null = null;
161
+
const scheduleCacheWrite = () => {
162
+
if (!cache) return;
163
+
if (cacheWriteTimer) return;
164
+
cacheWriteTimer = setTimeout(() => {
165
+
cacheWriteTimer = null;
166
+
void flushCache();
167
+
}, 200);
168
+
};
169
+
const flushCache = async () => {
170
+
if (!cache) return;
171
+
// Snapshot the current server-confirmed records (exclude optimistic)
172
+
// and trim to cacheMax.
173
+
const confirmed: WatchRecord[] = [];
174
+
for (const r of byKey.values()) confirmed.push(r);
175
+
confirmed.sort(compare);
176
+
const trimmed = confirmed.slice(0, cacheMax);
177
+
try {
178
+
await cache.write(cacheKey, trimmed);
179
+
} catch (err) {
180
+
log.warn?.("cache write failed", err);
181
+
}
182
+
};
183
+
105
184
// Per-snapshot reconcile: during a snapshot we track which keys were
106
185
// included, and on snapshot.end we evict any stale keys the previous
107
186
// snapshot had but this one didn't. Keeps data visible across reconnects
···
119
198
const notify = () => {
120
199
const s = stateSnapshot();
121
200
for (const l of listeners) l(s);
201
+
scheduleCacheWrite();
122
202
};
123
203
124
204
const resort = () => {
125
-
sorted = Array.from(byKey.values()).sort(compare);
205
+
const merged: WatchRecord[] = [];
206
+
for (const r of byKey.values()) merged.push(r);
207
+
for (const r of optimisticByRkey.values()) merged.push(r);
208
+
sorted = merged.sort(compare);
126
209
};
127
210
128
211
const setStatus = (next: WatchStoreStatus, nextError: Error | null = null) => {
···
141
224
const k = key(record);
142
225
byKey.set(k, record);
143
226
snapshotSeen?.add(k);
227
+
// If this rkey was optimistic, it's now server-confirmed.
228
+
optimisticByRkey.delete(record.rkey);
144
229
};
145
230
146
231
const applyCreated = (record: WatchRecord) => {
232
+
// Drop any optimistic entry with the same rkey — the server-confirmed
233
+
// record replaces it.
234
+
optimisticByRkey.delete(record.rkey);
147
235
byKey.set(key(record), record);
148
236
resort();
149
237
notify();
···
270
358
}
271
359
};
272
360
361
+
const resolveTicket = async (): Promise<string | null> => {
362
+
const src = options.mintTicket;
363
+
if (!src) return null;
364
+
return typeof src === "string" ? src : await src();
365
+
};
366
+
273
367
const openSse = async () => {
274
368
let url = options.url;
275
-
if (options.fetchTicket) {
276
-
const ticket = await options.fetchTicket();
369
+
const ticket = await resolveTicket();
370
+
if (ticket) {
277
371
const sep = url.includes("?") ? "&" : "?";
278
372
url += `${sep}ticket=${encodeURIComponent(ticket)}`;
279
373
}
···
310
404
311
405
const openWs = async () => {
312
406
// Step 1: snapshot + handshake. Authenticated by the watch-scoped
313
-
// ticket from `fetchTicket`, passed as `?ticket=...`. The server
407
+
// ticket from `mintTicket`, passed as `?ticket=...`. The server
314
408
// returns a ticket bound to (did, spaceUri, querySpec); we pass that
315
409
// back embedded in the wsUrl on the WS upgrade so the WS itself
316
410
// doesn't need any other auth.
317
411
let snapshotUrl = options.url;
318
412
const sep = snapshotUrl.includes("?") ? "&" : "?";
319
413
snapshotUrl += `${sep}mode=ws`;
320
-
if (options.fetchTicket) {
321
-
const ticket = await options.fetchTicket();
414
+
const ticket = await resolveTicket();
415
+
if (ticket) {
322
416
snapshotUrl += `&ticket=${encodeURIComponent(ticket)}`;
323
417
}
324
418
const res = await fetch(snapshotUrl, { headers: { accept: "application/json" } });
···
428
522
if (started) return;
429
523
started = true;
430
524
stopped = false;
431
-
void openOnce();
525
+
void (async () => {
526
+
// Cache warm: populate from disk before opening the connection
527
+
// so listeners see instant records on first paint.
528
+
if (cache) {
529
+
try {
530
+
const cached = await cache.read(cacheKey);
531
+
if (cached && cached.length > 0 && !stopped) {
532
+
for (const r of cached) byKey.set(key(r), r);
533
+
resort();
534
+
notify();
535
+
}
536
+
} catch (err) {
537
+
log.warn?.("cache read failed", err);
538
+
}
539
+
}
540
+
if (stopped) return;
541
+
void openOnce();
542
+
})();
432
543
},
433
544
stop() {
434
545
stopped = true;
435
546
closeConnections();
547
+
if (cacheWriteTimer) {
548
+
clearTimeout(cacheWriteTimer);
549
+
cacheWriteTimer = null;
550
+
}
551
+
// Best-effort final flush so the cache reflects the last-known
552
+
// confirmed state before teardown.
553
+
void flushCache();
436
554
byKey.clear();
555
+
optimisticByRkey.clear();
437
556
sorted = [];
438
557
setStatus("closed");
558
+
},
559
+
addOptimistic(input) {
560
+
const now = Date.now();
561
+
const record: WatchRecord = {
562
+
uri: input.uri ?? `at://${input.did}/${input.collection ?? ""}/${input.rkey}`,
563
+
did: input.did,
564
+
rkey: input.rkey,
565
+
collection: input.collection ?? "",
566
+
record: input.record,
567
+
time_us: input.time_us ?? now * 1000,
568
+
indexed_at: now,
569
+
cid: null,
570
+
optimistic: "pending"
571
+
};
572
+
optimisticByRkey.set(input.rkey, record);
573
+
resort();
574
+
notify();
575
+
},
576
+
markFailed(rkey, err) {
577
+
const existing = optimisticByRkey.get(rkey);
578
+
if (!existing) return;
579
+
optimisticByRkey.set(rkey, {
580
+
...existing,
581
+
optimistic: "failed",
582
+
optimisticError: err
583
+
});
584
+
resort();
585
+
notify();
586
+
},
587
+
removeOptimistic(rkey) {
588
+
if (!optimisticByRkey.delete(rkey)) return;
589
+
resort();
590
+
notify();
439
591
}
440
592
};
441
593
}
+120
todo/cross-space-watch-records.md
+120
todo/cross-space-watch-records.md
···
1
+
# Cross-space watchRecords (actor-scoped live queries)
2
+
3
+
## Context
4
+
5
+
Today `<collection>.watchRecords` is per-space only — the endpoint rejects
6
+
requests without `spaceUri`:
7
+
8
+
```ts
9
+
// src/core/router/collection.ts:563-568
10
+
if (!spaceUri) {
11
+
return c.json(
12
+
{ error: "InvalidRequest", message: "spaceUri required (cross-space watch is deferred)" },
13
+
400
14
+
);
15
+
}
16
+
```
17
+
18
+
This means the app's `createWatchQuery` helper only works for queries
19
+
scoped to one permissioned space. `listRecords` has no such restriction —
20
+
it already supports `?actor=<did>` for cross-space listings.
21
+
22
+
## The concrete gap in the group-chat example
23
+
24
+
The channel list spans multiple spaces: each channel lives in its own
25
+
space, authored by the community DID. Current flow:
26
+
27
+
- **SSR**: `+layout.server.ts` calls `<ns>.channel.listRecords` with
28
+
`actor=<communityDid>`.
29
+
- **Live updates**: `connectCommunityRealtime` in `realtime.svelte.ts`
30
+
opens an EventSource on `<ns>.realtime.subscribe?ticket=...` keyed to
31
+
`community:<did>`, filters for `tools.atmo.chat.channel` events, and
32
+
calls `invalidateAll()` to re-run the loader.
33
+
34
+
That works, but it means:
35
+
- Two realtime mechanisms in the app (watchRecords for messages, raw
36
+
`realtime.subscribe` + `invalidateAll` for channels).
37
+
- Channel list is re-fetched from the server on every update rather than
38
+
being incrementally maintained client-side.
39
+
40
+
## The fix
41
+
42
+
Allow `watchRecords` to be actor-scoped, mirroring `listRecords`:
43
+
44
+
- Remove the "spaceUri required" gate when `actor` is supplied.
45
+
- Snapshot: reuse the existing `runPipeline` path (already handles
46
+
actor-filtered + permissioned-space union for the caller).
47
+
- Live topic: `actor:<did>` (contrail already exposes `actorTopic`)
48
+
or `community:<did>` (via `communityTopic` when the actor is a
49
+
registered community DID — the ticket endpoint already resolves this).
50
+
- Ticket minting: the `<ns>.realtime.ticket` endpoint already accepts
51
+
`{ topic: "community:<did>" }` and returns a scoped ticket.
52
+
53
+
Once done, the app reads:
54
+
55
+
```ts
56
+
const channelsQuery = $derived(
57
+
createWatchQuery({
58
+
endpoint: 'tools.atmo.chat.channel',
59
+
params: { actor: data.communityDid, limit: 100 }
60
+
})
61
+
);
62
+
```
63
+
64
+
And `connectCommunityRealtime` + the `invalidateAll` dance goes away
65
+
(channel list is maintained incrementally; messages already use their
66
+
own per-space watch query).
67
+
68
+
## Open question to resolve before implementing
69
+
70
+
**ACL on the live stream.** `listRecords` filters per-space membership
71
+
on the query (via `runPipeline`'s `memberDid` path). For `watchRecords`
72
+
in this mode, events come from the pubsub via the `community:<did>` (or
73
+
`actor:<did>`) topic. **It needs to be verified** that the stream
74
+
actually filters events against the caller's per-space memberships —
75
+
otherwise a non-member could see `record.created` for a private channel
76
+
they're not in.
77
+
78
+
Audit targets:
79
+
- `src/core/realtime/publishing-adapter.ts` — what topics does it publish
80
+
a space write to? If it publishes to `community:<did>` unconditionally,
81
+
that event is delivered to every community-topic subscriber regardless
82
+
of their per-space access.
83
+
- `src/core/realtime/router.ts` / `resolveTopicForCaller` — does the
84
+
ticket issuer verify the caller's membership at ticket-mint time, or
85
+
is `community:<did>` treated as a public-ish topic available to anyone
86
+
who can name the community?
87
+
- `src/core/router/collection.ts:653-697` — the watchRecords WS path
88
+
filters by `querySpec`; the SSE path has its own filtering. Either
89
+
way, filtering happens server-side; make sure it evaluates
90
+
per-space membership, not just collection/actor match.
91
+
92
+
If the existing filter story is insufficient, a small gate needs to be
93
+
added: each event that leaves the stream is checked against the
94
+
caller's membership of the event's `space`, and dropped if the caller
95
+
has no access.
96
+
97
+
## Scope estimate
98
+
99
+
- If ACL is already correct: ~1 hour. Small change to remove the
100
+
spaceUri gate and wire actor → topic resolution.
101
+
- If ACL filter needs adding: ~half-day. Introduce a per-event
102
+
membership check in the stream's filter path and cache memberships
103
+
on the connection to keep it cheap.
104
+
105
+
## Out of scope for this work
106
+
107
+
- Combining multiple `actor` values (single-actor queries are enough
108
+
for the community-channels use case).
109
+
- Hydration relations across the actor-scoped snapshot — same machinery
110
+
as today; just confirm it works without `spaceUri` context.
111
+
112
+
## Cleanup after this ships
113
+
114
+
- Delete `connectCommunityRealtime` from the example's
115
+
`lib/rooms/realtime.svelte.ts` (and its ticket fetching helper).
116
+
- `+layout.server.ts` stops fetching channels and pass them via page
117
+
data; the layout uses `createWatchQuery` directly.
118
+
- Consider whether `channelMessages` (the per-space message store) is
119
+
still needed — it's there for the prior SSR-message pattern; messages
120
+
are now read directly from `messagesQuery.records`. Probably delete.
+104
todo/realtime-do-dev-wiring.md
+104
todo/realtime-do-dev-wiring.md
···
1
+
# Fix DO wiring in Vite dev (currently papered over with a `dev` flag)
2
+
3
+
## Context
4
+
5
+
In `examples/sveltekit-group-chat/src/lib/contrail/index.ts` the realtime
6
+
PubSub is selected with:
7
+
8
+
```ts
9
+
const pubsub = dev ? new InMemoryPubSub() : new DurableObjectPubSub(env.REALTIME);
10
+
```
11
+
12
+
That `dev` check is standing in for "the `RealtimePubSubDO` class isn't
13
+
actually registered with miniflare right now." It happens to line up with
14
+
Vite dev today, but it's not really a dev-vs-prod concern — it's a
15
+
build-wiring concern.
16
+
17
+
### Why the DO isn't available in Vite dev
18
+
19
+
- `wrangler.jsonc` declares the DO binding `REALTIME` with
20
+
`class_name: "RealtimePubSubDO"` and `main: ".svelte-kit/cloudflare/_worker.js"`.
21
+
- `@sveltejs/adapter-cloudflare@7.x` sets up miniflare via `platformProxy`
22
+
automatically in dev, so `platform.env.REALTIME` is bound.
23
+
- But the class lives inside the built worker. It's re-exported into
24
+
`_worker.js` by `scripts/append-scheduled.ts` *after* `pnpm build`.
25
+
- Vite dev doesn't run that post-build step, so the DO binding points at
26
+
nothing and `stub.fetch(...)` crashes with `Internal Error`.
27
+
28
+
## Three fixes, smallest → largest
29
+
30
+
### 1. App-level probe (smallest)
31
+
32
+
In the example only. Add a helper that tries a cheap `stub.fetch(health)`
33
+
at bundle-build time and falls back to `InMemoryPubSub` if it throws.
34
+
35
+
```ts
36
+
async function pickPubSub(ns: DurableObjectNamespace | undefined): Promise<PubSub> {
37
+
if (!ns) return new InMemoryPubSub();
38
+
try {
39
+
const stub = ns.get(ns.idFromName('__probe__'));
40
+
const res = await stub.fetch('http://probe/__health', { signal: AbortSignal.timeout(250) });
41
+
if (res.ok) return new DurableObjectPubSub(ns);
42
+
} catch {}
43
+
return new InMemoryPubSub();
44
+
}
45
+
```
46
+
47
+
- Pro: no library change, no build change.
48
+
- Con: async bundle construction (currently sync), every app using contrail
49
+
has to reinvent this.
50
+
51
+
### 2. Library helper with probe (medium, recommended)
52
+
53
+
Ship the probe in `@atmo-dev/contrail` as something like:
54
+
55
+
```ts
56
+
export async function resolveRealtimePubSub(
57
+
ns?: DurableObjectNamespace,
58
+
fallback: () => PubSub = () => new InMemoryPubSub(),
59
+
): Promise<PubSub>
60
+
```
61
+
62
+
Every app using contrail gets the correct behavior for free. Still async
63
+
construction, but now that lives at the library boundary where it's
64
+
easier to swallow.
65
+
66
+
- Pro: reusable primitive, fixes the smell at the right layer.
67
+
- Con: contrail now has opinions about "is the DO actually usable"; async
68
+
bundle setup ripples to every consumer (our current `getBundle` is sync).
69
+
70
+
### 3. Proper dev-prod parity (largest, most correct)
71
+
72
+
Actually register the DO class with miniflare in Vite dev so `env.REALTIME`
73
+
works identically to prod.
74
+
75
+
- Restructure the DO re-export so it lives at a stable source path (no
76
+
post-build patching).
77
+
- Configure `adapter({ platformProxy: { ... } })` with miniflare options
78
+
that point at that stable entry for DO hosting. Likely needs a
79
+
dedicated DO worker script in miniflare's `workers` config.
80
+
- Adjust `scripts/append-scheduled.ts` so it no longer duplicates the
81
+
re-export; it only keeps the `scheduled` handler append.
82
+
- Drop the `dev ? InMemory : DO` branch in the example entirely — always
83
+
`new DurableObjectPubSub(env.REALTIME)`.
84
+
85
+
- Pro: dev === prod, no branching anywhere, also unblocks WS-in-dev
86
+
(which today forces `transport: dev ? 'sse' : 'ws'` in
87
+
`+page.svelte`).
88
+
- Con: touches build config, wrangler config, and adapter-cloudflare
89
+
internals. Non-trivial to get right; worth its own session.
90
+
91
+
## Related cleanup (do in the same pass)
92
+
93
+
- Once the DO works in dev, drop `transport: dev ? 'sse' : 'ws'` at
94
+
`src/routes/c/[communityDid]/[channelKey]/+page.svelte` — always `'ws'`.
95
+
- Remove the `dev` import from `src/lib/contrail/index.ts` and
96
+
`+page.svelte` (only remaining consumer).
97
+
- Revisit whether `InMemoryPubSub` should stay exported at all from the
98
+
example config; if option 3 lands, it's only needed for non-CF hosts.
99
+
100
+
## Recommendation
101
+
102
+
Go with **(2)** next. It's the right layer, and the async-construction
103
+
cost is modest (one-time probe at first use). Tackle (3) when we do the
104
+
broader "unify transport" pass.