-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathserver.js
More file actions
8894 lines (8327 loc) · 339 KB
/
Copy pathserver.js
File metadata and controls
8894 lines (8327 loc) · 339 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
// @ts-check
import { assert, wireAssertionMetrics } from './shared/assert.js';
import { safeAssign as _safeAssignSnapshot } from './shared/safe-assign.js';
export { assert, getAssertionCounters, _resetAssertCounters } from './shared/assert.js';
const textDecoder = new TextDecoder();
const _validPathRe = /^[a-zA-Z0-9_-]+(?:\/[a-zA-Z0-9_-]+)+$/;
const _validSegmentRe = /^[a-zA-Z0-9_]+$/;
/**
* Max accepted length for a userId that flows into a topic name via
* `__signal:${userId}` / `__push:${userId}` and similar server-built
* system topics. 256 chars is generous for any realistic identifier
* (UUIDs, opaque session tokens, prefixed-by-tenant ids) without
* bloating log lines or stressing the adapter's wire-topic budget.
*/
const _MAX_USER_ID_LENGTH = 256;
/**
* Validate that a userId is safe to interpolate into a system topic
* name. Returns `null` if valid, otherwise a short error reason string
* suitable for embedding in a thrown LiveError / Error message.
*
* Server-side helpers that build `__signal:${userId}` / `__push:${userId}`
* topic names from caller-supplied identifiers go through this gate so
* malformed identifiers (control bytes, CR/LF, NUL, quotes, backslash,
* empty, non-string, oversized) cannot poison the topic namespace,
* corrupt log lines, or escape the system-topic prefix into the
* user-topic space. Non-ASCII bytes are allowed for parity with the
* adapter's `allowNonAsciiTopics` opt-in; the server-side builder
* trusts identifier shapes set by upgrade hooks.
*
* @param {unknown} userId
* @returns {string | null}
*/
function _validUserIdReason(userId) {
if (typeof userId !== 'string') return 'userId must be a string (got ' + (typeof userId) + ')';
if (userId.length === 0) return 'userId must be non-empty';
if (userId.length > _MAX_USER_ID_LENGTH) return 'userId exceeds maximum length ' + _MAX_USER_ID_LENGTH + ' (got ' + userId.length + ')';
for (let i = 0; i < userId.length; i++) {
const c = userId.charCodeAt(i);
// Reject ASCII C0 controls (0x00-0x1F), DEL (0x7F), and the two
// characters the adapter's wire-topic validator forbids:
// 0x22 (double-quote), 0x5C (backslash).
if (c < 0x20 || c === 0x7F || c === 0x22 || c === 0x5C) {
return 'userId contains invalid character at index ' + i + ' (charCode ' + c + ')';
}
}
return null;
}
// - Bounded-by-default capacity caps (server side) -------------------------
// Every per-process Map / Set with caller-driven growth is bounded. Numbers
// are deliberately generous - far above any healthy single-instance workload
// - so they catch obvious bugs (subscribe-leak, register-without-deregister)
// without biting real apps. Saturation behavior is one of:
//
// REJECT caller gets an explicit error or a silent skip
// WARN-ONLY logs once per category; map keeps growing because eviction
// would corrupt routing
// FIFO-EVICT drops oldest insertion-order entries; safe for dedup state
// where re-warn or duplicate is acceptable
//
// Existing caps not re-declared here (already enforced at their sites):
// _RATE_LIMIT_MAX 5000 REJECT after stale-sweep (per-identity)
// _THROTTLE_DEBOUNCE_MAX 5000 fall-back to direct publish (per-key)
// idempotency maxEntries 10000 FIFO 10% (in-process result store)
//
// See README "Capacity model" for the full taxonomy.
/** Max distinct userIds tracked in the per-process push registry. WARN-then-skip on cap: new registrations are dropped (the connection still works, it just can't be the target of `live.push({ userId })` until existing entries clear). Matches the cluster-scale convention from svelte-adapter-uws-extensions (`MAX_REGISTRY_SESSIONS_PER_INSTANCE`). */
export const MAX_PUSH_REGISTRY = 10_000_000;
/** Threshold for the per-process topic-subscribers index. WARN-ONLY: the map keeps growing because eviction would corrupt subscribe / unsubscribe routing. Surfaces a structured warning the first time the threshold is crossed. Matches svelte-adapter-uws `TOPIC_SEQS_WARN_THRESHOLD`. */
export const TOPIC_WS_COUNTS_WARN_THRESHOLD = 1_000_000;
/** Max distinct topics in the dev-mode silent-topic warning dedup. FIFO-evict on cap: dropping the oldest entry just lets that topic re-warn on its next over-threshold subscribe. Matches svelte-adapter-uws `PUBLISH_WARN_DEDUP_MAX`. */
export const SILENT_TOPIC_WARN_DEDUP_MAX = 1_000_000;
/** Max distinct topics in the dev-mode publish-rate warning dedup. FIFO-evict on cap: dropping the oldest entry just lets that topic re-warn on its next sample tick. Matches svelte-adapter-uws `PUBLISH_WARN_DEDUP_MAX`. */
export const PUBLISH_RATE_WARN_DEDUP_MAX = 1_000_000;
/** Max distinct (user, room) pairs the in-memory presence-ref map holds. FIFO-evict graces first, then WARN-then-skip on cap: new joiners don't get registered, so they're invisible in any subscriber's roster until existing entries clear. Matches the per-process safety-net convention used by the other caps in this section. For multi-instance deploys, wire `platform.presence` (e.g. `svelte-adapter-uws-extensions/presence`) to bypass this map entirely. */
export const MAX_PRESENCE_REF = 1_000_000;
// Mutable internal copies: the public `export const` values above are the
// canonical defaults; tests use `_setCapsForTest` to lower them for fast
// saturation scenarios. Production never touches these.
let _maxPushRegistry = MAX_PUSH_REGISTRY;
let _topicWsCountsWarnThreshold = TOPIC_WS_COUNTS_WARN_THRESHOLD;
let _silentTopicWarnDedupMax = SILENT_TOPIC_WARN_DEDUP_MAX;
let _publishRateWarnDedupMax = PUBLISH_RATE_WARN_DEDUP_MAX;
let _maxPresenceRef = MAX_PRESENCE_REF;
/**
* Override capacity caps for testing. Pass any subset of the cap names
* (omit `MAX_` / `_THRESHOLD` / `_MAX` suffix; use `pushRegistry`,
* `topicWsCountsWarn`, `silentTopicWarnDedup`, `publishRateWarnDedup`,
* `presenceRef`). Pair with `_resetCapsForTest()` in afterEach.
* @internal
* @param {{ pushRegistry?: number, topicWsCountsWarn?: number, silentTopicWarnDedup?: number, publishRateWarnDedup?: number, presenceRef?: number }} overrides
*/
export function _setCapsForTest(overrides) {
if (overrides.pushRegistry !== undefined) _maxPushRegistry = overrides.pushRegistry;
if (overrides.topicWsCountsWarn !== undefined) _topicWsCountsWarnThreshold = overrides.topicWsCountsWarn;
if (overrides.silentTopicWarnDedup !== undefined) _silentTopicWarnDedupMax = overrides.silentTopicWarnDedup;
if (overrides.publishRateWarnDedup !== undefined) _publishRateWarnDedupMax = overrides.publishRateWarnDedup;
if (overrides.presenceRef !== undefined) _maxPresenceRef = overrides.presenceRef;
if (overrides.uploadPendingMaxAggregate !== undefined) _UPLOAD_PENDING_MAX_AGGREGATE = overrides.uploadPendingMaxAggregate;
}
/**
* Restore capacity caps to their default values.
* @internal
*/
export function _resetCapsForTest() {
_maxPushRegistry = MAX_PUSH_REGISTRY;
_topicWsCountsWarnThreshold = TOPIC_WS_COUNTS_WARN_THRESHOLD;
_silentTopicWarnDedupMax = SILENT_TOPIC_WARN_DEDUP_MAX;
_publishRateWarnDedupMax = PUBLISH_RATE_WARN_DEDUP_MAX;
_maxPresenceRef = MAX_PRESENCE_REF;
_presenceRefWarnFired = false;
_UPLOAD_PENDING_MAX_AGGREGATE = 64 * 1024 * 1024;
_pendingUploadBytes = 0;
}
/** @type {Map<string, Function>} */
const registry = new Map();
/** @type {Map<string, Function>} */
const guards = new Map();
/** @type {Set<Function>} Streams with onUnsubscribe hooks (for iterating static matches in close) */
const _streamsWithUnsubscribe = new Set();
/**
* Tag a topic function with __topicUsesCtx by inspecting its first parameter name.
*
* Auto-detects only named ctx params: ctx, context, _ctx → __topicUsesCtx = true.
* Everything else is left unset, falling back to fn.length in _callTopicFn.
*
* If fn.length is wrong (defaults, destructuring with defaults), the user must
* opt in explicitly by setting fn.__topicUsesCtx = true before registering.
* live.room already does this for its topic function.
*
* @param {Function} fn
*/
function _tagTopicFn(fn) {
try {
const src = fn.toString();
// Bare arrow: ctx => ... or context => ...
const arrow = src.match(/^\s*([\w$]+)\s*=>/);
if (arrow) {
const name = arrow[1];
if (name === 'ctx' || name === 'context' || name === '_ctx') {
/** @type {any} */ (fn).__topicUsesCtx = true;
}
return;
}
// Parenthesized: extract first token inside (...)
const paren = src.match(/\(\s*([\w$]+)/);
if (paren) {
const name = paren[1];
if (name === 'ctx' || name === 'context' || name === '_ctx') {
/** @type {any} */ (fn).__topicUsesCtx = true;
}
}
// Destructured, rest, empty, or unrecognized → leave unset
} catch {}
}
/**
* Call a topic factory function, deciding whether to inject ctx.
*
* If __topicUsesCtx was set by _tagTopicFn or explicitly, honor it.
* Otherwise fall back to fn.length vs args.length heuristic.
*
* @param {Function} fn
* @param {any} ctx
* @param {any[]} args
* @returns {any}
*/
function _callTopicFn(fn, ctx, args) {
let result;
if (fn.__topicUsesCtx === true) result = fn(ctx, ...args);
else if (fn.__topicUsesCtx === false) result = fn(...args);
else {
result = fn.length <= args.length ? fn(...args) : fn(ctx, ...args);
}
if (typeof result !== 'string') {
throw new LiveError('INVALID_REQUEST',
'Topic function must return a string, got ' + (result && typeof result === 'object' && typeof result.then === 'function' ? 'Promise (topic functions must not be async)' : typeof result)
);
}
return result;
}
/**
* Per-socket stream ownership. Maps ws -> topic -> [{fn, count}].
* Each entry tracks a logical stream subscription with its hook function and refcount.
* Used for gauge tracking, onUnsubscribe dispatch, and rollback.
* @type {WeakMap<object, Map<string, Array<{fn: Function, count: number}>>>}
*/
const _wsStreamOwners = new WeakMap();
/**
* Per-userId connection registry. Source of truth for `live.push({ userId })`
* routing. Populated by `pushHooks.open`, drained by `pushHooks.close`.
* Stores the platform alongside the ws so the wrapper can call
* `platform.request(ws, ...)` without separate threading.
*
* Last-write-wins on multi-device: a second connection by the same user
* replaces the first as the push target. Older connections still receive
* topic publishes via their own subscriptions; only push routing flips.
* Cluster-wide push (any instance routing to any user's ws) is a separate
* primitive in the extensions package.
* @type {Map<string, { ws: any, platform: any }>}
*/
const _pushRegistry = new Map();
/**
* Reverse index from ws back to its registered userId. Used by
* `pushHooks.close` to deregister without re-running identify(ws),
* which may not be reliable on close (some platforms clear userData).
* WeakMap so sockets remain GC-eligible if close is missed.
* @type {WeakMap<object, string>}
*/
const _wsToPushUserId = new WeakMap();
/** One-shot flag for the MAX_PUSH_REGISTRY warning. Reset by `_resetPushRegistry`. */
let _pushRegistryWarnFired = false;
/** One-shot flag for the TOPIC_WS_COUNTS_WARN_THRESHOLD warning. Reset by `_resetTopicWsCounts`. */
let _topicWsCountsWarnFired = false;
/** One-shot flag for the MAX_PRESENCE_REF saturation warning. Reset by `_resetCapsForTest`. */
let _presenceRefWarnFired = false;
/** @type {((ws: any) => string | null | undefined) | null} */
let _pushIdentify = null;
/**
* Default identify: read user_id then userId from ws.getUserData().
* Returns undefined for anonymous connections (skipped by pushHooks.open).
* @param {any} ws
* @returns {string | null | undefined}
*/
function _defaultPushIdentify(ws) {
let data;
try { data = ws.getUserData?.(); } catch { return undefined; }
if (!data) return undefined;
return data.user_id != null ? data.user_id : data.userId;
}
function _getPushIdentify() {
return _pushIdentify || _defaultPushIdentify;
}
/**
* Per-topic set of WebSockets currently holding at least one realtime
* stream subscription. Maintained as the source of truth for the
* `remainingSubscribers` argument the realtime layer passes to
* `__onUnsubscribe(ctx, topic, remainingSubscribers)` - apps use this
* to decide "should I tear down the upstream feed?" once the count
* hits zero. Distinct from the adapter's own ws.isSubscribed bookkeeping
* because it tracks realtime-stream subscriptions specifically (not
* arbitrary `on(topic)` topic listeners).
* @type {Map<string, Set<object>>}
*/
const _topicWsCounts = new Map();
/**
* Per-topic staleness watchdog. When a stream is configured with
* `staleAfterMs`, the realtime layer arms a timer on first subscribe
* for the topic. Every publish to the topic resets the timer (a
* publish proves the topic is live). When the timer fires, the
* realtime layer re-runs the stream's loader and broadcasts the new
* data as a `refreshed` event; the client merges it as a full-state
* replacement across every merge strategy.
*
* Captured ctx, args, fn, and platform reference are taken from the
* FIRST subscriber for the topic. Subsequent subscribers do not
* replace these (any subscriber's ctx works for a shared loader call
* since the topic identifies the data scope). When the topic's
* subscriber count drops to zero the watchdog clears; a new subscriber
* after that captures a fresh ctx.
*
* @type {Map<string, {
* timerId: ReturnType<typeof setTimeout>,
* staleAfterMs: number,
* fn: Function,
* ctx: any,
* args: any[],
* platform: any,
* onError: Function | null,
* reloading: boolean
* }>}
*/
const _topicStaleWatch = new Map();
/**
* Per-process state for the dev-mode silent-topic warning. Arms a one-shot
* timer on the first subscribe to a topic; if no event arrives within
* `thresholdMs`, logs a warning suggesting common causes (missing pg_notify
* trigger, missing handler-side publish, intentionally low-traffic topic).
*
* Hard-gated to development. Production-side cost: one boolean check on
* the publish hot path, constant-folded out by Vite/Rollup when
* `process.env.NODE_ENV === 'production'`.
*
* Shape mirrors `_topicStaleWatch` (per-topic timer + flags); the watchdog
* fires once per topic per process and dedupes via `_silentTopicWarned`
* so re-subscribes after a warn don't re-fire.
*/
const _silentTopicConfig = {
enabled: true,
thresholdMs: 30000,
/** @type {Set<string>} */
suppress: new Set()
};
/** @type {Map<string, { timerId: ReturnType<typeof setTimeout>, sawEvent: boolean }>} */
const _silentTopicWatch = new Map();
/** @type {Set<string>} Topics already warned about; prevents re-warning across re-subscribe cycles. */
const _silentTopicWarned = new Set();
/** @type {Array<(ctx: any, next: () => Promise<any>) => Promise<any>>} */
const _globalMiddleware = [];
/**
* Copy stream metadata from a source function to a wrapper.
* Single source of truth for all metadata properties - add new fields here.
* @param {any} target
* @param {any} source
*/
function _copyStreamMeta(target, source) {
target.__isStream = source.__isStream;
target.__isLive = source.__isLive;
target.__streamTopic = source.__streamTopic;
target.__streamOptions = source.__streamOptions;
if (source.__replay) target.__replay = source.__replay;
if (source.__delta) target.__delta = source.__delta;
if (source.__onSubscribe) target.__onSubscribe = source.__onSubscribe;
if (source.__onUnsubscribe) target.__onUnsubscribe = source.__onUnsubscribe;
if (source.__streamFilter) target.__streamFilter = source.__streamFilter;
if (source.__streamArgs) target.__streamArgs = source.__streamArgs;
if (source.__streamTransform) target.__streamTransform = source.__streamTransform;
if (source.__streamVolatile) target.__streamVolatile = source.__streamVolatile;
if (source.__streamVersion !== undefined) target.__streamVersion = source.__streamVersion;
if (source.__streamMigrate) target.__streamMigrate = source.__streamMigrate;
if (source.__streamStaleAfterMs !== undefined) target.__streamStaleAfterMs = source.__streamStaleAfterMs;
if (source.__streamOnError) target.__streamOnError = source.__streamOnError;
if (source.__isChannel) target.__isChannel = source.__isChannel;
if (source.__isDerived) target.__isDerived = source.__isDerived;
if (source.__derivedDynamic) {
target.__derivedDynamic = source.__derivedDynamic;
target.__derivedSourceFactory = source.__derivedSourceFactory;
target.__derivedTopicArgs = source.__derivedTopicArgs;
target.__derivedDebounce = source.__derivedDebounce;
}
if (source.__derivedSources) target.__derivedSources = source.__derivedSources;
if (source.__isGated) {
target.__isGated = true;
target.__gatePredicate = source.__gatePredicate;
}
}
/**
* Per-topic coalesce registry. When a stream registered with `coalesceBy`
* is subscribed, its topic is recorded here along with the live set of
* subscriber sockets. The publish helper uses this to decide between
* `platform.publish` (default broadcast) and per-socket
* `platform.sendCoalesced` fan-out.
*
* Hot-path cost on the default (no-coalesce) branch: one Map.get on an
* almost-always-empty map. See bench/publish.js for numbers.
*
* @type {Map<string, { coalesceBy: Function, onError: Function | null, ws: Set<any> }>}
*/
const _topicCoalesce = new Map();
/** @internal Exported only so tests can drive the coalesce registry without a full subscribe round-trip. */
export function _registerCoalesce(ws, topic, coalesceBy, onError) {
let entry = _topicCoalesce.get(topic);
if (!entry) {
entry = { coalesceBy, onError: onError || null, ws: new Set() };
_topicCoalesce.set(topic, entry);
}
entry.ws.add(ws);
}
function _unregisterCoalesce(ws, topic) {
const entry = _topicCoalesce.get(topic);
if (!entry) return;
entry.ws.delete(ws);
if (entry.ws.size === 0) _topicCoalesce.delete(topic);
}
/**
* Per-topic transform registry. When a stream registered with `transform`
* is subscribed, its topic is recorded here. The publish helper applies
* the transform once per publish, BEFORE platform.publish (or the
* sendCoalesced fan-out), so subscribers see the projected wire shape.
*
* Refcounted by ws-topic contributions - evicted when the last
* subscriber leaves so HMR-changed stream definitions can re-register.
*
* Each entry also carries the registering stream's `onError` reference
* (if configured). The publish helper wraps the transform call in
* try/catch and routes throws to that observer; without an observer,
* the throw propagates as before. First subscriber-for-topic wins on
* `onError` selection (same rule as for `transform` itself).
*
* @type {Map<string, { transform: Function, onError: Function | null, refcount: number }>}
*/
const _topicTransform = new Map();
/** Per-ws set of topics where this ws has contributed a transform refcount.
* Lets the unregister side be idempotent and ws-aware. */
const _wsTransformContrib = new WeakMap();
function _registerTransform(ws, topic, transform, onError) {
let entry = _topicTransform.get(topic);
if (!entry) {
entry = { transform, onError: onError || null, refcount: 0 };
_topicTransform.set(topic, entry);
}
entry.refcount++;
let contrib = _wsTransformContrib.get(ws);
if (!contrib) { contrib = new Set(); _wsTransformContrib.set(ws, contrib); }
contrib.add(topic);
}
function _unregisterTransform(ws, topic) {
const contrib = _wsTransformContrib.get(ws);
if (!contrib || !contrib.has(topic)) return;
contrib.delete(topic);
const entry = _topicTransform.get(topic);
if (!entry) return;
entry.refcount--;
if (entry.refcount <= 0) _topicTransform.delete(topic);
}
/**
* Reset the per-topic transform registry. Tests only.
* @internal
*/
export function _resetTransformRegistry() {
_topicTransform.clear();
}
/**
* Per-topic volatile registry. When a stream registered with `volatile: true`
* is subscribed, its topic is recorded here. The publish helper translates
* `volatile` topics + per-call `options.volatile === true` into the adapter's
* `seq: false` per-event option, so seq stamping is skipped for these
* messages - a reconnect carrying `lastSeenSeq` won't try to backfill them.
*
* Wire-level "drop on backpressure" behavior is the adapter's job:
* platform.publish / platform.publishBatched / platform.send all skip a
* subscriber whose outbound buffer is over the configured maxBackpressure
* threshold (default 64 KB). This registry only governs seq stamping and
* intent declaration on the realtime side.
*
* Refcounted by ws-topic contributions, mirroring the transform registry.
* @type {Map<string, { refcount: number }>}
*/
const _topicVolatile = new Map();
/** Per-ws set of topics where this ws has contributed a volatile refcount. */
const _wsVolatileContrib = new WeakMap();
/** @internal Exported only so tests can drive the volatile registry without a full subscribe round-trip. */
export function _registerVolatile(ws, topic) {
let entry = _topicVolatile.get(topic);
if (!entry) {
entry = { refcount: 0 };
_topicVolatile.set(topic, entry);
}
entry.refcount++;
let contrib = _wsVolatileContrib.get(ws);
if (!contrib) { contrib = new Set(); _wsVolatileContrib.set(ws, contrib); }
contrib.add(topic);
}
function _unregisterVolatile(ws, topic) {
const contrib = _wsVolatileContrib.get(ws);
if (!contrib || !contrib.has(topic)) return;
contrib.delete(topic);
const entry = _topicVolatile.get(topic);
if (!entry) return;
entry.refcount--;
if (entry.refcount <= 0) _topicVolatile.delete(topic);
}
/** Reset the per-topic volatile registry. Tests only. @internal */
export function _resetVolatileRegistry() {
_topicVolatile.clear();
}
/**
* Per-topic invalidation registry. When a stream is registered with
* `invalidateOn: '<pattern>'`, an entry is created here keyed by the
* pattern's compiled regex. The publish helper checks every publish
* against these patterns and triggers a loader re-run for any stream
* whose pattern matches the publish topic. Distinct from the stale
* watchdog (timer-driven) - this one is event-driven.
*
* Each watcher stores everything `_staleReload` needs to re-execute the
* loader: stream topic, init fn (with stashed __streamTransform /
* __streamOnError), captured ctx + args, and the platform reference
* for the publish path.
*
* @type {Map<string, { regex: RegExp, watchers: Array<{
* topic: string,
* fn: any,
* ctx: any,
* args: any[],
* platform: any,
* onError: Function | null,
* reloading: boolean
* }> }>}
*/
const _topicInvalidationWatch = new Map();
/**
* Convert an `invalidateOn` pattern into a fast matcher. `*` matches any
* sequence of one or more characters (greedy, including colons); other
* regex specials are escaped. Patterns must be non-empty strings.
*
* Returns the compiled regex along with the literal-prefix and `prefixOnly`
* flag so the publish hot path can short-circuit without entering the regex
* engine for the common `prefix*` shape.
*
* @param {string} pattern
* @returns {{ regex: RegExp, prefix: string, prefixOnly: boolean }}
*/
function _compileInvalidatePattern(pattern) {
const escaped = pattern.replace(/[.+?^${}()|[\]\\]/g, '\\$&').replace(/\*/g, '.+');
const regex = new RegExp('^' + escaped + '$');
const firstStar = pattern.indexOf('*');
const prefix = firstStar === -1 ? pattern : pattern.slice(0, firstStar);
const lastStar = pattern.lastIndexOf('*');
const prefixOnly = firstStar !== -1 && lastStar === pattern.length - 1 && firstStar === lastStar;
return { regex, prefix, prefixOnly };
}
/**
* Register the invalidation watchers for a stream's first subscriber.
* No-op when `fn.__streamInvalidateOn` is unset. Idempotent per
* (pattern, topic) pair: re-subscribing the same stream won't
* accumulate duplicate watchers.
*
* @param {string} topic
* @param {any} fn
* @param {any} ctx
* @param {any[]} args
* @param {any} platform
*/
function _registerInvalidationWatch(topic, fn, ctx, args, platform) {
const patterns = fn.__streamInvalidateOn;
if (!patterns) return;
const onError = fn.__streamOnError || null;
for (const p of patterns) {
let entry = _topicInvalidationWatch.get(p);
if (!entry) {
const { regex, prefix, prefixOnly } = _compileInvalidatePattern(p);
entry = { regex, prefix, prefixOnly, watchers: [] };
_topicInvalidationWatch.set(p, entry);
}
if (entry.watchers.some(w => w.topic === topic && w.fn === fn)) continue;
entry.watchers.push({ topic, fn, ctx, args, platform, onError, reloading: false });
}
}
/**
* Drop a stream's invalidation watchers when its last subscriber
* leaves. Removes empty pattern entries so the size-zero fast path in
* the publish helper short-circuits cleanly.
*
* @param {string} topic
* @param {any} fn
*/
function _unregisterInvalidationWatch(topic, fn) {
const patterns = fn.__streamInvalidateOn;
if (!patterns) return;
for (const p of patterns) {
const entry = _topicInvalidationWatch.get(p);
if (!entry) continue;
const idx = entry.watchers.findIndex(w => w.topic === topic && w.fn === fn);
if (idx >= 0) entry.watchers.splice(idx, 1);
if (entry.watchers.length === 0) _topicInvalidationWatch.delete(p);
}
}
/** Reset the per-topic invalidation registry. Tests only. @internal */
export function _resetInvalidationWatch() {
_topicInvalidationWatch.clear();
}
/**
* Re-run a stream's loader because an invalidation pattern matched.
* Mirrors `_staleReload` - captures the same ctx+args, applies the
* init transform, and broadcasts the result as a `refreshed` event on
* the stream's own topic. Concurrent triggers are deduped via the
* `reloading` flag (we never queue or merge them; the next match after
* the in-flight reload completes will re-run regardless).
*
* @param {{ topic: string, fn: any, ctx: any, args: any[], platform: any, onError: Function | null, reloading: boolean }} watcher
*/
async function _invalidationReload(watcher) {
if (watcher.reloading) return;
watcher.reloading = true;
try {
const result = await watcher.fn(watcher.ctx, ...watcher.args);
const initTransform = /** @type {any} */ (watcher.fn).__streamTransform;
const finalData = (initTransform && result != null) ? _applyInitTransform(initTransform, result) : result;
try { watcher.platform.publish(watcher.topic, 'refreshed', finalData); } catch {}
} catch (err) {
if (watcher.onError) {
try { await watcher.onError(err, watcher.ctx, watcher.topic); } catch {}
}
} finally {
watcher.reloading = false;
}
}
/**
* Arm the per-topic stale watchdog if `fn` declares `__streamStaleAfterMs`.
* Idempotent per topic: if a watchdog already exists (from an earlier
* subscriber), this is a no-op so we don't replace the captured ctx.
*
* @param {string} topic
* @param {any} fn The stream's init function (with __streamStaleAfterMs / __streamOnError stashed)
* @param {any} ctx
* @param {any[]} args
* @param {any} platform
*/
function _registerStaleWatch(topic, fn, ctx, args, platform) {
const staleMs = fn.__streamStaleAfterMs;
if (!staleMs) return;
if (_topicStaleWatch.has(topic)) return;
const entry = {
staleAfterMs: staleMs,
fn,
ctx,
args,
platform,
onError: fn.__streamOnError || null,
reloading: false,
timerId: setTimeout(() => _staleReload(topic), staleMs)
};
_topicStaleWatch.set(topic, entry);
}
/**
* Clear the per-topic stale watchdog. Called when the last subscriber for
* the topic leaves. No-op when no watchdog exists.
* @param {string} topic
*/
function _unregisterStaleWatch(topic) {
const entry = _topicStaleWatch.get(topic);
if (!entry) return;
clearTimeout(entry.timerId);
_topicStaleWatch.delete(topic);
}
/**
* Reset the watchdog timer for a topic. Called from the publish helper on
* every publish to that topic - a publish proves the topic is live, so
* the staleness clock restarts.
* @param {string} topic
*/
function _resetStaleTimer(topic) {
const entry = _topicStaleWatch.get(topic);
if (!entry) return;
clearTimeout(entry.timerId);
entry.timerId = setTimeout(() => _staleReload(topic), entry.staleAfterMs);
}
/**
* Run the stale-reload for a topic: re-invoke the stream's loader with the
* captured ctx + args, broadcast the new data as a `refreshed` event, and
* re-arm the timer. Loader throws are routed to the stream's onError if
* configured; the timer always re-arms so transient failures do not leave
* the topic in a permanently-stale state.
*
* @param {string} topic
*/
async function _staleReload(topic) {
const entry = _topicStaleWatch.get(topic);
if (!entry) return;
if (entry.reloading) return;
entry.reloading = true;
try {
const result = await entry.fn(entry.ctx, ...entry.args);
const initTransform = /** @type {any} */ (entry.fn).__streamTransform;
const finalData = (initTransform && result != null) ? _applyInitTransform(initTransform, result) : result;
try { entry.platform.publish(topic, 'refreshed', finalData); } catch {}
} catch (err) {
if (entry.onError) {
try { await entry.onError(err, entry.ctx, topic); } catch {}
}
} finally {
entry.reloading = false;
// Re-arm only if this entry is still registered. The last
// subscriber may have left during the async loader call, in
// which case _unregisterStaleWatch has already cleared the
// timer slot.
const stillThere = _topicStaleWatch.get(topic);
if (stillThere === entry) {
stillThere.timerId = setTimeout(() => _staleReload(topic), entry.staleAfterMs);
}
}
}
/** Reset the per-topic stale-watch registry. Tests only. @internal */
export function _resetStaleWatch() {
for (const entry of _topicStaleWatch.values()) clearTimeout(entry.timerId);
_topicStaleWatch.clear();
}
/**
* Arm the silent-topic watchdog for a topic on its first subscriber.
* Skips system topics (`__`-prefixed: `__realtime`, `__signal:*`, etc.)
* which are intentionally quiet until something publishes. Skips topics
* the user has explicitly suppressed and topics already warned about.
* Idempotent per topic; second-sub-on-same-topic is a no-op.
*
* @param {string} topic
*/
export function _armSilentTopicWatch(topic) {
if (!_IS_DEV) return;
if (!_silentTopicConfig.enabled) return;
if (_silentTopicWatch.has(topic)) return;
if (_silentTopicWarned.has(topic)) return;
if (_silentTopicConfig.suppress.has(topic)) return;
if (topic.charCodeAt(0) === 95 && topic.charCodeAt(1) === 95) return;
const entry = {
sawEvent: false,
timerId: setTimeout(() => {
const e = _silentTopicWatch.get(topic);
if (!e || e.sawEvent) return;
if (_silentTopicWarned.size >= _silentTopicWarnDedupMax && !_silentTopicWarned.has(topic)) {
const oldest = _silentTopicWarned.values().next().value;
if (oldest !== undefined) _silentTopicWarned.delete(oldest);
}
_silentTopicWarned.add(topic);
console.warn(
"[svelte-realtime] Topic '" + topic + "' has subscribers but no events arrived within " +
_silentTopicConfig.thresholdMs + "ms.\n" +
" Common causes:\n" +
" - missing pg_notify trigger on the underlying table\n" +
" - no ctx.publish() call in the relevant handler\n" +
" - intentionally low-traffic topic (extend threshold or suppress)\n" +
" Configure: live.silentTopicWarning({ thresholdMs: 60000 })\n" +
" Suppress: live.silentTopicWarning({ suppress: ['" + topic + "'] })\n" +
" Disable: live.silentTopicWarning(false)\n" +
" See: https://svti.me/silent-topic"
);
}, _silentTopicConfig.thresholdMs)
};
if (typeof (/** @type {any} */ (entry.timerId).unref) === 'function') {
/** @type {any} */ (entry.timerId).unref();
}
_silentTopicWatch.set(topic, entry);
}
/**
* Mark a topic as having seen an event. Called from the publish-helper
* closure on every publish. Once an event arrives, the watchdog is
* disarmed for that topic for the lifetime of the process (the warning
* never fires for a topic that has been live).
*
* @param {string} topic
*/
function _observeSilentTopicPublish(topic) {
const entry = _silentTopicWatch.get(topic);
if (!entry) return;
entry.sawEvent = true;
clearTimeout(entry.timerId);
_silentTopicWatch.delete(topic);
}
/**
* Clear the silent-topic watchdog when the last subscriber leaves the
* topic. Mirrors `_unregisterStaleWatch`. No-op when no watchdog exists.
*
* @param {string} topic
*/
function _disarmSilentTopicWatch(topic) {
const entry = _silentTopicWatch.get(topic);
if (!entry) return;
clearTimeout(entry.timerId);
_silentTopicWatch.delete(topic);
}
/** Reset the silent-topic watchdog state. Tests only. @internal */
export function _resetSilentTopicWarning() {
for (const entry of _silentTopicWatch.values()) clearTimeout(entry.timerId);
_silentTopicWatch.clear();
_silentTopicWarned.clear();
_silentTopicConfig.enabled = true;
_silentTopicConfig.thresholdMs = 30000;
_silentTopicConfig.suppress.clear();
}
/**
* Apply a transform function to initial-load data. Per-item for arrays
* (covers crud/latest/presence/cursor merge), whole-value for
* non-arrays (covers set merge).
* @param {Function} transform
* @param {any} data
* @returns {any}
*/
function _applyInitTransform(transform, data) {
if (Array.isArray(data)) {
const out = new Array(data.length);
for (let i = 0; i < data.length; i++) out[i] = transform(data[i]);
return out;
}
return transform(data);
}
/** @type {WeakMap<any, { publish: Function, publishThrottled: Function, publishDebounced: Function, throttle: Function, debounce: Function, signal: Function, batch: Function, shed: Function, skip: Function }>} */
const _ctxHelpersCache = new WeakMap();
/** @type {boolean} */
const _IS_DEV = typeof process !== 'undefined' && process.env?.NODE_ENV !== 'production';
/** Dev-warn dedup: one-time "ctx.throttle is deprecated" warning. */
let _throttleDeprecatedWarned = false;
/** Dev-warn dedup: one-time "ctx.debounce is deprecated" warning. */
let _debounceDeprecatedWarned = false;
/** Dev-warn dedup: per-helper bad-args warning. Keys: 'publishThrottled', 'publishDebounced', 'throttle', 'debounce'. */
/** @type {Record<string, boolean>} */
const _publishHelperBadArgsWarned = Object.create(null);
/** Dev-warn dedup: one-time "ctx.skip gate map at capacity" warning. */
let _skipGateCapWarned = false;
/**
* Topics declared with `live.stream(..., { replay: true })`. Populated at
* declaration time for static topics and at first-subscribe time for
* dynamic-topic factories (when the topic resolves). When a publish to one
* of these topics happens (from `ctx.publish`, cron auto-publish, derived /
* aggregate, anywhere), the framework auto-routes through
* `platform.replay.publish(...)` so the buffer captures it for gap-fill on
* resume -- regardless of which seam (RPC / cron / etc.) the publisher
* sits on. Pre-fix, only publishes that flowed through a user-managed
* `wrapWithReplay` proxy reached the buffer; cron-published events bypassed
* it silently because the cron platform was captured separately.
*
* @type {Set<string>}
*/
const _replayEligibleTopics = new Set();
/**
* Dev-warn dedup for "stream declared replay: true but the adapter does
* not expose `platform.replay`" misconfigurations. Per-topic so each
* misconfigured topic surfaces once; the warning includes the install
* pointer for the replay extension. Cleared by `_resetReplayRouting()`
* for tests.
*
* @type {Set<string>}
*/
const _replayMissingWarned = new Set();
/**
* Public well-known marker: a user-managed platform proxy that already
* routes replay-eligible publishes through `platform.replay.publish(...)`
* itself can opt out of the framework's auto-routing by setting this
* symbol-keyed property to `true`. Without the marker, the framework's
* `_publish` would call `platform.replay.publish(platform, ...)`, which
* internally calls `platform.publish(topic, event, data)` -- the user
* proxy's intercept would then call `replay.publish(target, ...)` again,
* doubling the Redis write. The marker lets the framework defer to the
* user proxy in that case.
*
* Most users should drop their bespoke `wrapWithReplay` proxy and let the
* framework own routing; the marker is a back-compat escape hatch.
*/
export const WRAPPED_FOR_REPLAY = Symbol.for('svelte-realtime.wrapped-for-replay');
/**
* Register a topic as replay-eligible. Called from `live.stream` declaration
* for static topics, and from `_executeStreamRpc` for dynamic topics on first
* subscribe. Idempotent; a topic registered twice stays in the set once.
*
* @param {string} topic
*/
function _registerReplayTopic(topic) {
if (typeof topic === 'string' && topic.length > 0) {
_replayEligibleTopics.add(topic);
}
}
/**
* Replay-route a publish through `platform.replay.publish(...)` when the
* topic is in the replay-eligible registry AND the platform exposes a
* `replay` surface. Returns `true` when the publish was routed (caller
* should NOT fall back to `platform.publish` -- replay.publish handles the
* local broadcast internally) and `false` when it was not (caller should
* call its own publish path: bare `platform.publish`, batched, or
* coalesced).
*
* Skips routing when:
* - The topic is not registered as replay-eligible.
* - The adapter exposes no `platform.replay` (replay extension not
* installed). A one-time dev warn fires per topic so the misconfig is
* visible during development.
* - The platform is marked `[WRAPPED_FOR_REPLAY] = true` (a user-managed
* proxy is already routing replay; framework defers).
*
* Failures inside `replay.publish` are logged in dev and otherwise
* swallowed -- the local broadcast still happens via the extension's own
* fallback, and we don't want a Redis hiccup to crash the publisher.
*
* @param {any} platform
* @param {string} topic
* @param {string} event
* @param {any} data
* @returns {boolean} true if routed through replay, false if caller should fall back
*/
function _maybeReplayPublish(platform, topic, event, data) {
if (!_replayEligibleTopics.has(topic)) return false;
const replay = platform && /** @type {any} */ (platform).replay;
if (!replay || typeof replay.publish !== 'function') {
if (_IS_DEV && !_replayMissingWarned.has(topic)) {
_replayMissingWarned.add(topic);
console.warn(
"[svelte-realtime] live.stream('" + topic + "', ..., { replay: true }) is declared but " +
"the adapter exposes no `platform.replay` -- the bounded replay buffer is not engaged " +
"and clients will not receive missed events on resume. Install the replay extension " +
"(svelte-adapter-uws-extensions/redis/replay or postgres/replay) and wire it via " +
"`platform.replay = createReplay(redisClient)` so `platform.replay` is exposed. " +
"Warned once per topic per session."
);
}
return false;
}
if (/** @type {any} */ (platform)[WRAPPED_FOR_REPLAY]) return false;
try {
const ret = replay.publish(platform, topic, event, data);
if (ret && typeof ret.then === 'function') {
ret.catch((err) => {
if (_IS_DEV) {
console.warn(
"[svelte-realtime] replay.publish('" + topic + "') failed:",
err
);
}
});
}
} catch (err) {
if (_IS_DEV) {
console.warn(
"[svelte-realtime] replay.publish('" + topic + "') threw synchronously:",
err
);
}
// On sync throw the local broadcast didn't happen; fall back to
// platform.publish so subscribers still receive the event live.
return false;
}
return true;
}
/**
* Reset replay-routing state. Tests only.
* @internal
*/
export function _resetReplayRouting() {
_replayEligibleTopics.clear();
_replayMissingWarned.clear();
}
/**
* Per-process state for the dev-mode publish-rate warning. Sampler is lazy:
* activated on the first ctx-helpers cache miss per platform, runs at the
* configured interval, reads `platform.pressure.topPublishers` (already
* computed by the adapter sampler), and emits one warn per topic per
* process when a topic is over threshold. Production has zero cost --
* `_IS_DEV` is constant-folded so the activation branch is dead code.
*/
const _publishRateConfig = {
enabled: true,
threshold: 200,
intervalMs: 5000
};
/** @type {Set<string>} */
const _publishRateWarned = new Set();
/** @type {WeakMap<any, ReturnType<typeof setInterval>>} */
const _publishRateSamplers = new WeakMap();
/**
* Bumped by `_resetPublishRateWarning` and by `live.publishRateWarning(false)`.
* Each sampler captures its activation-time epoch and self-clears on the next
* fire when the epoch no longer matches. Pattern used in place of the prior
* strong-reference `Set<platform>` because that Set held every dev-mode
* platform alive across the process lifetime, defeating the WeakMap above and