1/**
2 * Stores and runs atoms for one reactive runtime.
3 *
4 * An `AtomRegistry` evaluates atoms, caches their current values, tracks
5 * dependencies, applies writes and refreshes, manages subscriptions, and
6 * disposes unused nodes. Each registry is independent, so the same atom can hold
7 * different values in different registries. Serializable atom values can also be
8 * preloaded before the first read.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import * as Context from "../Context.ts"
14import * as Effect from "../Effect.ts"
15import * as Exit from "../Exit.ts"
16import * as Fiber from "../Fiber.ts"
17import { constVoid, dual } from "../Function.ts"
18import * as Layer from "../Layer.ts"
19import * as Option from "../Option.ts"
20import { hasProperty } from "../Predicate.ts"
21import * as Queue from "../Queue.ts"
22import type { Scheduler, SchedulerDispatcher } from "../Scheduler.ts"
23import { MixedScheduler } from "../Scheduler.ts"
24import * as Scope from "../Scope.ts"
25import * as Stream from "../Stream.ts"
26import * as Result from "./AsyncResult.ts"
27import type * as Atom from "./Atom.ts"
28
29/**
30 * The literal type used to identify `AtomRegistry` services and values.
31 *
32 * @stability unstable
33 * @category type IDs
34 * @since 4.0.0
35 */
36export type TypeId = "~effect/reactivity/AtomRegistry"
37
38/**
39 * The runtime type id used to identify `AtomRegistry` services and values.
40 *
41 * @stability unstable
42 * @category type IDs
43 * @since 4.0.0
44 */
45export const TypeId: TypeId = "~effect/reactivity/AtomRegistry"
46
47/**
48 * Returns `true` when the value has the `AtomRegistry` type id.
49 *
50 * @stability unstable
51 * @category guards
52 * @since 4.0.0
53 */
54export const isAtomRegistry = (u: unknown): u is AtomRegistry => hasProperty(u, TypeId)
55
56/**
57 * The runtime registry that stores atom nodes and coordinates reads, writes,
58 * refreshes, subscriptions, and disposal.
59 *
60 * **Details**
61 *
62 * It also manages scheduler configuration, serializable preloaded values, and node
63 * addition/removal callbacks.
64 *
65 * @stability unstable
66 * @category models
67 * @since 4.0.0
68 */
69export interface AtomRegistry {
70 readonly [TypeId]: TypeId
71 readonly scheduler: Scheduler
72 readonly schedulerAsync: Scheduler
73 readonly getNodes: () => ReadonlyMap<Atom.Atom<any> | string, Node<any>>
74 readonly get: <A>(atom: Atom.Atom<A>) => A
75 readonly mount: <A>(atom: Atom.Atom<A>) => () => void
76 readonly refresh: <A>(atom: Atom.Atom<A>) => void
77 readonly set: <R, W>(atom: Atom.Writable<R, W>, value: W) => void
78 readonly setSerializable: (key: string, encoded: unknown) => void
79 readonly modify: <R, W, A>(atom: Atom.Writable<R, W>, f: (_: R) => [returnValue: A, nextValue: W]) => A
80 readonly update: <R, W>(atom: Atom.Writable<R, W>, f: (_: R) => W) => void
81 readonly subscribe: <A>(atom: Atom.Atom<A>, f: (_: A) => void, options?: {
82 readonly immediate?: boolean
83 }) => () => void
84 readonly reset: () => void
85 readonly dispose: () => void
86 onNodeAdded?: ((node: Node<any>) => void) | undefined
87 onNodeRemoved?: ((node: Node<any>) => void) | undefined
88}
89
90/**
91 * A registry node for a single atom.
92 *
93 * **Details**
94 *
95 * Nodes expose the current value, parent and child dependency links, listener set,
96 * and current lifecycle state.
97 *
98 * @stability unstable
99 * @category models
100 * @since 4.0.0
101 */
102export interface Node<A> {
103 readonly atom: Atom.Atom<A>
104 readonly value: () => A
105 parents: Set<Node<any>>
106 children: Set<Node<any>>
107 listeners: Set<() => void>
108 currentState(): "uninitialized" | "stale" | "valid" | "removed"
109}
110
111/**
112 * Creates an `AtomRegistry`.
113 *
114 * **Details**
115 *
116 * Options can preload initial atom values, provide a custom task scheduler,
117 * configure timeout bucket resolution, and set a default idle time-to-live for
118 * unused atoms.
119 *
120 * @stability unstable
121 * @category constructors
122 * @since 4.0.0
123 */
124export const make = (
125 options?: {
126 readonly initialValues?: Iterable<readonly [Atom.Atom<any>, any]> | undefined
127 readonly scheduleTask?: ((f: () => void) => () => void) | undefined
128 readonly timeoutResolution?: number | undefined
129 readonly defaultIdleTTL?: number | undefined
130 } | undefined
131): AtomRegistry =>
132 new RegistryImpl(
133 options?.initialValues,
134 options?.scheduleTask,
135 options?.timeoutResolution,
136 options?.defaultIdleTTL
137 )
138
139/**
140 * Service tag for the active atom runtime cache.
141 *
142 * **When to use**
143 *
144 * Use to access or provide the registry that stores atom values,
145 * dependencies, subscriptions, and disposal state for a reactive lifetime.
146 *
147 * @stability unstable
148 * @category services
149 * @since 4.0.0
150 */
151export const AtomRegistry = Context.Service<AtomRegistry>(TypeId)
152
153/**
154 * Creates a layer that provides an `AtomRegistry` configured with the supplied
155 * options.
156 *
157 * **Details**
158 *
159 * The registry is disposed when the layer scope is finalized.
160 *
161 * @stability unstable
162 * @category layers
163 * @since 4.0.0
164 */
165export const layerOptions = (options?: {
166 readonly initialValues?: Iterable<readonly [Atom.Atom<any>, any]> | undefined
167 readonly scheduleTask?: ((f: () => void) => () => void) | undefined
168 readonly timeoutResolution?: number | undefined
169 readonly defaultIdleTTL?: number | undefined
170}): Layer.Layer<AtomRegistry> =>
171 Layer.effect(
172 AtomRegistry,
173 Effect.gen(function*() {
174 const scope = yield* Effect.scope
175 const registry = make({
176 ...options,
177 scheduleTask: options?.scheduleTask
178 })
179 yield* Scope.addFinalizer(scope, Effect.sync(() => registry.dispose()))
180 return registry
181 })
182 )
183
184/**
185 * The default layer that provides a fresh `AtomRegistry`.
186 *
187 * @stability unstable
188 * @category layers
189 * @since 4.0.0
190 */
191export const layer: Layer.Layer<AtomRegistry> = layerOptions()
192
193// -----------------------------------------------------------------------------
194// conversions
195// -----------------------------------------------------------------------------
196
197/**
198 * Converts an atom in this registry into a stream.
199 *
200 * **Details**
201 *
202 * The stream emits the current value immediately, emits subsequent changes, and
203 * unsubscribes from the registry when the stream scope closes.
204 *
205 * @stability unstable
206 * @category converting
207 * @since 4.0.0
208 */
209export const toStream: {
210 <A>(atom: Atom.Atom<A>): (self: AtomRegistry) => Stream.Stream<A>
211 <A>(self: AtomRegistry, atom: Atom.Atom<A>): Stream.Stream<A>
212} = dual(
213 2,
214 <A>(self: AtomRegistry, atom: Atom.Atom<A>) =>
215 Stream.callback<A>((queue) =>
216 Effect.suspend(() => {
217 const fiber = Fiber.getCurrent()!
218 const scope = Context.getUnsafe(fiber.context, Scope.Scope)
219 const cancel = self.subscribe(atom, (value) => Queue.offerUnsafe(queue, value), {
220 immediate: true
221 })
222 return Scope.addFinalizer(scope, Effect.sync(cancel))
223 })
224 )
225)
226
227/**
228 * Converts an `AsyncResult` atom in this registry into a stream of successful
229 * values.
230 *
231 * **Details**
232 *
233 * Initial results are skipped, failures fail the stream with their cause, and
234 * duplicate stream values are dropped with `Stream.changes`.
235 *
236 * @stability unstable
237 * @category converting
238 * @since 4.0.0
239 */
240export const toStreamResult: {
241 <A, E>(atom: Atom.Atom<Result.AsyncResult<A, E>>): (self: AtomRegistry) => Stream.Stream<A, E>
242 <A, E>(self: AtomRegistry, atom: Atom.Atom<Result.AsyncResult<A, E>>): Stream.Stream<A, E>
243} = dual(
244 2,
245 <A, E>(self: AtomRegistry, atom: Atom.Atom<Result.AsyncResult<A, E>>): Stream.Stream<A, E> =>
246 toStream(self, atom).pipe(
247 Stream.filter(Result.isNotInitial),
248 Stream.mapEffect((result) =>
249 result._tag === "Success" ? Effect.succeed(result.value) : Effect.failCause(result.cause)
250 ),
251 Stream.changes
252 )
253)
254
255/**
256 * Reads an `AsyncResult` atom from this registry as an effect.
257 *
258 * **Details**
259 *
260 * The effect waits for the result to leave `Initial`, and also waits through
261 * waiting results when `suspendOnWaiting` is enabled.
262 *
263 * @stability unstable
264 * @category converting
265 * @since 4.0.0
266 */
267export const getResult: {
268 <A, E>(atom: Atom.Atom<Result.AsyncResult<A, E>>, options?: {
269 readonly suspendOnWaiting?: boolean | undefined
270 }): (self: AtomRegistry) => Effect.Effect<A, E>
271 <A, E>(self: AtomRegistry, atom: Atom.Atom<Result.AsyncResult<A, E>>, options?: {
272 readonly suspendOnWaiting?: boolean | undefined
273 }): Effect.Effect<A, E>
274} = dual(
275 (args) => isAtomRegistry(args[0]),
276 <A, E>(self: AtomRegistry, atom: Atom.Atom<Result.AsyncResult<A, E>>, options?: {
277 readonly suspendOnWaiting?: boolean | undefined
278 }): Effect.Effect<A, E> => {
279 const suspendOnWaiting = options?.suspendOnWaiting ?? false
280 return Effect.callback((resume) => {
281 const result = self.get(atom)
282 if (result._tag !== "Initial" && !(suspendOnWaiting && result.waiting)) {
283 return resume(Result.toExit(result) as any)
284 }
285 const cancel = self.subscribe(atom, (value) => {
286 if (value._tag !== "Initial" && !(suspendOnWaiting && value.waiting)) {
287 resume(Result.toExit(value) as any)
288 cancel()
289 }
290 })
291 return Effect.sync(cancel)
292 })
293 }
294)
295
296/**
297 * Mounts an atom in this registry for the lifetime of the current scope.
298 *
299 * **Details**
300 *
301 * The atom is subscribed with a no-op listener and the subscription is released
302 * when the scope finalizer runs.
303 *
304 * @stability unstable
305 * @category converting
306 * @since 4.0.0
307 */
308export const mount: {
309 <A>(atom: Atom.Atom<A>): (self: AtomRegistry) => Effect.Effect<void, never, Scope.Scope>
310 <A>(self: AtomRegistry, atom: Atom.Atom<A>): Effect.Effect<void, never, Scope.Scope>
311} = dual(
312 2,
313 <A>(self: AtomRegistry, atom: Atom.Atom<A>) =>
314 Effect.acquireRelease(
315 Effect.sync(() => self.mount(atom)),
316 (release) => Effect.sync(release)
317 )
318)
319
320// -----------------------------------------------------------------------------
321// internal
322// -----------------------------------------------------------------------------
323
324const constImmediate = { immediate: true }
325
326const invoke = (f: () => void): void => {
327 f()
328}
329
330const noFailure = {}
331
332// Runs `f` for every item, then rethrows the first failure. One failing step
333// must not skip the others: those would leak a lifetime or miss an update.
334function runAll<T>(items: Iterable<T>, f: (item: T) => void): void {
335 let failure: unknown = noFailure
336 for (const item of items) {
337 try {
338 f(item)
339 } catch (error) {
340 if (failure === noFailure) failure = error
341 }
342 }
343 if (failure !== noFailure) throw failure
344}
345
346const SerializableTypeId: Atom.SerializableTypeId = "~effect-atom/atom/Atom/Serializable"
347const atomKey = <A>(atom: Atom.Atom<A>): Atom.Atom<A> | string =>
348 SerializableTypeId in atom ? (atom as Atom.Serializable<any>)[SerializableTypeId].key : atom
349
350class RegistryImpl implements AtomRegistry {
351 readonly [TypeId]: TypeId
352 readonly timeoutResolution: number
353 readonly defaultIdleTTL: number | undefined
354 readonly scheduler: Scheduler
355 readonly schedulerAsync: Scheduler
356 readonly dispatcher: SchedulerDispatcher
357 onNodeAdded?: ((node: Node<any>) => void) | undefined
358 onNodeRemoved?: ((node: Node<any>) => void) | undefined
359
360 constructor(
361 initialValues?: Iterable<readonly [Atom.Atom<any>, any]>,
362 scheduleTask?: (cb: () => void) => () => void,
363 timeoutResolution?: number,
364 defaultIdleTTL?: number
365 ) {
366 this[TypeId] = TypeId
367 this.scheduler = new MixedScheduler("sync", scheduleTask)
368 this.schedulerAsync = new MixedScheduler("async", scheduleTask)
369 this.dispatcher = this.schedulerAsync.makeDispatcher()
370 this.defaultIdleTTL = defaultIdleTTL
371
372 if (timeoutResolution === undefined && defaultIdleTTL !== undefined) {
373 this.timeoutResolution = Math.round(defaultIdleTTL / 2)
374 } else {
375 this.timeoutResolution = timeoutResolution ?? 1000
376 }
377 if (initialValues !== undefined) {
378 for (const [atom, value] of initialValues) {
379 let target = atom
380 while (target.initialValueTarget) {
381 target = target.initialValueTarget
382 }
383 this.ensureNode(target).setInitialValue(value)
384 }
385 }
386 }
387
388 readonly nodes = new Map<Atom.Atom<any> | string, NodeImpl<any>>()
389 readonly preloadedSerializable = new Map<string, unknown>()
390 readonly timeoutBuckets = new Map<number, readonly [nodes: Set<NodeImpl<any>>, handle: number]>()
391 readonly nodeTimeoutBucket = new Map<NodeImpl<any>, number>()
392 disposed = false
393
394 getNodes() {
395 return this.nodes
396 }
397
398 get<A>(atom: Atom.Atom<A>): A {
399 return this.ensureNode(atom).value()
400 }
401
402 set<R, W>(atom: Atom.Writable<R, W>, value: W): void {
403 atom.write(this.ensureNode(atom).writeContext, value)
404 }
405
406 setSerializable(key: string, encoded: unknown): void {
407 this.preloadedSerializable.set(key, encoded)
408 }
409
410 modify<R, W, A>(atom: Atom.Writable<R, W>, f: (_: R) => [returnValue: A, nextValue: W]): A {
411 const node = this.ensureNode(atom)
412 const result = f(node.value())
413 atom.write(node.writeContext, result[1])
414 return result[0]
415 }
416
417 update<R, W>(atom: Atom.Writable<R, W>, f: (_: R) => W): void {
418 const node = this.ensureNode(atom)
419 atom.write(node.writeContext, f(node.value()))
420 }
421
422 refresh = <A>(atom: Atom.Atom<A>): void => {
423 if (atom.refresh !== undefined) {
424 atom.refresh(this.refresh)
425 } else {
426 this.invalidateAtom(atom)
427 }
428 }
429
430 subscribe<A>(atom: Atom.Atom<A>, f: (_: A) => void, options?: { readonly immediate?: boolean }): () => void {
431 const node = this.ensureNode(atom)
432 if (options?.immediate) {
433 f(node.value())
434 } else if (node.state === NodeState.stale) {
435 // a listener hears the next change, which only reaches an up-to-date node
436 node.value()
437 }
438 const remove = node.subscribe(function() {
439 f(node._value)
440 })
441 return () => {
442 remove()
443 if (node.canBeRemoved) {
444 this.scheduleNodeRemoval(node)
445 }
446 }
447 }
448
449 mount<A>(atom: Atom.Atom<A>) {
450 return this.subscribe(atom, constVoid, constImmediate)
451 }
452
453 atomHasTtl(atom: Atom.Atom<any>): boolean {
454 return !atom.keepAlive && atom.idleTTL !== 0 && (atom.idleTTL !== undefined || this.defaultIdleTTL !== undefined)
455 }
456
457 ensureNode<A>(atom: Atom.Atom<A>): NodeImpl<A> {
458 const key = atomKey(atom)
459 let node = this.nodes.get(key)
460 if (node === undefined) {
461 node = this.createNode(atom)
462 this.nodes.set(key, node)
463 this.onNodeAdded?.(node)
464 } else if (this.atomHasTtl(atom)) {
465 this.removeNodeTimeout(node)
466 }
467 if (typeof key === "string" && this.preloadedSerializable.has(key)) {
468 const encoded = this.preloadedSerializable.get(key)
469 this.preloadedSerializable.delete(key)
470 const decoded = (atom as any as Atom.Serializable<any>)[SerializableTypeId].decode(encoded)
471 let target = atom
472 while (target.initialValueTarget) {
473 target = target.initialValueTarget
474 }
475 if (target === atom) {
476 node.setValue(decoded)
477 } else {
478 this.ensureNode(target).setInitialValue(decoded)
479 }
480 }
481 return node
482 }
483
484 createNode<A>(atom: Atom.Atom<A>): NodeImpl<A> {
485 if (this.disposed) {
486 throw new Error(`Cannot access Atom ${atom}: registry is disposed`)
487 }
488
489 if (!atom.keepAlive) {
490 this.scheduleAtomRemoval(atom)
491 }
492 return new NodeImpl(this, atom)
493 }
494
495 invalidateAtom = <A>(atom: Atom.Atom<A>): void => {
496 this.ensureNode(atom).invalidate()
497 }
498
499 scheduleAtomRemoval(atom: Atom.Atom<any>): void {
500 this.dispatcher.scheduleTask(() => {
501 const node = this.nodes.get(atomKey(atom))
502 if (node !== undefined && node.canBeRemoved) {
503 this.removeNode(node)
504 }
505 }, 0)
506 }
507
508 scheduleNodeRemoval(node: NodeImpl<any>): void {
509 this.dispatcher.scheduleTask(() => {
510 if (node.canBeRemoved) {
511 this.removeNode(node)
512 }
513 }, 0)
514 }
515
516 removeNode(node: NodeImpl<any>): void {
517 if (this.atomHasTtl(node.atom)) {
518 this.setNodeTimeout(node)
519 } else {
520 this.nodes.delete(atomKey(node.atom))
521 node.remove()
522 this.onNodeRemoved?.(node)
523 }
524 }
525
526 setNodeTimeout(node: NodeImpl<any>): void {
527 if (this.nodeTimeoutBucket.has(node)) {
528 return
529 }
530
531 let idleTTL = node.atom.idleTTL ?? this.defaultIdleTTL!
532 if (this.#currentSweepTTL !== null) {
533 idleTTL -= this.#currentSweepTTL
534 if (idleTTL <= 0) {
535 this.nodes.delete(atomKey(node.atom))
536 node.remove()
537 this.onNodeRemoved?.(node)
538 return
539 }
540 }
541 const ttl = Math.ceil(idleTTL! / this.timeoutResolution) * this.timeoutResolution
542 const timestamp = Date.now() + ttl
543 const bucket = timestamp - (timestamp % this.timeoutResolution) + this.timeoutResolution
544
545 let entry = this.timeoutBuckets.get(bucket)
546 if (entry === undefined) {
547 entry = [
548 new Set<NodeImpl<any>>(),
549 setTimeout(() => this.sweepBucket(bucket), bucket - Date.now()) as any
550 ]
551 this.timeoutBuckets.set(bucket, entry)
552 }
553 entry[0].add(node)
554 this.nodeTimeoutBucket.set(node, bucket)
555 }
556
557 removeNodeTimeout(node: NodeImpl<any>): void {
558 const bucket = this.nodeTimeoutBucket.get(node)
559 if (bucket === undefined) return
560 this.nodeTimeoutBucket.delete(node)
561 this.scheduleNodeRemoval(node)
562
563 const [nodes, handle] = this.timeoutBuckets.get(bucket)!
564 nodes.delete(node)
565 if (nodes.size === 0) {
566 clearTimeout(handle)
567 this.timeoutBuckets.delete(bucket)
568 }
569 }
570
571 #currentSweepTTL: number | null = null
572 sweepBucket(bucket: number): void {
573 const nodes = this.timeoutBuckets.get(bucket)![0]
574 this.timeoutBuckets.delete(bucket)
575 runAll(nodes, this.removeIdleNode)
576 }
577
578 removeIdleNode = (node: NodeImpl<any>): void => {
579 this.nodeTimeoutBucket.delete(node)
580 if (!node.canBeRemoved) return
581 this.nodes.delete(atomKey(node.atom))
582 this.onNodeRemoved?.(node)
583 this.#currentSweepTTL = node.atom.idleTTL ?? this.defaultIdleTTL!
584 try {
585 node.remove()
586 } finally {
587 this.#currentSweepTTL = null
588 }
589 }
590
591 reset(): void {
592 this.timeoutBuckets.forEach(([, handle]) => clearTimeout(handle))
593 this.timeoutBuckets.clear()
594 this.nodeTimeoutBucket.clear()
595
596 const nodes = Array.from(this.nodes.values())
597 this.nodes.clear()
598 // Mark all nodes before removing any, so a parent cannot be rescheduled for removal.
599 for (const node of nodes) {
600 node.state = NodeState.removed
601 }
602 runAll(nodes, this.destroyNode)
603 }
604
605 destroyNode = (node: NodeImpl<any>): void => {
606 node.remove()
607 this.onNodeRemoved?.(node)
608 }
609
610 dispose(): void {
611 this.disposed = true
612 this.reset()
613 }
614}
615
616const NodeFlags = {
617 alive: 1, // 1 << 0
618 initialized: 2, // 1 << 1,
619 waitingForValue: 4 // 1 << 2
620} as const
621type NodeFlags = typeof NodeFlags[keyof typeof NodeFlags]
622
623const NodeState = {
624 uninitialized: NodeFlags.alive | NodeFlags.waitingForValue,
625 stale: NodeFlags.alive | NodeFlags.initialized | NodeFlags.waitingForValue,
626 valid: NodeFlags.alive | NodeFlags.initialized,
627 removed: 0
628} as const
629type NodeState = number
630
631class NodeImpl<A> {
632 constructor(
633 registry: RegistryImpl,
634 atom: Atom.Atom<A>
635 ) {
636 this.registry = registry
637 this.atom = atom
638 this.writeContext = new WriteContextImpl(registry, this)
639 }
640
641 readonly registry: RegistryImpl
642 readonly atom: Atom.Atom<A>
643 state: NodeState = NodeState.uninitialized
644 lifetime: Lifetime<A> | undefined
645 writeContext: WriteContextImpl<A>
646 preserveInitialValueOnBuild = false
647 invalidatedDuringBuild = false
648 failed = false
649
650 // Every parent that holds this node as a child. A stale node keeps the
651 // parents of its last build, so they are not released and their next change
652 // still reaches it. A build drops the parents it did not read.
653 parents = new Set<NodeImpl<any>>()
654 children = new Set<NodeImpl<any>>()
655 listeners = new Set<() => void>()
656 observedChildren = 0
657
658 currentState() {
659 switch (this.state) {
660 case NodeState.uninitialized:
661 return "uninitialized"
662 case NodeState.stale:
663 return "stale"
664 case NodeState.valid:
665 return "valid"
666 default:
667 return "removed"
668 }
669 }
670
671 get canBeRemoved(): boolean {
672 return !this.atom.keepAlive && this.listeners.size === 0 && this.children.size === 0 && this.state !== 0
673 }
674
675 get building(): boolean {
676 return this.lifetime !== undefined && this.lifetime.reads !== undefined
677 }
678
679 // An observed node is rebuilt as soon as it goes stale, because something
680 // waits for its value without reading it. The count spares walking the
681 // children on every invalidation.
682 get isObserved(): boolean {
683 return !this.atom.lazy || this.listeners.size > 0 || this.observedChildren > 0
684 }
685
686 childObserved(delta: 1 | -1): void {
687 const observed = this.isObserved
688 this.observedChildren += delta
689 if (observed !== this.isObserved) {
690 this.observedChanged(delta)
691 }
692 }
693
694 observedChanged(delta: 1 | -1): void {
695 for (const parent of this.parents) {
696 parent.childObserved(delta)
697 }
698 }
699
700 _value: A = undefined as any
701 value(): A {
702 // a build can be superseded by a nested one, and that one staled again
703 // before the outer read returns
704 while ((this.state & NodeFlags.waitingForValue) !== 0) {
705 this.build()
706 }
707 return this._value
708 }
709
710 build(): void {
711 // a lifetime still attached to a node that needs a value belongs to a
712 // build that failed, or to one that is running and is now superseded
713 this.disposeLifetime()
714 this.invalidatedDuringBuild = false
715 const lifetime = makeLifetime(this)
716 this.lifetime = lifetime
717 let value: A
718 let read = false
719 try {
720 value = this.atom.read(lifetime)
721 read = true
722 } finally {
723 const reads = lifetime.reads!
724 lifetime.reads = undefined
725 if (this.lifetime === lifetime) {
726 this.dropUnread(reads)
727 if (!read) this.fail()
728 }
729 }
730 if (this.lifetime === lifetime && (this.state & NodeFlags.waitingForValue) !== 0) {
731 if (this.preserveInitialValueOnBuild) {
732 this.preserveInitialValueOnBuild = false
733 this.failed = false
734 this.state = NodeState.valid
735 } else {
736 this.setValue(value)
737 }
738 }
739 if (lifetime.failure !== noFailure) throw lifetime.failure
740 }
741
742 // A build that threw leaves its dependents holding, or reading, a value
743 // this node no longer stands behind, so they go stale and rebuild when
744 // pulled or observed. The read's error is the one reported.
745 fail(): void {
746 this.failed = true
747 try {
748 this.invalidateChildren()
749 } catch {
750 // an observed dependent rebuilt and met the same failure
751 }
752 }
753
754 readTracked<B>(atom: Atom.Atom<B>, lifetime: Lifetime<A>): B {
755 const parent = this.registry.ensureNode(atom)
756 // the dependency holds even when the parent cannot produce a value, so
757 // that its recovery reaches this node; a build superseded while the
758 // parent builds has the edge dropped by its replacement
759 if (!this.parents.has(parent)) {
760 this.parents.add(parent)
761 parent.children.add(this)
762 if (this.isObserved) {
763 parent.childObserved(1)
764 }
765 }
766 try {
767 return parent.value()
768 } catch (error) {
769 // the parent may have got its value and failed only to update another
770 // dependent: this build still takes the value, and rethrows once it is done
771 if (lifetime.reads === undefined || parent.state !== NodeState.valid) throw error
772 if (lifetime.failure === noFailure) lifetime.failure = error
773 return parent._value
774 } finally {
775 // read once the parent has answered: its commit while it is being
776 // pulled is the value this build is about to receive
777 if (lifetime.reads !== undefined && !lifetime.disposed) {
778 lifetime.reads.add(parent)
779 }
780 }
781 }
782
783 dropUnread(reads: Set<NodeImpl<any>>): void {
784 if (reads.size === this.parents.size) return
785 const observed = this.isObserved
786 for (const parent of this.parents) {
787 if (reads.has(parent)) continue
788 this.parents.delete(parent)
789 parent.children.delete(this)
790 if (observed) {
791 parent.childObserved(-1)
792 }
793 if (parent.canBeRemoved) {
794 this.registry.scheduleNodeRemoval(parent)
795 }
796 }
797 }
798
799 valueOption(): Option.Option<A> {
800 if ((this.state & NodeFlags.initialized) === 0) {
801 return Option.none()
802 }
803 return Option.some(this._value)
804 }
805
806 setInitialValue(value: A): void {
807 if ((this.state & NodeFlags.initialized) === 0) {
808 this.preserveInitialValueOnBuild = true
809 this.state = NodeState.stale
810 this._value = value
811 if (this.listeners.size > 0) {
812 this.announce()
813 }
814 return
815 }
816
817 this.setValue(value)
818 }
819
820 setValue(value: A): void {
821 // dependents that failed along with this node get another try
822 const recovered = this.failed
823 this.failed = false
824 const initialized = (this.state & NodeFlags.initialized) !== 0
825 this.state = NodeState.valid
826 if (initialized && this.atom.equals(this._value, value)) {
827 if (recovered) this.invalidateChildren()
828 return
829 }
830
831 this._value = value
832 let failure: unknown = noFailure
833 try {
834 this.invalidateChildren()
835 } catch (error) {
836 failure = error
837 }
838 // the value is committed, so its listeners hear of it even when a dependent failed
839 if (this.listeners.size > 0) {
840 try {
841 this.announce()
842 } catch (error) {
843 if (failure === noFailure) failure = error
844 }
845 }
846 if (failure !== noFailure) throw failure
847 }
848
849 announce(): void {
850 if (batchState.phase === BatchPhase.collect) {
851 batchState.notify.add(this)
852 } else {
853 this.notify()
854 }
855 }
856
857 invalidate(): void {
858 if (this.state === NodeState.valid) {
859 this.state = NodeState.stale
860 try {
861 this.disposeLifetime()
862 } finally {
863 this.propagate()
864 }
865 } else if (this.building && batchState.phase !== BatchPhase.collect) {
866 // the running build read a parent that has changed since
867 this.build()
868 } else {
869 this.propagate()
870 }
871 }
872
873 propagate(): void {
874 if (batchState.phase === BatchPhase.collect) {
875 if (this.building) {
876 this.invalidatedDuringBuild = true
877 }
878 batchState.stale.push(this)
879 } else if (this.isObserved) {
880 this.value()
881 } else {
882 this.invalidateChildren()
883 }
884 }
885
886 invalidateChildren(): void {
887 if (this.children.size === 0) {
888 return
889 }
890 let failure: unknown = noFailure
891 // a snapshot: a rebuild adds and drops children
892 for (const child of Array.from(this.children)) {
893 if (!child.affectedBy(this)) continue
894 try {
895 child.invalidate()
896 } catch (error) {
897 if (failure === noFailure) failure = error
898 }
899 }
900 if (failure !== noFailure) throw failure
901 }
902
903 // Whether a change of the parent reaches this node: it holds the parent's
904 // value, or a running build has already read it. A stale node is stale
905 // already, and a build that has not read the parent yet will read the new
906 // value. The exception is an observed node that is still stale: its rebuild
907 // failed, and a parent that has a value again is its chance to recover.
908 affectedBy(parent: NodeImpl<any>): boolean {
909 if (this.state === NodeState.valid) return true
910 const reads = this.lifetime?.reads
911 if (reads !== undefined) return reads.has(parent)
912 return this.isObserved && parent.state === NodeState.valid
913 }
914
915 notify(): void {
916 if (batchState.phase === BatchPhase.commit) {
917 batchState.notify.delete(this)
918 }
919 runAll(this.listeners, invoke)
920 }
921
922 disposeLifetime(): void {
923 const lifetime = this.lifetime
924 if (lifetime !== undefined) {
925 this.lifetime = undefined
926 lifetime.dispose()
927 }
928 }
929
930 remove() {
931 const observed = this.isObserved
932 this.state = NodeState.removed
933 this.listeners.clear()
934 try {
935 this.disposeLifetime()
936 } finally {
937 const parents = Array.from(this.parents)
938 this.parents.clear()
939 runAll(parents, (parent) => {
940 parent.children.delete(this)
941 if (observed) {
942 parent.childObserved(-1)
943 }
944 if (parent.canBeRemoved) {
945 this.registry.removeNode(parent)
946 }
947 })
948 }
949 }
950
951 subscribe(listener: () => void): () => void {
952 const observed = this.isObserved
953 this.listeners.add(listener)
954 if (!observed) {
955 this.observedChanged(1)
956 }
957 return () => {
958 if (this.listeners.delete(listener) && !this.isObserved) {
959 this.observedChanged(-1)
960 }
961 }
962 }
963}
964
965interface Lifetime<A> extends Atom.AtomContext {
966 isFn: boolean
967 readonly node: NodeImpl<A>
968 // the parents read so far by the build; `undefined` once the build has returned
969 reads: Set<NodeImpl<any>> | undefined
970 // a failure met while reading a parent that still produced its value
971 failure: unknown
972 finalizers: Array<() => void> | undefined
973 disposed: boolean
974 readonly dispose: () => void
975}
976
977const LifetimeProto: Omit<Lifetime<any>, "node" | "finalizers" | "disposed" | "isFn" | "reads" | "failure"> = {
978 get registry(): RegistryImpl {
979 return (this as Lifetime<any>).node.registry
980 },
981
982 addFinalizer(this: Lifetime<any>, f: () => void): void {
983 if (this.disposed) return f()
984 this.finalizers ??= []
985 this.finalizers.push(f)
986 },
987
988 get<A>(this: Lifetime<any>, atom: Atom.Atom<A>): A {
989 if (this.disposed) {
990 return this.node.registry.get(atom)
991 }
992 return this.node.readTracked(atom, this)
993 },
994
995 result<A, E>(this: Lifetime<any>, atom: Atom.Atom<Result.AsyncResult<A, E>>, options?: {
996 readonly suspendOnWaiting?: boolean | undefined
997 }): Effect.Effect<A, E> {
998 if (this.disposed || this.isFn) {
999 return this.resultOnce(atom, options)
1000 }
1001 const result = this.get(atom)
1002 if (options?.suspendOnWaiting && result.waiting) {
1003 return Effect.never
1004 }
1005 switch (result._tag) {
1006 case "Initial": {
1007 return Effect.never
1008 }
1009 case "Failure": {
1010 return Exit.failCause(result.cause)
1011 }
1012 case "Success": {
1013 return Effect.succeed(result.value)
1014 }
1015 }
1016 },
1017
1018 resultOnce<A, E>(this: Lifetime<any>, atom: Atom.Atom<Result.AsyncResult<A, E>>, options?: {
1019 readonly suspendOnWaiting?: boolean | undefined
1020 }): Effect.Effect<A, E> {
1021 return Effect.callback<A, E>((resume) => {
1022 const result = this.once(atom)
1023 if (result._tag !== "Initial" && !(options?.suspendOnWaiting && result.waiting)) {
1024 return resume(Result.toExit(result) as any)
1025 }
1026 const cancel = this.node.registry.subscribe(atom, (result) => {
1027 if (result._tag === "Initial" || (options?.suspendOnWaiting && result.waiting)) return
1028 cancel()
1029 resume(Result.toExit(result) as any)
1030 }, { immediate: false })
1031 return Effect.sync(cancel)
1032 })
1033 },
1034
1035 setResult<A, E, W>(
1036 this: Lifetime<any>,
1037 atom: Atom.Writable<Result.AsyncResult<A, E>, W>,
1038 value: W
1039 ): Effect.Effect<A, E> {
1040 if (this.disposed) return Effect.never
1041 this.node.registry.set(atom, value)
1042 return this.resultOnce(atom, { suspendOnWaiting: true })
1043 },
1044
1045 some<A>(this: Lifetime<any>, atom: Atom.Atom<Option.Option<A>>): Effect.Effect<A> {
1046 if (this.disposed || this.isFn) {
1047 return this.someOnce(atom)
1048 }
1049 const result = this.get(atom)
1050 return result._tag === "None" ? Effect.never : Effect.succeed(result.value)
1051 },
1052
1053 someOnce<A>(this: Lifetime<any>, atom: Atom.Atom<Option.Option<A>>): Effect.Effect<A> {
1054 return Effect.callback<A>((resume) => {
1055 const result = this.once(atom)
1056 if (Option.isSome(result)) {
1057 return resume(Effect.succeed(result.value))
1058 }
1059 const cancel = this.node.registry.subscribe(atom, (result) => {
1060 if (Option.isNone(result)) return
1061 cancel()
1062 resume(Effect.succeed(result.value))
1063 }, { immediate: false })
1064 return Effect.sync(cancel)
1065 })
1066 },
1067
1068 once<A>(this: Lifetime<any>, atom: Atom.Atom<A>): A {
1069 return this.node.registry.get(atom)
1070 },
1071
1072 self<A>(this: Lifetime<any>): Option.Option<A> {
1073 if (this.disposed) return Option.none()
1074 return this.node.valueOption() as any
1075 },
1076
1077 refresh<A>(this: Lifetime<any>, atom: Atom.Atom<A>): void {
1078 if (this.disposed) return
1079 this.node.registry.refresh(atom)
1080 },
1081
1082 refreshSelf(this: Lifetime<any>): void {
1083 if (this.disposed) return
1084 this.node.invalidate()
1085 },
1086
1087 mount<A>(this: Lifetime<any>, atom: Atom.Atom<A>): void {
1088 if (this.disposed) return
1089 this.addFinalizer(this.node.registry.mount(atom))
1090 },
1091
1092 subscribe<A>(this: Lifetime<any>, atom: Atom.Atom<A>, f: (_: A) => void, options?: {
1093 readonly immediate?: boolean
1094 }): void {
1095 if (this.disposed) return
1096 this.addFinalizer(this.node.registry.subscribe(atom, f, options))
1097 },
1098
1099 setSelf<A>(this: Lifetime<any>, a: A): void {
1100 if (this.disposed) return
1101 this.node.setValue(a as any)
1102 },
1103
1104 set<R, W>(this: Lifetime<any>, atom: Atom.Writable<R, W>, value: W): void {
1105 if (this.disposed) return
1106 this.node.registry.set(atom, value)
1107 },
1108
1109 stream<A>(this: Lifetime<any>, atom: Atom.Atom<A>, options?: {
1110 readonly withoutInitialValue?: boolean
1111 }) {
1112 if (this.disposed) return Stream.empty
1113 return Stream.callback<A>((queue) =>
1114 Effect.sync(() => {
1115 this.subscribe(atom, (value) => Queue.offerUnsafe(queue, value), {
1116 immediate: !options?.withoutInitialValue
1117 })
1118 })
1119 )
1120 },
1121
1122 streamResult<A, E>(this: Lifetime<any>, atom: Atom.Atom<Result.AsyncResult<A, E>>, options?: {
1123 readonly withoutInitialValue?: boolean
1124 readonly bufferSize?: number
1125 }): Stream.Stream<A, E> {
1126 return this.stream(atom, options).pipe(
1127 Stream.filter(Result.isNotInitial),
1128 Stream.mapEffect((result) =>
1129 result._tag === "Success" ? Effect.succeed(result.value) : Effect.failCause(result.cause)
1130 )
1131 )
1132 },
1133
1134 dispose(this: Lifetime<any>): void {
1135 this.disposed = true
1136 if (this.finalizers === undefined) {
1137 return
1138 }
1139
1140 const finalizers = this.finalizers
1141 this.finalizers = undefined
1142 runAll(finalizers.reverse(), invoke)
1143 }
1144}
1145
1146const makeLifetime = <A>(node: NodeImpl<A>): Lifetime<A> => {
1147 function get<A>(atom: Atom.Atom<A>): A {
1148 if (get.disposed || get.isFn) {
1149 return node.registry.get(atom)
1150 }
1151 return node.readTracked(atom, get as any)
1152 }
1153 Object.setPrototypeOf(get, LifetimeProto)
1154 get.isFn = false
1155 get.disposed = false
1156 get.finalizers = undefined
1157 get.node = node
1158 get.reads = new Set<NodeImpl<any>>()
1159 get.failure = noFailure
1160 return get as any
1161}
1162
1163class WriteContextImpl<A> implements Atom.WriteContext<A> {
1164 constructor(
1165 registry: RegistryImpl,
1166 node: NodeImpl<A>
1167 ) {
1168 this.registry = registry
1169 this.node = node
1170 }
1171 readonly registry: RegistryImpl
1172 readonly node: NodeImpl<A>
1173 get<A>(atom: Atom.Atom<A>): A {
1174 return this.registry.get(atom)
1175 }
1176 set<R, W>(atom: Atom.Writable<R, W>, value: W) {
1177 return this.registry.set(atom, value)
1178 }
1179 setSelf(value: any) {
1180 return this.node.setValue(value)
1181 }
1182 refreshSelf() {
1183 return this.node.invalidate()
1184 }
1185}
1186
1187// -----------------------------------------------------------------------------
1188// batching
1189// -----------------------------------------------------------------------------
1190
1191/**
1192 * @internal
1193 */
1194export const BatchPhase = {
1195 disabled: 0,
1196 collect: 1,
1197 commit: 2
1198} as const
1199
1200/**
1201 * @internal
1202 */
1203export type BatchPhase = typeof BatchPhase[keyof typeof BatchPhase]
1204
1205/**
1206 * @internal
1207 */
1208export const batchState = {
1209 phase: BatchPhase.disabled as BatchPhase,
1210 depth: 0,
1211 stale: [] as Array<NodeImpl<any>>,
1212 notify: new Set<NodeImpl<any>>()
1213}
1214
1215/**
1216 * @internal
1217 */
1218export function batch(f: () => void): void {
1219 const previousPhase = batchState.phase
1220 batchState.phase = BatchPhase.collect
1221 batchState.depth++
1222 let failure: unknown = noFailure
1223 try {
1224 f()
1225 } catch (error) {
1226 failure = error
1227 }
1228 if (batchState.depth === 1) {
1229 // what `f` changed before it failed is committed, so it is still rebuilt and heard
1230 const { notify, stale } = batchState
1231 let i = 0
1232 do {
1233 batchState.phase = BatchPhase.collect
1234 for (; i < stale.length; i++) {
1235 try {
1236 batchRebuildNode(stale[i])
1237 } catch (error) {
1238 if (failure === noFailure) failure = error
1239 }
1240 }
1241 batchState.phase = BatchPhase.commit
1242 for (const node of notify) {
1243 try {
1244 node.notify()
1245 } catch (error) {
1246 if (failure === noFailure) failure = error
1247 }
1248 }
1249 } while (i < stale.length)
1250 batchState.stale = []
1251 notify.clear()
1252 }
1253 batchState.depth--
1254 batchState.phase = previousPhase
1255 if (failure !== noFailure) throw failure
1256}
1257
1258function batchRebuildNode(node: NodeImpl<any>) {
1259 if (node.state === NodeState.valid) {
1260 if (!node.invalidatedDuringBuild) {
1261 return
1262 }
1263 node.invalidatedDuringBuild = false
1264 node.state = NodeState.stale
1265 node.disposeLifetime()
1266 }
1267 node.value()
1268}