1/**
2 * Reactive state primitives for values managed by an `AtomRegistry`.
3 *
4 * An `Atom` describes how to produce or update one piece of reactive state. The
5 * registry runs atom reads, remembers current values, tracks dependencies
6 * between atoms, starts effects and streams, and cleans up atoms that are no
7 * longer used. This module includes the atom constructors and update helpers
8 * used for cached values, effect-backed values, streams, browser state, stored
9 * values, and server-rendered values.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Arr from "../Array.ts"
15import * as Cause from "../Cause.ts"
16import * as Channel from "../Channel.ts"
17import * as Context from "../Context.ts"
18import * as Duration from "../Duration.ts"
19import * as Effect from "../Effect.ts"
20import * as Exit from "../Exit.ts"
21import * as Fiber from "../Fiber.ts"
22import type { LazyArg } from "../Function.ts"
23import { constant, constTrue, constVoid, dual, pipe } from "../Function.ts"
24import type * as Inspectable from "../Inspectable.ts"
25import { PipeInspectableProto } from "../internal/core.ts"
26import { getStackTraceLimit } from "../internal/stackTraceLimit.ts"
27import * as Layer from "../Layer.ts"
28import * as MutableHashMap from "../MutableHashMap.ts"
29import * as Option from "../Option.ts"
30import * as KeyValueStore from "../persistence/KeyValueStore.ts"
31import type { Pipeable } from "../Pipeable.ts"
32import { hasProperty } from "../Predicate.ts"
33import * as Pull from "../Pull.ts"
34import type { ReadonlyRecord } from "../Record.ts"
35import * as Scheduler from "../Scheduler.ts"
36import * as Schema from "../Schema.ts"
37import * as Scope from "../Scope.ts"
38import * as Stream from "../Stream.ts"
39import * as SubscriptionRef from "../SubscriptionRef.ts"
40import type { Mutable, NoInfer } from "../Types.ts"
41import * as AsyncResult from "./AsyncResult.ts"
42import { AtomRegistry } from "./AtomRegistry.ts"
43import * as Registry from "./AtomRegistry.ts"
44import * as Reactivity from "./Reactivity.ts"
45
46/**
47 * Type-level identifier used to recognize `Atom` values.
48 *
49 * @stability unstable
50 * @category type IDs
51 * @since 4.0.0
52 */
53export type TypeId = "~effect/reactivity/Atom"
54
55/**
56 * Runtime identifier attached to `Atom` values and used by `isAtom`.
57 *
58 * @stability unstable
59 * @category type IDs
60 * @since 4.0.0
61 */
62export const TypeId: TypeId = "~effect/reactivity/Atom"
63
64/**
65 * Reactive value read by an `AtomRegistry`, with metadata controlling caching, laziness, refresh behavior, and initial value targeting.
66 *
67 * @stability unstable
68 * @category models
69 * @since 4.0.0
70 */
71export interface Atom<A> extends Pipeable, Inspectable.Inspectable {
72 readonly [TypeId]: TypeId
73 readonly keepAlive: boolean
74 readonly lazy: boolean
75 readonly read: (get: AtomContext) => A
76 equals(value: A, next: A): boolean
77 readonly refresh?: (f: <A>(atom: Atom<A>) => void) => void
78 readonly label?: readonly [name: string, stack: string]
79 readonly idleTTL?: number
80 readonly initialValueTarget?: Atom<A>
81}
82
83/**
84 * Returns `true` when a value is an `Atom`.
85 *
86 * @stability unstable
87 * @category guards
88 * @since 4.0.0
89 */
90export const isAtom = (u: unknown): u is Atom<any> => hasProperty(u, TypeId)
91
92/**
93 * Extracts the value type produced by an `Atom`.
94 *
95 * @stability unstable
96 * @category utility types
97 * @since 4.0.0
98 */
99export type Type<T extends Atom<any>> = T extends Atom<infer A> ? A : never
100
101/**
102 * Extracts the success value type from an atom whose value is an `AsyncResult`.
103 *
104 * @stability unstable
105 * @category utility types
106 * @since 4.0.0
107 */
108export type Success<T extends Atom<any>> = T extends Atom<AsyncResult.AsyncResult<infer A, infer _>> ? A : never
109
110/**
111 * Extracts the item type from an atom whose value is a `PullResult`.
112 *
113 * @stability unstable
114 * @category utility types
115 * @since 4.0.0
116 */
117export type PullSuccess<T extends Atom<any>> = T extends Atom<PullResult<infer A, infer _>> ? A : never
118
119/**
120 * Extracts the failure error type from an atom whose value is an `AsyncResult`.
121 *
122 * @stability unstable
123 * @category utility types
124 * @since 4.0.0
125 */
126export type Failure<T extends Atom<any>> = T extends Atom<AsyncResult.AsyncResult<infer _, infer E>> ? E : never
127
128/**
129 * Returns an atom type without serializable metadata, preserving `Writable` read and write types when the input atom is writable.
130 *
131 * @stability unstable
132 * @category utility types
133 * @since 4.0.0
134 */
135export type WithoutSerializable<T extends Atom<any>> = T extends Writable<infer R, infer W> ? Writable<R, W>
136 : Atom<Type<T>>
137
138/**
139 * Runtime identifier attached to writable atoms and used by `isWritable`.
140 *
141 * @stability unstable
142 * @category type IDs
143 * @since 4.0.0
144 */
145export const WritableTypeId: WritableTypeId = "~effect/reactivity/Atom/Writable"
146
147/**
148 * Type-level identifier used to recognize writable atoms.
149 *
150 * @stability unstable
151 * @category type IDs
152 * @since 4.0.0
153 */
154export type WritableTypeId = "~effect/reactivity/Atom/Writable"
155
156/**
157 * Atom that can also be written to, using a `WriteContext` and an input value to update reactive state.
158 *
159 * @stability unstable
160 * @category models
161 * @since 4.0.0
162 */
163export interface Writable<R, W = R> extends Atom<R> {
164 readonly [WritableTypeId]: WritableTypeId
165 readonly write: (ctx: WriteContext<R>, value: W) => void
166}
167
168/**
169 * Context passed to atom read functions for reading dependencies, awaiting `AsyncResult` or `Option` values, managing subscriptions and finalizers, refreshing atoms, and updating writable atoms.
170 *
171 * @stability unstable
172 * @category context
173 * @since 4.0.0
174 */
175export interface AtomContext {
176 <A>(atom: Atom<A>): A
177 get<A>(this: AtomContext, atom: Atom<A>): A
178 result<A, E>(this: AtomContext, atom: Atom<AsyncResult.AsyncResult<A, E>>, options?: {
179 readonly suspendOnWaiting?: boolean | undefined
180 }): Effect.Effect<A, E>
181 resultOnce<A, E>(this: AtomContext, atom: Atom<AsyncResult.AsyncResult<A, E>>, options?: {
182 readonly suspendOnWaiting?: boolean | undefined
183 }): Effect.Effect<A, E>
184 once<A>(this: AtomContext, atom: Atom<A>): A
185 addFinalizer(this: AtomContext, f: () => void): void
186 mount<A>(this: AtomContext, atom: Atom<A>): void
187 refresh<A>(this: AtomContext, atom: Atom<A>): void
188 refreshSelf(this: AtomContext): void
189 self<A>(this: AtomContext): Option.Option<A>
190 setSelf<A>(this: AtomContext, a: A): void
191 set<R, W>(this: AtomContext, atom: Writable<R, W>, value: W): void
192 setResult<A, E, W>(this: AtomContext, atom: Writable<AsyncResult.AsyncResult<A, E>, W>, value: W): Effect.Effect<A, E>
193 some<A>(this: AtomContext, atom: Atom<Option.Option<A>>): Effect.Effect<A>
194 someOnce<A>(this: AtomContext, atom: Atom<Option.Option<A>>): Effect.Effect<A>
195 stream<A>(this: AtomContext, atom: Atom<A>, options?: {
196 readonly withoutInitialValue?: boolean
197 readonly bufferSize?: number
198 }): Stream.Stream<A>
199 streamResult<A, E>(this: AtomContext, atom: Atom<AsyncResult.AsyncResult<A, E>>, options?: {
200 readonly withoutInitialValue?: boolean
201 readonly bufferSize?: number
202 }): Stream.Stream<A, E>
203 subscribe<A>(this: AtomContext, atom: Atom<A>, f: (_: A) => void, options?: {
204 readonly immediate?: boolean
205 }): void
206 readonly registry: Registry.AtomRegistry
207}
208
209/**
210 * Context passed to writable atom write functions for reading atoms, refreshing or setting the current atom, and writing to other writable atoms.
211 *
212 * @stability unstable
213 * @category context
214 * @since 4.0.0
215 */
216export interface WriteContext<A> {
217 get<T>(this: WriteContext<A>, atom: Atom<T>): T
218 refreshSelf(this: WriteContext<A>): void
219 setSelf(this: WriteContext<A>, a: A): void
220 set<R, W>(this: WriteContext<A>, atom: Writable<R, W>, value: W): void
221}
222
223/**
224 * Returns a copy of an atom with an idle time-to-live: finite durations dispose it after inactivity, while an infinite duration keeps it alive.
225 *
226 * @stability unstable
227 * @category combinators
228 * @since 4.0.0
229 */
230export const setIdleTTL: {
231 (duration: Duration.Input): <A extends Atom<any>>(self: A) => A
232 <A extends Atom<any>>(self: A, duration: Duration.Input): A
233} = dual<
234 (duration: Duration.Input) => <A extends Atom<any>>(self: A) => A,
235 <A extends Atom<any>>(self: A, duration: Duration.Input) => A
236>(2, (self, durationInput) => {
237 const duration = Duration.fromInputUnsafe(durationInput)
238 const isFinite = Duration.isFinite(duration)
239 return Object.assign(Object.create(Object.getPrototypeOf(self)), {
240 ...self,
241 keepAlive: !isFinite,
242 idleTTL: isFinite ? Duration.toMillis(duration) : undefined
243 })
244})
245
246const removeTtl = setIdleTTL(0)
247
248const AtomProto = {
249 [TypeId]: TypeId,
250 equals: Object.is,
251 ...PipeInspectableProto,
252 toJSON(this: Atom<any>) {
253 return {
254 _id: "Atom",
255 keepAlive: this.keepAlive,
256 lazy: this.lazy,
257 label: this.label
258 }
259 }
260} as const
261
262const RuntimeProto = {
263 ...AtomProto,
264 atom(this: AtomRuntime<any, any>, arg: any, options?: {
265 readonly initialValue?: unknown
266 readonly uninterruptible?: boolean | undefined
267 }) {
268 const read = makeRead(arg, options)
269 return readable((get) => {
270 const previous = get.self<AsyncResult.AsyncResult<any, any>>()
271 const runtimeResult = get(this)
272 if (runtimeResult._tag !== "Success") {
273 return AsyncResult.replacePrevious(runtimeResult, previous)
274 }
275 return read(get, runtimeResult.value)
276 })
277 },
278
279 fn(this: AtomRuntime<any, any>, arg: any, options?: {
280 readonly initialValue?: unknown
281 readonly reactivityKeys?: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>> | undefined
282 readonly concurrent?: boolean | undefined
283 }) {
284 if (arguments.length === 0) {
285 return (arg: any, options?: {}) => makeFnRuntime(this, arg, options)
286 }
287 return makeFnRuntime(this, arg, options)
288 },
289
290 pull(this: AtomRuntime<any, any>, arg: any, options?: {
291 readonly disableAccumulation?: boolean
292 readonly initialValue?: ReadonlyArray<any>
293 }) {
294 const pullSignal = removeTtl(state(0))
295 const pullAtom = readable((get) => {
296 const previous = get.self<AsyncResult.AsyncResult<any, any>>()
297 const runtimeResult = get(this)
298 if (runtimeResult._tag !== "Success") {
299 return AsyncResult.replacePrevious(runtimeResult, previous)
300 }
301 return makeEffect(
302 get,
303 makeStreamPullEffect(get, pullSignal, arg, options),
304 AsyncResult.initial(true),
305 runtimeResult.value
306 )
307 })
308 return makeStreamPull(pullSignal, pullAtom)
309 },
310
311 subscriptionRef(this: AtomRuntime<any, any>, ref: any) {
312 return makeSubRef(
313 removeTtl(readable((get) => {
314 const previous = get.self<AsyncResult.AsyncResult<any, any>>()
315 const runtimeResult = get(this)
316 if (runtimeResult._tag !== "Success") {
317 return AsyncResult.replacePrevious(runtimeResult, previous)
318 }
319 const value = typeof ref === "function" ? ref(get) : ref
320 return SubscriptionRef.isSubscriptionRef(value)
321 ? value
322 : makeEffect(get, value, AsyncResult.initial(true), runtimeResult.value)
323 })),
324 (get, ref) => {
325 const runtime = AsyncResult.getOrThrow(get(this))
326 return readSubscriptionRef(get, ref, runtime)
327 }
328 )
329 }
330}
331
332const makeFnRuntime = (
333 self: AtomRuntime<any, any>,
334 arg: (
335 arg: any,
336 get: FnContext
337 ) =>
338 | Effect.Effect<any, any, Scope.Scope | AtomRegistry>
339 | Stream.Stream<any, any, AtomRegistry>,
340 options?: {
341 readonly initialValue?: unknown
342 readonly reactivityKeys?: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>> | undefined
343 }
344) => {
345 const [read, write, argAtom] = makeResultFn(
346 options?.reactivityKeys ?
347 ((a: any, get: FnContext) => {
348 const effect = arg(a, get)
349 return Effect.isEffect(effect)
350 ? Reactivity.mutation(effect, options.reactivityKeys!)
351 : Stream.ensuring(effect, Reactivity.invalidate(options.reactivityKeys!))
352 }) as any :
353 arg,
354 options
355 )
356 return writable((get) => {
357 get.get(argAtom)
358 const previous = get.self<AsyncResult.AsyncResult<any, any>>()
359 const runtimeResult = get.get(self)
360 if (runtimeResult._tag !== "Success") {
361 return AsyncResult.replacePrevious(runtimeResult, previous)
362 }
363 return read(get, runtimeResult.value)
364 }, write)
365}
366
367const WritableProto = {
368 ...AtomProto,
369 [WritableTypeId]: WritableTypeId
370} as const
371
372/**
373 * Returns `true` when an atom is writable.
374 *
375 * @stability unstable
376 * @category guards
377 * @since 4.0.0
378 */
379export const isWritable = <R, W>(atom: Atom<R>): atom is Writable<R, W> => WritableTypeId in atom
380
381/**
382 * Creates a read-only atom from a read function and an optional custom refresh registration callback.
383 *
384 * @stability unstable
385 * @category constructors
386 * @since 4.0.0
387 */
388export const readable = <A>(
389 read: (get: AtomContext) => A,
390 refresh?: (f: <A>(atom: Atom<A>) => void) => void
391): Atom<A> => {
392 const self = Object.create(AtomProto)
393 self.keepAlive = false
394 self.lazy = true
395 self.read = read
396 self.refresh = refresh
397 return self
398}
399
400/**
401 * Creates a writable atom from read and write functions, with an optional custom refresh registration callback.
402 *
403 * @stability unstable
404 * @category constructors
405 * @since 4.0.0
406 */
407export const writable = <R, W>(
408 read: (get: AtomContext) => R,
409 write: (ctx: WriteContext<R>, value: W) => void,
410 refresh?: (f: <A>(atom: Atom<A>) => void) => void
411): Writable<R, W> => {
412 const self = Object.create(WritableProto)
413 self.keepAlive = false
414 self.lazy = true
415 self.read = read
416 self.write = write
417 self.refresh = refresh
418 return self
419}
420
421function constSetSelf<A>(ctx: WriteContext<A>, value: A) {
422 ctx.setSelf(value)
423}
424
425// -----------------------------------------------------------------------------
426// constructors
427// -----------------------------------------------------------------------------
428
429/**
430 * Creates an atom from a synchronous value or read function, or from an `Effect` or `Stream` whose state is exposed as an `AsyncResult`; plain values create writable state atoms.
431 *
432 * @stability unstable
433 * @category constructors
434 * @since 4.0.0
435 */
436export const make: {
437 <A, E>(create: (get: AtomContext) => Effect.Effect<A, E, Scope.Scope | AtomRegistry>, options?: {
438 readonly initialValue?: A | undefined
439 readonly uninterruptible?: boolean | undefined
440 }): Atom<AsyncResult.AsyncResult<A, E>>
441 <A, E>(effect: Effect.Effect<A, E, Scope.Scope | AtomRegistry>, options?: {
442 readonly initialValue?: A
443 readonly uninterruptible?: boolean | undefined
444 }): Atom<AsyncResult.AsyncResult<A, E>>
445 <A, E>(create: (get: AtomContext) => Stream.Stream<A, E, AtomRegistry>, options?: {
446 readonly initialValue?: A
447 }): Atom<AsyncResult.AsyncResult<A, E | Cause.NoSuchElementError>>
448 <A, E>(stream: Stream.Stream<A, E, AtomRegistry>, options?: {
449 readonly initialValue?: A
450 }): Atom<AsyncResult.AsyncResult<A, E | Cause.NoSuchElementError>>
451 <A>(create: (get: AtomContext) => A): Atom<A>
452 <A>(initialValue: A): Writable<A>
453} = (arg: any, options?: {
454 readonly initialValue?: unknown
455 readonly uninterruptible?: boolean | undefined
456}) => {
457 const readOrAtom = makeRead(arg, options)
458 if (TypeId in readOrAtom) {
459 return readOrAtom as any
460 }
461 return readable(readOrAtom)
462}
463
464// -----------------------------------------------------------------------------
465// constructors - effect
466// -----------------------------------------------------------------------------
467
468const makeRead: {
469 <A, E>(effect: Effect.Effect<A, E, Scope.Scope | AtomRegistry>, options?: {
470 readonly initialValue?: A
471 readonly uninterruptible?: boolean | undefined
472 }): (get: AtomContext, services?: Context.Context<any>) => AsyncResult.AsyncResult<A, E>
473 <A, E>(create: (get: AtomContext) => Effect.Effect<A, E, Scope.Scope | AtomRegistry>, options?: {
474 readonly initialValue?: A
475 readonly uninterruptible?: boolean | undefined
476 }): (get: AtomContext, services?: Context.Context<any>) => AsyncResult.AsyncResult<A, E>
477 <A, E>(stream: Stream.Stream<A, E, AtomRegistry>, options?: {
478 readonly initialValue?: A
479 readonly uninterruptible?: boolean | undefined
480 }): (get: AtomContext, services?: Context.Context<any>) => AsyncResult.AsyncResult<A, E | Cause.NoSuchElementError>
481 <A, E>(create: (get: AtomContext) => Stream.Stream<A, E, AtomRegistry>, options?: {
482 readonly initialValue?: A
483 readonly uninterruptible?: boolean | undefined
484 }): (get: AtomContext, services?: Context.Context<any>) => AsyncResult.AsyncResult<A, E | Cause.NoSuchElementError>
485 <A>(create: (get: AtomContext) => A): (get: AtomContext, services?: Context.Context<any>) => A
486 <A>(initialValue: A): Writable<A>
487} = <A, E>(
488 arg:
489 | Effect.Effect<A, E, Scope.Scope | AtomRegistry>
490 | ((get: AtomContext) => Effect.Effect<A, E, Scope.Scope | AtomRegistry>)
491 | Stream.Stream<A, E, AtomRegistry>
492 | ((get: AtomContext) => Stream.Stream<A, E, AtomRegistry>)
493 | ((get: AtomContext) => A)
494 | A,
495 options?: {
496 readonly initialValue?: unknown
497 readonly uninterruptible?: boolean | undefined
498 }
499) => {
500 if (typeof arg === "function" && !Effect.isEffect(arg) && !Stream.isStream(arg)) {
501 const create = arg as (get: AtomContext) => any
502 return function(get: AtomContext, providedServices?: Context.Context<any>) {
503 const value = create(get)
504 switch (typeof value) {
505 case "function":
506 case "object": {
507 if (value === null) return value
508 else if (EffectTypeId in value) {
509 return effect(get, value as any, options, providedServices)
510 } else if (StreamTypeId in value) {
511 return stream(get, value as any, options, providedServices)
512 }
513 return value
514 }
515 default:
516 return value
517 }
518 }
519 } else if (Effect.isEffect(arg)) {
520 return function(get: AtomContext, providedServices?: Context.Context<any>) {
521 return effect(get, arg as any, options, providedServices)
522 }
523 } else if (Stream.isStream(arg)) {
524 return function(get: AtomContext, providedServices?: Context.Context<any>) {
525 return stream(get, arg as any, options, providedServices)
526 }
527 }
528
529 return state(arg) as any
530}
531
532const EffectTypeId: keyof Effect.Effect<any> = "~effect/Effect"
533const StreamTypeId: keyof Stream.Stream<any> = "~effect/Stream"
534
535const state = <A>(
536 initialValue: A
537): Writable<A> =>
538 writable(function(_get) {
539 return initialValue
540 }, constSetSelf)
541
542const effect = <A, E>(
543 get: AtomContext,
544 effect: Effect.Effect<A, E, Scope.Scope | AtomRegistry>,
545 options?: {
546 readonly initialValue?: A
547 readonly uninterruptible?: boolean | undefined
548 },
549 services?: Context.Context<any>
550): AsyncResult.AsyncResult<A, E> => {
551 const initialValue = options?.initialValue !== undefined
552 ? AsyncResult.success<A, E>(options.initialValue)
553 : AsyncResult.initial<A, E>()
554 return makeEffect(get, effect, initialValue, services, options?.uninterruptible)
555}
556
557function makeEffect<A, E>(
558 ctx: AtomContext,
559 effect: Effect.Effect<A, E, Scope.Scope | AtomRegistry>,
560 initialValue: AsyncResult.AsyncResult<A, E>,
561 services = Context.empty(),
562 uninterruptible = false
563): AsyncResult.AsyncResult<A, E> {
564 const previous = ctx.self<AsyncResult.AsyncResult<A, E>>()
565 const scope = Scope.makeUnsafe()
566 ctx.addFinalizer(() => {
567 Effect.runForkWith(services)(Scope.close(scope, Exit.void))
568 })
569 let syncResult: AsyncResult.AsyncResult<A, E> | undefined
570 let isAsync = false
571 const cancel = runCallbackSync(
572 services.pipe(
573 Context.add(Scope.Scope, scope),
574 Context.add(AtomRegistry, ctx.registry),
575 Context.add(Scheduler.Scheduler, ctx.registry.scheduler)
576 ),
577 effect,
578 function(exit) {
579 syncResult = AsyncResult.fromExitWithPrevious(exit, previous)
580 if (isAsync) {
581 ctx.setSelf(syncResult)
582 }
583 },
584 uninterruptible
585 )
586 isAsync = true
587 if (cancel !== undefined) {
588 ctx.addFinalizer(cancel)
589 }
590 if (syncResult !== undefined) {
591 return syncResult
592 } else if (previous._tag === "Some") {
593 return AsyncResult.waitingFrom(previous)
594 }
595 return AsyncResult.waiting(initialValue)
596}
597
598function runCallbackSync<R, A, E, ER = never>(
599 services: Context.Context<R>,
600 effect: Effect.Effect<A, E, R>,
601 onExit: (exit: Exit.Exit<A, E | ER>) => void,
602 uninterruptible = false
603): (() => void) | undefined {
604 if (Exit.isExit(effect)) {
605 onExit(effect as any)
606 return undefined
607 }
608 const runFork = Effect.runForkWith(services)
609 const fiber = runFork(effect)
610 fiber.currentDispatcher?.flush()
611 const result = fiber.pollUnsafe()
612 if (result) {
613 onExit(result)
614 return undefined
615 }
616 const remove = fiber.addObserver(onExit)
617 function cancel() {
618 remove()
619 if (!uninterruptible) {
620 fiber.interruptUnsafe()
621 }
622 }
623 return cancel
624}
625
626// -----------------------------------------------------------------------------
627// context
628// -----------------------------------------------------------------------------
629
630/**
631 * Atom that builds a `Context` from a `Layer` and exposes constructors for atoms, functions, pulls, and subscription refs that run with that context.
632 *
633 * @stability unstable
634 * @category models
635 * @since 4.0.0
636 */
637export interface AtomRuntime<R, ER = never> extends Atom<AsyncResult.AsyncResult<Context.Context<R>, ER>> {
638 readonly factory: RuntimeFactory
639
640 readonly layer: Atom<Layer.Layer<R, ER>>
641
642 readonly atom: {
643 <A, E>(
644 create: (get: AtomContext) => Effect.Effect<A, E, Scope.Scope | R | AtomRegistry | Reactivity.Reactivity>,
645 options?: {
646 readonly initialValue?: A
647 readonly uninterruptible?: boolean | undefined
648 }
649 ): Atom<AsyncResult.AsyncResult<A, E | ER>>
650 <A, E>(effect: Effect.Effect<A, E, Scope.Scope | R | AtomRegistry | Reactivity.Reactivity>, options?: {
651 readonly initialValue?: A
652 readonly uninterruptible?: boolean | undefined
653 }): Atom<AsyncResult.AsyncResult<A, E | ER>>
654 <A, E>(create: (get: AtomContext) => Stream.Stream<A, E, AtomRegistry | Reactivity.Reactivity | R>, options?: {
655 readonly initialValue?: A
656 }): Atom<AsyncResult.AsyncResult<A, E | ER | Cause.NoSuchElementError>>
657 <A, E>(stream: Stream.Stream<A, E, AtomRegistry | Reactivity.Reactivity | R>, options?: {
658 readonly initialValue?: A
659 }): Atom<AsyncResult.AsyncResult<A, E | ER | Cause.NoSuchElementError>>
660 }
661
662 readonly fn: {
663 <Arg>(): {
664 <E, A>(
665 fn: (arg: Arg, get: FnContext) => Effect.Effect<A, E, Scope.Scope | AtomRegistry | Reactivity.Reactivity | R>,
666 options?: {
667 readonly initialValue?: A | undefined
668 readonly reactivityKeys?: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>> | undefined
669 readonly concurrent?: boolean | undefined
670 }
671 ): AtomResultFn<Arg, A, E | ER>
672 <E, A>(
673 fn: (arg: Arg, get: FnContext) => Stream.Stream<A, E, AtomRegistry | Reactivity.Reactivity | R>,
674 options?: {
675 readonly initialValue?: A | undefined
676 readonly reactivityKeys?: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>> | undefined
677 readonly concurrent?: boolean | undefined
678 }
679 ): AtomResultFn<Arg, A, E | ER | Cause.NoSuchElementError>
680 }
681 <E, A, Arg = void>(
682 fn: (arg: Arg, get: FnContext) => Effect.Effect<A, E, Scope.Scope | AtomRegistry | Reactivity.Reactivity | R>,
683 options?: {
684 readonly initialValue?: A | undefined
685 readonly reactivityKeys?: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>> | undefined
686 readonly concurrent?: boolean | undefined
687 }
688 ): AtomResultFn<Arg, A, E | ER>
689 <E, A, Arg = void>(
690 fn: (arg: Arg, get: FnContext) => Stream.Stream<A, E, AtomRegistry | Reactivity.Reactivity | R>,
691 options?: {
692 readonly initialValue?: A | undefined
693 readonly reactivityKeys?: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>> | undefined
694 readonly concurrent?: boolean | undefined
695 }
696 ): AtomResultFn<Arg, A, E | ER | Cause.NoSuchElementError>
697 }
698
699 readonly pull: <A, E>(
700 create:
701 | ((get: AtomContext) => Stream.Stream<A, E, R | AtomRegistry | Reactivity.Reactivity>)
702 | Stream.Stream<A, E, R | AtomRegistry | Reactivity.Reactivity>,
703 options?: {
704 readonly disableAccumulation?: boolean
705 readonly initialValue?: ReadonlyArray<A>
706 }
707 ) => Writable<PullResult<A, E | ER>, void>
708
709 readonly subscriptionRef: <A, E>(
710 create:
711 | Effect.Effect<SubscriptionRef.SubscriptionRef<A>, E, Scope.Scope | R | AtomRegistry | Reactivity.Reactivity>
712 | ((
713 get: AtomContext
714 ) => Effect.Effect<SubscriptionRef.SubscriptionRef<A>, E, Scope.Scope | R | AtomRegistry | Reactivity.Reactivity>)
715 ) => Writable<AsyncResult.AsyncResult<A, E>, A>
716}
717
718/**
719 * Factory for `AtomRuntime` values that share a set of global layers.
720 *
721 * @stability unstable
722 * @category models
723 * @since 4.0.0
724 */
725export interface RuntimeFactory {
726 <R, E>(
727 create:
728 | Layer.Layer<R, E, AtomRegistry | Reactivity.Reactivity>
729 | ((get: AtomContext) => Layer.Layer<R, E, AtomRegistry | Reactivity.Reactivity>)
730 ): AtomRuntime<R, E>
731 readonly addGlobalLayer: <A, E>(layer: Layer.Layer<A, E, AtomRegistry | Reactivity.Reactivity>) => void
732
733 /**
734 * Uses the `Reactivity` service from the runtime to refresh the atom whenever
735 * the keys change.
736 */
737 readonly withReactivity: (
738 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>
739 ) => <A extends Atom<any>>(atom: A) => A
740}
741
742/**
743 * A `RuntimeFactory` backed by an atom whose memo map is scoped to each registry.
744 *
745 * @stability unstable
746 * @category models
747 * @since 4.0.0
748 */
749export interface RegistryRuntimeFactory extends RuntimeFactory {
750 readonly memoMap: Atom<Layer.MemoMap>
751}
752
753/**
754 * A `RuntimeFactory` backed by a concrete memo map shared across registries.
755 *
756 * @stability unstable
757 * @category models
758 * @since 4.0.0
759 */
760export interface SharedRuntimeFactory extends RuntimeFactory {
761 readonly memoMap: Layer.MemoMap
762}
763
764/**
765 * Creates a `RuntimeFactory` backed by a registry-scoped memo map by default,
766 * or by the supplied atom or concrete `Layer.MemoMap`.
767 *
768 * @stability unstable
769 * @category constructors
770 * @since 4.0.0
771 */
772export function context(): RegistryRuntimeFactory
773/**
774 * @stability unstable
775 */
776export function context(options: { readonly memoMap: Atom<Layer.MemoMap> }): RegistryRuntimeFactory
777/**
778 * @stability unstable
779 */
780export function context(options: { readonly memoMap: Layer.MemoMap }): SharedRuntimeFactory
781/**
782 * @stability unstable
783 */
784export function context(options?: {
785 readonly memoMap: Atom<Layer.MemoMap> | Layer.MemoMap
786}): RegistryRuntimeFactory | SharedRuntimeFactory {
787 const memoMap = options?.memoMap ?? removeTtl(make(() => Layer.makeMemoMapUnsafe()))
788 const resolveMemoMap = (get: AtomContext): Layer.MemoMap => isAtom(memoMap) ? get(memoMap) : memoMap
789 let globalLayer: Layer.Layer<any, any, AtomRegistry> = Reactivity.layer
790 function factory<E, R>(
791 create:
792 | Layer.Layer<R, E, AtomRegistry | Reactivity.Reactivity>
793 | ((get: AtomContext) => Layer.Layer<R, E, AtomRegistry | Reactivity.Reactivity>)
794 ): AtomRuntime<R, E> {
795 const self = Object.create(RuntimeProto)
796 self.keepAlive = false
797 self.lazy = true
798 self.refresh = undefined
799 self.factory = factory
800
801 const layerAtom = keepAlive(
802 typeof create === "function"
803 ? readable((get) => Layer.provideMerge(create(get), globalLayer))
804 : readable(() => Layer.provideMerge(create, globalLayer))
805 )
806 self.layer = layerAtom
807
808 self.read = function read(get: AtomContext) {
809 const layer = get(layerAtom)
810 const build = Effect.flatMap(Effect.scope, (scope) => Layer.buildWithMemoMap(layer, resolveMemoMap(get), scope))
811 return effect(get, build, { uninterruptible: true })
812 }
813
814 return self
815 }
816 factory.memoMap = memoMap
817 factory.addGlobalLayer = (layer: Layer.Layer<any, any, AtomRegistry | Reactivity.Reactivity>) => {
818 globalLayer = Layer.provideMerge(globalLayer, Layer.provide(layer, Reactivity.layer))
819 }
820 const reactivityAtom = removeTtl(
821 make((get) =>
822 Effect.contextWith((services: Context.Context<Scope.Scope>) =>
823 Layer.buildWithMemoMap(Reactivity.layer, resolveMemoMap(get), Context.get(services, Scope.Scope))
824 ).pipe(
825 Effect.map(Context.get(Reactivity.Reactivity))
826 )
827 )
828 )
829 factory.withReactivity =
830 (keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>) =>
831 <A extends Atom<any>>(atom: A): A =>
832 transform(atom, (get) => {
833 const reactivity = AsyncResult.getOrThrow(get(reactivityAtom))
834 get.addFinalizer(reactivity.registerUnsafe(keys, () => {
835 get.refresh(atom)
836 }))
837 get.subscribe(atom, (value) => get.setSelf(value))
838 return get.once(atom)
839 }, { initialValueTarget: atom }) as any as A
840 return factory as any
841}
842
843/**
844 * Default registry-scoped `RuntimeFactory`.
845 *
846 * @stability unstable
847 * @category context
848 * @since 4.0.0
849 */
850export const runtime: RegistryRuntimeFactory = context()
851
852/**
853 * Returns `Rx.runtime.withReactivity` for refreshing an atom whenever the
854 * keys change in the `Reactivity` service.
855 *
856 * **When to use**
857 *
858 * Use to refresh an atom whenever one or more invalidation keys change in the
859 * default reactivity runtime.
860 *
861 * @stability unstable
862 * @category reactivity
863 * @since 4.0.0
864 */
865export const withReactivity: (
866 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>
867) => <A extends Atom<any>>(atom: A) => A = runtime.withReactivity
868
869// -----------------------------------------------------------------------------
870// constructors - stream
871// -----------------------------------------------------------------------------
872
873const stream = <A, E>(
874 get: AtomContext,
875 stream: Stream.Stream<A, E, AtomRegistry>,
876 options?: {
877 readonly initialValue?: A
878 },
879 services?: Context.Context<any>
880): AsyncResult.AsyncResult<A, E | Cause.NoSuchElementError> => {
881 const initialValue = options?.initialValue !== undefined
882 ? AsyncResult.success<A, E>(options.initialValue)
883 : AsyncResult.initial<A, E>()
884 return makeStream(get, stream, initialValue, services)
885}
886
887function makeStream<A, E>(
888 ctx: AtomContext,
889 stream: Stream.Stream<A, E, AtomRegistry>,
890 initialValue: AsyncResult.AsyncResult<A, E | Cause.NoSuchElementError>,
891 services = Context.empty()
892): AsyncResult.AsyncResult<A, E | Cause.NoSuchElementError> {
893 const previous = ctx.self<AsyncResult.AsyncResult<A, E | Cause.NoSuchElementError>>()
894 services = Context.add(services, AtomRegistry, ctx.registry)
895
896 const run = Effect.scopedWith((scope) =>
897 Effect.flatMap(Channel.toPullScoped(stream.channel, scope), (pull) =>
898 Effect.whileLoop({
899 while: constTrue,
900 body: () => pull,
901 step(arr) {
902 ctx.setSelf(AsyncResult.success(Arr.lastNonEmpty(arr), {
903 waiting: true
904 }))
905 }
906 }))
907 ).pipe(
908 Effect.catchCause((cause) => {
909 if (Pull.isDoneCause(cause)) {
910 pipe(
911 ctx.self<AsyncResult.AsyncResult<A, E | Cause.NoSuchElementError>>(),
912 Option.flatMap(AsyncResult.value),
913 Option.match({
914 onNone: () =>
915 ctx.setSelf(
916 AsyncResult.failWithPrevious(new Cause.NoSuchElementError(), {
917 previous: ctx.self<AsyncResult.AsyncResult<A, E | Cause.NoSuchElementError>>()
918 })
919 ),
920 onSome: (a) => ctx.setSelf(AsyncResult.success(a))
921 })
922 )
923 } else {
924 ctx.setSelf(AsyncResult.failureWithPrevious(cause as Cause.Cause<E>, {
925 previous: ctx.self<AsyncResult.AsyncResult<A, E | Cause.NoSuchElementError>>()
926 }))
927 }
928 return Effect.void
929 })
930 )
931
932 const cancel = runCallbackSync(
933 services.pipe(
934 Context.add(AtomRegistry, ctx.registry),
935 Context.add(Scheduler.Scheduler, ctx.registry.scheduler)
936 ),
937 run,
938 constVoid,
939 false
940 )
941 if (cancel !== undefined) {
942 ctx.addFinalizer(cancel)
943 }
944
945 if (previous._tag === "Some") {
946 return AsyncResult.waitingFrom(previous)
947 }
948 return AsyncResult.waiting(initialValue)
949}
950
951// -----------------------------------------------------------------------------
952// constructors - subscription ref
953// -----------------------------------------------------------------------------
954
955/**
956 * Creates a writable atom backed by a `SubscriptionRef`, or by an effect that produces one, updating from ref changes and writing atom updates back to the ref.
957 *
958 * @stability unstable
959 * @category constructors
960 * @since 4.0.0
961 */
962export const subscriptionRef: {
963 <A>(ref: SubscriptionRef.SubscriptionRef<A> | ((get: AtomContext) => SubscriptionRef.SubscriptionRef<A>)): Writable<A>
964 <A, E>(
965 effect:
966 | Effect.Effect<SubscriptionRef.SubscriptionRef<A>, E, Scope.Scope | AtomRegistry>
967 | ((get: AtomContext) => Effect.Effect<SubscriptionRef.SubscriptionRef<A>, E, Scope.Scope | AtomRegistry>)
968 ): Writable<AsyncResult.AsyncResult<A, E>, A>
969} = (
970 ref:
971 | SubscriptionRef.SubscriptionRef<any>
972 | ((get: AtomContext) => SubscriptionRef.SubscriptionRef<any>)
973 | Effect.Effect<SubscriptionRef.SubscriptionRef<any>, any, Scope.Scope | AtomRegistry>
974 | ((get: AtomContext) => Effect.Effect<SubscriptionRef.SubscriptionRef<any>, any, Scope.Scope | AtomRegistry>)
975) =>
976 makeSubRef(
977 readable((get) => {
978 const value = typeof ref === "function" ? ref(get) : ref
979 return SubscriptionRef.isSubscriptionRef(value)
980 ? value
981 : makeEffect(get, value, AsyncResult.initial(true))
982 }),
983 readSubscriptionRef
984 ) as any
985
986const readSubscriptionRef = (
987 get: AtomContext,
988 sub:
989 | SubscriptionRef.SubscriptionRef<any>
990 | AsyncResult.AsyncResult<SubscriptionRef.SubscriptionRef<any>, any>,
991 services = Context.empty()
992) => {
993 if (SubscriptionRef.isSubscriptionRef(sub)) {
994 get.addFinalizer(
995 SubscriptionRef.changes(sub).pipe(
996 Stream.runForEachArray((arr) => {
997 for (let i = 0; i < arr.length; i++) {
998 get.setSelf(arr[i])
999 }
1000 return Effect.void
1001 }),
1002 Effect.runCallbackWith(services)
1003 )
1004 )
1005 return Effect.runSyncWith(services)(SubscriptionRef.get(sub))
1006 } else if (sub._tag !== "Success") {
1007 return sub
1008 }
1009 return makeStream(get, SubscriptionRef.changes(sub.value), AsyncResult.initial(true), services)
1010}
1011
1012const makeSubRef = (
1013 refAtom: Atom<
1014 SubscriptionRef.SubscriptionRef<any> | AsyncResult.AsyncResult<SubscriptionRef.SubscriptionRef<any>, any>
1015 >,
1016 read: (
1017 get: AtomContext,
1018 ref: SubscriptionRef.SubscriptionRef<any> | AsyncResult.Success<SubscriptionRef.SubscriptionRef<any>, any>
1019 ) => any
1020) => {
1021 function write(ctx: WriteContext<SubscriptionRef.SubscriptionRef<any>>, value: any) {
1022 const ref = ctx.get(refAtom)
1023 if (SubscriptionRef.isSubscriptionRef(ref)) {
1024 Effect.runSync(SubscriptionRef.set(ref, value))
1025 } else if (AsyncResult.isSuccess(ref)) {
1026 Effect.runSync(SubscriptionRef.set(ref.value, value))
1027 }
1028 }
1029 return writable((get) => {
1030 const ref = get(refAtom)
1031 if (SubscriptionRef.isSubscriptionRef(ref)) {
1032 return read(get, ref)
1033 } else if (AsyncResult.isSuccess(ref)) {
1034 return read(get, ref)
1035 }
1036 return ref
1037 }, write)
1038}
1039
1040// -----------------------------------------------------------------------------
1041// constructors - functions
1042// -----------------------------------------------------------------------------
1043
1044/**
1045 * Context passed to `fn` and `fnSync` computations for reading atoms, awaiting results, registering finalizers, refreshing atoms, subscribing to changes, and writing updates.
1046 *
1047 * @stability unstable
1048 * @category models
1049 * @since 4.0.0
1050 */
1051export interface FnContext {
1052 <A>(atom: Atom<A>): A
1053 result<A, E>(this: FnContext, atom: Atom<AsyncResult.AsyncResult<A, E>>, options?: {
1054 readonly suspendOnWaiting?: boolean | undefined
1055 }): Effect.Effect<A, E>
1056 addFinalizer(this: FnContext, f: () => void): void
1057 mount<A>(this: FnContext, atom: Atom<A>): void
1058 refresh<A>(this: FnContext, atom: Atom<A>): void
1059 self<A>(this: FnContext): Option.Option<A>
1060 setSelf<A>(this: FnContext, a: A): void
1061 set<R, W>(this: FnContext, atom: Writable<R, W>, value: W): void
1062 setResult<A, E, W>(this: FnContext, atom: Writable<AsyncResult.AsyncResult<A, E>, W>, value: W): Effect.Effect<A, E>
1063 some<A>(this: FnContext, atom: Atom<Option.Option<A>>): Effect.Effect<A>
1064 stream<A>(this: FnContext, atom: Atom<A>, options?: {
1065 readonly withoutInitialValue?: boolean
1066 readonly bufferSize?: number
1067 }): Stream.Stream<A>
1068 streamResult<A, E>(this: FnContext, atom: Atom<AsyncResult.AsyncResult<A, E>>, options?: {
1069 readonly withoutInitialValue?: boolean
1070 readonly bufferSize?: number
1071 }): Stream.Stream<A, E>
1072 subscribe<A>(this: FnContext, atom: Atom<A>, f: (_: A) => void, options?: {
1073 readonly immediate?: boolean
1074 }): void
1075 readonly registry: Registry.AtomRegistry
1076}
1077
1078/**
1079 * Creates a writable atom for a synchronous function; writing an argument re-runs the function, returning `Option.none` before the first call unless an initial value is supplied.
1080 *
1081 * @stability unstable
1082 * @category constructors
1083 * @since 4.0.0
1084 */
1085export const fnSync: {
1086 <Arg>(): {
1087 <A>(
1088 f: (arg: Arg, get: FnContext) => A
1089 ): Writable<Option.Option<A>, Arg>
1090 <A>(
1091 f: (arg: Arg, get: FnContext) => A,
1092 options: { readonly initialValue: A }
1093 ): Writable<A, Arg>
1094 }
1095 <A, Arg = void>(
1096 f: (arg: Arg, get: FnContext) => A
1097 ): Writable<Option.Option<A>, Arg>
1098 <A, Arg = void>(
1099 f: (arg: Arg, get: FnContext) => A,
1100 options: { readonly initialValue: A }
1101 ): Writable<A, Arg>
1102} = function(...args: ReadonlyArray<any>) {
1103 if (args.length === 0) {
1104 return makeFnSync
1105 }
1106 return makeFnSync(...args as [any, any]) as any
1107}
1108
1109const makeFnSync = <Arg, A>(f: (arg: Arg, get: FnContext) => A, options?: {
1110 readonly initialValue?: A
1111}): Writable<Option.Option<A> | A, Arg> => {
1112 const argAtom = removeTtl(state<[number, Arg]>([0, undefined as any]))
1113 const hasInitialValue = options?.initialValue !== undefined
1114 return writable(function(get) {
1115 ;(get as any).isFn = true
1116 const [counter, arg] = get.get(argAtom)
1117 if (counter === 0) {
1118 return hasInitialValue ? options.initialValue : Option.none()
1119 }
1120 return hasInitialValue ? f(arg, get) : Option.some(f(arg, get))
1121 }, function(ctx, arg) {
1122 batch(() => {
1123 ctx.set(argAtom, [ctx.get(argAtom)[0] + 1, arg as Arg])
1124 ctx.refreshSelf()
1125 })
1126 })
1127}
1128
1129/**
1130 * Writable async function atom whose value is an `AsyncResult` and whose writes accept function arguments plus `Reset` and `Interrupt` controls.
1131 *
1132 * @stability unstable
1133 * @category models
1134 * @since 4.0.0
1135 */
1136export interface AtomResultFn<Arg, A, E = never>
1137 extends Writable<AsyncResult.AsyncResult<A, E>, Arg | Reset | Interrupt>
1138{}
1139
1140/**
1141 * Defines the control symbol that can be written to an `AtomResultFn` to reset it to its initial state.
1142 *
1143 * **When to use**
1144 *
1145 * Use when you need an `AtomResultFn` write value that clears the current async
1146 * result and returns it to the initial state.
1147 *
1148 * @stability unstable
1149 * @category symbols
1150 * @since 4.0.0
1151 */
1152export const Reset = Symbol.for("effect/reactivity/atom/Atom/Reset")
1153
1154/**
1155 * Type of the `Reset` control symbol accepted by `AtomResultFn` writes.
1156 *
1157 * @stability unstable
1158 * @category symbols
1159 * @since 4.0.0
1160 */
1161export type Reset = typeof Reset
1162
1163/**
1164 * Defines the control symbol that can be written to an `AtomResultFn` to interrupt the current asynchronous computation.
1165 *
1166 * **When to use**
1167 *
1168 * Use when you need an `AtomResultFn` write value that interrupts the currently
1169 * running async computation.
1170 *
1171 * @stability unstable
1172 * @category symbols
1173 * @since 4.0.0
1174 */
1175export const Interrupt = Symbol.for("effect/reactivity/atom/Atom/Interrupt")
1176
1177/**
1178 * Type of the `Interrupt` control symbol accepted by `AtomResultFn` writes.
1179 *
1180 * @stability unstable
1181 * @category symbols
1182 * @since 4.0.0
1183 */
1184export type Interrupt = typeof Interrupt
1185
1186/**
1187 * Creates a writable atom for an `Effect` or `Stream` function; writing an argument starts the computation and exposes its state as an `AsyncResult`.
1188 *
1189 * @stability unstable
1190 * @category constructors
1191 * @since 4.0.0
1192 */
1193export const fn: {
1194 <Arg>(): <E, A>(fn: (arg: Arg, get: FnContext) => Effect.Effect<A, E, Scope.Scope | AtomRegistry>, options?: {
1195 readonly initialValue?: A | undefined
1196 readonly concurrent?: boolean | undefined
1197 }) => AtomResultFn<Arg, A, E>
1198 <E, A, Arg = void>(fn: (arg: Arg, get: FnContext) => Effect.Effect<A, E, Scope.Scope | AtomRegistry>, options?: {
1199 readonly initialValue?: A | undefined
1200 readonly concurrent?: boolean | undefined
1201 }): AtomResultFn<Arg, A, E>
1202 <Arg>(): <E, A>(fn: (arg: Arg, get: FnContext) => Stream.Stream<A, E, AtomRegistry>, options?: {
1203 readonly initialValue?: A | undefined
1204 readonly concurrent?: boolean | undefined
1205 }) => AtomResultFn<Arg, A, E | Cause.NoSuchElementError>
1206 <E, A, Arg = void>(fn: (arg: Arg, get: FnContext) => Stream.Stream<A, E, AtomRegistry>, options?: {
1207 readonly initialValue?: A | undefined
1208 readonly concurrent?: boolean | undefined
1209 }): AtomResultFn<Arg, A, E | Cause.NoSuchElementError>
1210} = function(...args: ReadonlyArray<any>) {
1211 if (args.length === 0) {
1212 return makeFn
1213 }
1214 return makeFn(...args as [any, any]) as any
1215}
1216
1217const makeFn = <Arg, E, A>(
1218 f: (arg: Arg, get: FnContext) => Stream.Stream<A, E, AtomRegistry> | Effect.Effect<A, E, Scope.Scope | AtomRegistry>,
1219 options?: {
1220 readonly initialValue?: A | undefined
1221 readonly concurrent?: boolean | undefined
1222 }
1223): AtomResultFn<Arg, A, E | Cause.NoSuchElementError> => {
1224 const [read, write] = makeResultFn(f, options)
1225 return writable(read, write) as any
1226}
1227
1228function makeResultFn<Arg, E, A>(
1229 f: (arg: Arg, get: FnContext) => Effect.Effect<A, E, Scope.Scope | AtomRegistry> | Stream.Stream<A, E, AtomRegistry>,
1230 options?: {
1231 readonly initialValue?: A
1232 readonly concurrent?: boolean | undefined
1233 }
1234) {
1235 const argAtom = removeTtl(state<[number, Arg | Interrupt]>([0, undefined as any]))
1236 const initialValue = options?.initialValue !== undefined
1237 ? AsyncResult.success<A, E>(options.initialValue)
1238 : AsyncResult.initial<A, E>()
1239 const fibersAtom = options?.concurrent
1240 ? removeTtl(readable((get) => {
1241 const fibers = new Set<Fiber.Fiber<any, any>>()
1242 get.addFinalizer(() => fibers.forEach((f) => f.interruptUnsafe()))
1243 return fibers
1244 }))
1245 : undefined
1246
1247 function read(
1248 get: AtomContext,
1249 services?: Context.Context<any>
1250 ): AsyncResult.AsyncResult<A, E | Cause.NoSuchElementError> {
1251 const fibers = fibersAtom ? get(fibersAtom) : undefined
1252 ;(get as any).isFn = true
1253 const [counter, arg] = get.get(argAtom)
1254 if (counter === 0) {
1255 return initialValue
1256 } else if (arg === Interrupt) {
1257 return AsyncResult.failureWithPrevious(Cause.interrupt(), { previous: get.self() })
1258 }
1259 let value = f(arg, get)
1260 if (EffectTypeId in value) {
1261 if (fibers) {
1262 const eff = value as Effect.Effect<A, E, Scope.Scope | AtomRegistry>
1263 value = Effect.flatMap(
1264 Effect.forkDetach(eff, { startImmediately: true }),
1265 (fiber) => {
1266 fibers.add(fiber)
1267 const runningFibers = Array.from(fibers)
1268 fiber.addObserver(() => fibers.delete(fiber))
1269 return Effect.map(Fiber.joinAll(runningFibers), (arr) => arr[0])
1270 }
1271 )
1272 }
1273 return makeEffect(get, value as any, initialValue, services, false)
1274 }
1275 return makeStream(get, value as any, initialValue, services)
1276 }
1277 function write(
1278 ctx: WriteContext<AsyncResult.AsyncResult<A, E | Cause.NoSuchElementError>>,
1279 arg: Arg | Reset | Interrupt
1280 ) {
1281 batch(() => {
1282 if (arg === Reset) {
1283 ctx.set(argAtom, [0, undefined as any])
1284 } else if (arg === Interrupt) {
1285 ctx.set(argAtom, [ctx.get(argAtom)[0] + 1, Interrupt])
1286 } else {
1287 ctx.set(argAtom, [ctx.get(argAtom)[0] + 1, arg])
1288 }
1289 ctx.refreshSelf()
1290 })
1291 }
1292 return [read, write, argAtom] as const
1293}
1294
1295/**
1296 * `AsyncResult` produced by `pull`, containing a non-empty batch of pulled items and a `done` flag, or `NoSuchElementError` when the stream completes without items.
1297 *
1298 * @stability unstable
1299 * @category models
1300 * @since 4.0.0
1301 */
1302export type PullResult<A, E = never> = AsyncResult.AsyncResult<{
1303 readonly done: boolean
1304 readonly items: Arr.NonEmptyArray<A>
1305}, E | Cause.NoSuchElementError>
1306
1307/**
1308 * Creates a writable atom that pulls an initial chunk from a stream and then pulls the next chunk whenever it is written to, accumulating items unless `disableAccumulation` is enabled.
1309 *
1310 * @stability unstable
1311 * @category constructors
1312 * @since 4.0.0
1313 */
1314export const pull = <A, E>(
1315 create: ((get: AtomContext) => Stream.Stream<A, E, AtomRegistry>) | Stream.Stream<A, E, AtomRegistry>,
1316 options?: {
1317 readonly disableAccumulation?: boolean | undefined
1318 }
1319): Writable<PullResult<A, E>, void> => {
1320 const pullSignal = removeTtl(state(0))
1321 const pullAtom = readable(makeRead(function(get) {
1322 return makeStreamPullEffect(get, pullSignal, create, options)
1323 }))
1324 return makeStreamPull(pullSignal, pullAtom)
1325}
1326
1327const makeStreamPullEffect = <A, E>(
1328 get: AtomContext,
1329 pullSignal: Atom<number>,
1330 create: Stream.Stream<A, E, AtomRegistry> | ((get: AtomContext) => Stream.Stream<A, E, AtomRegistry>),
1331 options?: {
1332 readonly disableAccumulation?: boolean | undefined
1333 }
1334): Effect.Effect<
1335 { readonly done: boolean; readonly items: Arr.NonEmptyArray<A> },
1336 E | Cause.NoSuchElementError,
1337 Scope.Scope | AtomRegistry
1338> =>
1339 Effect.flatMap(
1340 Stream.toPull(typeof create === "function" ? create(get) : create),
1341 (pullChunk) => {
1342 const fiber = Fiber.getCurrent()!
1343 const services = fiber.context as Context.Context<AtomRegistry | Scope.Scope>
1344 let acc: ReadonlyArray<A> = Arr.empty<A>()
1345 const pull: Effect.Effect<
1346 {
1347 done: boolean
1348 items: Arr.NonEmptyArray<A>
1349 },
1350 Cause.NoSuchElementError | E,
1351 Registry.AtomRegistry
1352 > = Effect.matchCauseEffect(pullChunk, {
1353 onFailure(cause): Effect.Effect<
1354 { done: boolean; items: Arr.NonEmptyArray<A> },
1355 Cause.NoSuchElementError | E
1356 > {
1357 if (Pull.isDoneCause(cause)) {
1358 if (!Arr.isReadonlyArrayNonEmpty(acc)) {
1359 return Effect.fail(new Cause.NoSuchElementError(`Atom.pull: no items`))
1360 }
1361 return Effect.succeed({ done: true, items: acc as Arr.NonEmptyArray<A> })
1362 }
1363 return Effect.failCause(cause as Cause.Cause<E>)
1364 },
1365 onSuccess(chunk) {
1366 let items: Arr.NonEmptyArray<A>
1367 if (options?.disableAccumulation) {
1368 items = chunk as any
1369 } else {
1370 items = Arr.appendAll(acc, chunk)
1371 acc = items
1372 }
1373 return Effect.succeed({ done: false, items })
1374 }
1375 })
1376
1377 const cancels = new Set<() => void>()
1378 get.addFinalizer(() => {
1379 for (const cancel of cancels) cancel()
1380 })
1381 get.once(pullSignal)
1382 get.subscribe(pullSignal, () => {
1383 get.setSelf(AsyncResult.waitingFrom(get.self<PullResult<A, E>>()))
1384 let cancel: (() => void) | undefined
1385 // eslint-disable-next-line prefer-const
1386 cancel = runCallbackSync(services, pull, (exit) => {
1387 if (cancel) cancels.delete(cancel)
1388 const result = AsyncResult.fromExitWithPrevious(exit, get.self())
1389 const pending = cancels.size > 0
1390 get.setSelf(pending ? AsyncResult.waiting(result) : result)
1391 })
1392 if (cancel) cancels.add(cancel)
1393 })
1394
1395 return pull
1396 }
1397 )
1398
1399const makeStreamPull = <A, E>(
1400 pullSignal: Writable<number>,
1401 pullAtom: Atom<PullResult<A, E>>
1402) =>
1403 writable(pullAtom.read, function(ctx, _) {
1404 ctx.set(pullSignal, ctx.get(pullSignal) + 1)
1405 })
1406
1407/**
1408 * Creates a memoized atom factory that returns the same object for the same argument, using weak references for cached values when the platform supports them.
1409 *
1410 * @stability unstable
1411 * @category constructors
1412 * @since 4.0.0
1413 */
1414export const family = typeof WeakRef === "undefined" || typeof FinalizationRegistry === "undefined" ?
1415 <Arg, T extends object>(
1416 f: (arg: Arg) => T
1417 ): (arg: Arg) => T => {
1418 const atoms = MutableHashMap.empty<Arg, T>()
1419 return function(arg) {
1420 const atomEntry = MutableHashMap.get(atoms, arg)
1421 if (atomEntry._tag === "Some") {
1422 return atomEntry.value
1423 }
1424 const newAtom = f(arg)
1425 MutableHashMap.set(atoms, arg, newAtom)
1426 return newAtom
1427 }
1428 } :
1429 <Arg, T extends object>(
1430 f: (arg: Arg) => T
1431 ): (arg: Arg) => T => {
1432 const atoms = MutableHashMap.empty<Arg, WeakRef<T>>()
1433 const registry = new FinalizationRegistry<Arg>((arg) => {
1434 const entry = MutableHashMap.get(atoms, arg)
1435 if (entry._tag === "Some" && entry.value.deref() === undefined) {
1436 MutableHashMap.remove(atoms, arg)
1437 }
1438 })
1439 return function(arg) {
1440 const atomEntry = MutableHashMap.get(atoms, arg).pipe(
1441 Option.flatMapNullishOr((ref) => ref.deref())
1442 )
1443
1444 if (atomEntry._tag === "Some") {
1445 return atomEntry.value
1446 }
1447 const newAtom = f(arg)
1448 MutableHashMap.set(atoms, arg, new WeakRef(newAtom))
1449 registry.register(newAtom, arg)
1450 return newAtom
1451 }
1452 }
1453
1454/**
1455 * Uses a fallback `AsyncResult` atom while the primary atom is `Initial`, marking the fallback result as waiting until the primary atom produces a non-initial result.
1456 *
1457 * @stability unstable
1458 * @category combinators
1459 * @since 4.0.0
1460 */
1461export const withFallback: {
1462 <E2, A2>(
1463 fallback: Atom<AsyncResult.AsyncResult<A2, E2>>
1464 ): <R extends Atom<AsyncResult.AsyncResult<any, any>>>(
1465 self: R
1466 ) => [R] extends [Writable<infer _, infer RW>] ? Writable<
1467 AsyncResult.AsyncResult<
1468 AsyncResult.AsyncResult.Success<Type<R>> | A2,
1469 AsyncResult.AsyncResult.Failure<Type<R>> | E2
1470 >,
1471 RW
1472 >
1473 : Atom<
1474 AsyncResult.AsyncResult<
1475 AsyncResult.AsyncResult.Success<Type<R>> | A2,
1476 AsyncResult.AsyncResult.Failure<Type<R>> | E2
1477 >
1478 >
1479 <R extends Atom<AsyncResult.AsyncResult<any, any>>, A2, E2>(
1480 self: R,
1481 fallback: Atom<AsyncResult.AsyncResult<A2, E2>>
1482 ): [R] extends [Writable<infer _, infer RW>] ? Writable<
1483 AsyncResult.AsyncResult<
1484 AsyncResult.AsyncResult.Success<Type<R>> | A2,
1485 AsyncResult.AsyncResult.Failure<Type<R>> | E2
1486 >,
1487 RW
1488 >
1489 : Atom<
1490 AsyncResult.AsyncResult<
1491 AsyncResult.AsyncResult.Success<Type<R>> | A2,
1492 AsyncResult.AsyncResult.Failure<Type<R>> | E2
1493 >
1494 >
1495} = dual(2, <R extends Atom<AsyncResult.AsyncResult<any, any>>, A2, E2>(
1496 self: R,
1497 fallback: Atom<AsyncResult.AsyncResult<A2, E2>>
1498): [R] extends [Writable<infer _, infer RW>] ? Writable<
1499 AsyncResult.AsyncResult<
1500 AsyncResult.AsyncResult.Success<Type<R>> | A2,
1501 AsyncResult.AsyncResult.Failure<Type<R>> | E2
1502 >,
1503 RW
1504 >
1505 : Atom<
1506 AsyncResult.AsyncResult<
1507 AsyncResult.AsyncResult.Success<Type<R>> | A2,
1508 AsyncResult.AsyncResult.Failure<Type<R>> | E2
1509 >
1510 > =>
1511{
1512 function withFallback(get: AtomContext) {
1513 const result = get(self)
1514 if (result._tag === "Initial") {
1515 return AsyncResult.waiting(get(fallback))
1516 }
1517 return result
1518 }
1519 return isWritable(self)
1520 ? writable(
1521 withFallback,
1522 function(ctx, value) {
1523 ctx.set(self, value)
1524 },
1525 self.refresh ?? function(refresh) {
1526 refresh(self)
1527 }
1528 ) as any
1529 : readable(
1530 withFallback,
1531 self.refresh ?? function(refresh) {
1532 refresh(self)
1533 }
1534 ) as any
1535})
1536
1537/**
1538 * Returns a copy of an atom that remains cached and mounted even when no subscribers are using it.
1539 *
1540 * @stability unstable
1541 * @category combinators
1542 * @since 4.0.0
1543 */
1544export const keepAlive = <A extends Atom<any>>(self: A): A =>
1545 Object.assign(Object.create(Object.getPrototypeOf(self)), {
1546 ...self,
1547 keepAlive: true
1548 })
1549
1550/**
1551 * Allows a reactive value to be disposed of when it is not in use.
1552 *
1553 * **Details**
1554 *
1555 * Atoms have this behavior by default, so use this to undo `keepAlive` on a copied atom.
1556 *
1557 * @stability unstable
1558 * @category combinators
1559 * @since 4.0.0
1560 */
1561export const autoDispose = <A extends Atom<any>>(self: A): A =>
1562 Object.assign(Object.create(Object.getPrototypeOf(self)), {
1563 ...self,
1564 keepAlive: false
1565 })
1566
1567/**
1568 * Sets whether an atom should be lazy.
1569 *
1570 * **Details**
1571 *
1572 * Lazy atoms defer recomputation while they have no active listeners or active
1573 * non-lazy dependents, rebuilding the next time their value is observed.
1574 *
1575 * @stability unstable
1576 * @category combinators
1577 * @since 4.0.0
1578 */
1579export const setLazy: {
1580 (lazy: boolean): <A extends Atom<any>>(self: A) => A
1581 <A extends Atom<any>>(self: A, lazy: boolean): A
1582} = dual(2, <A extends Atom<any>>(self: A, lazy: boolean) =>
1583 Object.assign(Object.create(Object.getPrototypeOf(self)), {
1584 ...self,
1585 lazy
1586 }))
1587
1588/**
1589 * Returns a copy of an atom that uses a custom equality function to detect
1590 * value changes.
1591 *
1592 * **Details**
1593 *
1594 * When an atom's value is rebuilt or written, the registry compares the new
1595 * value against the current one to decide whether dependents and listeners
1596 * should be notified. By default the comparison uses `Object.is`, so a
1597 * structurally equal but referentially distinct value still triggers
1598 * notifications. Providing an equality function lets the atom skip updates
1599 * when the new value is equal to the current one.
1600 *
1601 * **Example** (Comparing values structurally)
1602 *
1603 * ```ts import.meta.vitest
1604 * import { Atom } from "effect/reactivity"
1605 *
1606 * const point = Atom.make({ x: 0, y: 0 }).pipe(
1607 * Atom.withEquality<{ x: number; y: number }>((a, b) => a.x === b.x && a.y === b.y)
1608 * )
1609 * point.equals({ x: 1, y: 2 }, { x: 1, y: 2 }) // => true
1610 * ```
1611 *
1612 * @stability unstable
1613 * @category combinators
1614 * @since 4.0.0
1615 */
1616export const withEquality: {
1617 <A>(equals: (value: A, next: A) => boolean): <T extends Atom<A>>(self: T) => T
1618 <T extends Atom<any>>(self: T, equals: (value: Type<T>, next: Type<T>) => boolean): T
1619} = dual(
1620 2,
1621 <T extends Atom<any>>(self: T, equals: (value: Type<T>, next: Type<T>) => boolean): T =>
1622 Object.assign(Object.create(Object.getPrototypeOf(self)), {
1623 ...self,
1624 equals
1625 })
1626)
1627
1628/**
1629 * Attaches a diagnostic label to an atom.
1630 *
1631 * **Details**
1632 *
1633 * The label is used for inspection and debugging metadata and does not change the
1634 * atom's read or write behavior.
1635 *
1636 * @stability unstable
1637 * @category combinators
1638 * @since 4.0.0
1639 */
1640export const withLabel: {
1641 (name: string): <A extends Atom<any>>(self: A) => A
1642 <A extends Atom<any>>(self: A, name: string): A
1643} = dual<
1644 (name: string) => <A extends Atom<any>>(self: A) => A,
1645 <A extends Atom<any>>(self: A, name: string) => A
1646>(2, (self, name) =>
1647 Object.assign(Object.create(Object.getPrototypeOf(self)), {
1648 ...self,
1649 label: [name, getStackTraceLimit() === 0 ? "" : new Error().stack?.split("\n")[5] ?? ""]
1650 }))
1651
1652/**
1653 * Pairs an atom with an initial value for registry initialization.
1654 *
1655 * **When to use**
1656 *
1657 * Use to preload an atom value when constructing or seeding a registry.
1658 *
1659 * **Details**
1660 *
1661 * The returned tuple can be supplied to `AtomRegistry` initial values so the atom
1662 * starts with the provided value before it is first rebuilt.
1663 *
1664 * @stability unstable
1665 * @category combinators
1666 * @since 4.0.0
1667 */
1668export const initialValue: {
1669 <A>(initialValue: A): (self: Atom<A>) => readonly [Atom<A>, A]
1670 <A>(self: Atom<A>, initialValue: A): readonly [Atom<A>, A]
1671} = dual<
1672 <A>(initialValue: A) => (self: Atom<A>) => readonly [Atom<A>, A],
1673 <A>(self: Atom<A>, initialValue: A) => readonly [Atom<A>, A]
1674>(2, (self, initialValue) => [self, initialValue])
1675
1676/**
1677 * Creates a derived atom by reading another atom with a custom `AtomContext`
1678 * function.
1679 *
1680 * **Details**
1681 *
1682 * If the source is writable, the derived atom keeps the source write input and
1683 * forwards writes to the source. `initialValueTarget` controls which atom receives
1684 * preloaded initial values for the derived atom.
1685 *
1686 * @stability unstable
1687 * @category combinators
1688 * @since 4.0.0
1689 */
1690export const transform: {
1691 <R extends Atom<any>, B>(
1692 f: (get: AtomContext, atom: R) => B,
1693 options?: {
1694 readonly initialValueTarget?: Atom<B> | undefined
1695 }
1696 ): (self: R) => [R] extends [Writable<infer _, infer RW>] ? Writable<B, RW> : Atom<B>
1697 <R extends Atom<any>, B>(
1698 self: R,
1699 f: (get: AtomContext, atom: R) => B,
1700 options?: {
1701 readonly initialValueTarget?: Atom<B> | undefined
1702 }
1703 ): [R] extends [Writable<infer _, infer RW>] ? Writable<B, RW> : Atom<B>
1704} = dual(
1705 (args) => isAtom(args[0]),
1706 (<A, B>(
1707 self: Atom<A>,
1708 f: (get: AtomContext, atom: Atom<A>, options?: {
1709 readonly initialValueTarget?: Atom<B> | undefined
1710 }) => B,
1711 options?: {
1712 readonly initialValueTarget?: Atom<B> | undefined
1713 }
1714 ): Atom<B> => {
1715 const atom = removeTtl(
1716 isWritable(self)
1717 ? writable(
1718 (get) => f(get, self),
1719 function(ctx, value) {
1720 ctx.set(self, value)
1721 },
1722 self.refresh ?? function(refresh) {
1723 refresh(self)
1724 }
1725 )
1726 : readable(
1727 (get) => f(get, self),
1728 self.refresh ?? function(refresh) {
1729 refresh(self)
1730 }
1731 )
1732 )
1733 if (options?.initialValueTarget) {
1734 ;(atom as Mutable<Atom<B>>).initialValueTarget = getInitialValueTarget(options.initialValueTarget)
1735 }
1736 return atom
1737 }) as any
1738)
1739
1740const getInitialValueTarget = <A>(atom: Atom<A>): Atom<A> => {
1741 let target = atom
1742 while (target.initialValueTarget) {
1743 target = target.initialValueTarget
1744 }
1745 return target
1746}
1747
1748/**
1749 * Maps the current value of an atom with a pure function.
1750 *
1751 * **Details**
1752 *
1753 * When the source atom is writable, the returned atom remains writable and keeps
1754 * the source atom's write input type.
1755 *
1756 * @stability unstable
1757 * @category combinators
1758 * @since 4.0.0
1759 */
1760export const map: {
1761 <R extends Atom<any>, B>(
1762 f: (_: Type<R>) => B
1763 ): (self: R) => [R] extends [Writable<infer _, infer RW>] ? Writable<B, RW> : Atom<B>
1764 <R extends Atom<any>, B>(
1765 self: R,
1766 f: (_: Type<R>) => B
1767 ): [R] extends [Writable<infer _, infer RW>] ? Writable<B, RW> : Atom<B>
1768} = dual(
1769 2,
1770 <A, B>(self: Atom<A>, f: (_: A) => B): Atom<B> => transform(self, (get) => f(get(self)))
1771)
1772
1773/**
1774 * Maps the successful value inside an `AsyncResult` atom.
1775 *
1776 * **Details**
1777 *
1778 * Initial and failure states are preserved, and writable source atoms keep their
1779 * original write input type.
1780 *
1781 * @stability unstable
1782 * @category combinators
1783 * @since 4.0.0
1784 */
1785export const mapResult: {
1786 <R extends Atom<AsyncResult.AsyncResult<any, any>>, B>(
1787 f: (_: AsyncResult.AsyncResult.Success<Type<R>>) => B
1788 ): (
1789 self: R
1790 ) => [R] extends [Writable<infer _, infer RW>] ?
1791 Writable<AsyncResult.AsyncResult<B, AsyncResult.AsyncResult.Failure<Type<R>>>, RW>
1792 : Atom<AsyncResult.AsyncResult<B, AsyncResult.AsyncResult.Failure<Type<R>>>>
1793 <R extends Atom<AsyncResult.AsyncResult<any, any>>, B>(
1794 self: R,
1795 f: (_: AsyncResult.AsyncResult.Success<Type<R>>) => B
1796 ): [R] extends [Writable<infer _, infer RW>] ?
1797 Writable<AsyncResult.AsyncResult<B, AsyncResult.AsyncResult.Failure<Type<R>>>, RW>
1798 : Atom<AsyncResult.AsyncResult<B, AsyncResult.AsyncResult.Failure<Type<R>>>>
1799} = dual(2, <R extends Atom<AsyncResult.AsyncResult<any, any>>, B>(
1800 self: R,
1801 f: (_: AsyncResult.AsyncResult.Success<Type<R>>) => B
1802): [R] extends [Writable<infer _, infer RW>] ?
1803 Writable<AsyncResult.AsyncResult<B, AsyncResult.AsyncResult.Failure<Type<R>>>, RW>
1804 : Atom<AsyncResult.AsyncResult<B, AsyncResult.AsyncResult.Failure<Type<R>>>> => map(self, AsyncResult.map(f)))
1805
1806/**
1807 * Creates an atom that publishes source changes only after the source has stopped
1808 * changing for the specified duration.
1809 *
1810 * **Details**
1811 *
1812 * The current source value is used immediately, and any pending debounce timer is
1813 * cleared when the derived atom is disposed.
1814 *
1815 * @stability unstable
1816 * @category combinators
1817 * @since 4.0.0
1818 */
1819export const debounce: {
1820 (duration: Duration.Input): <A extends Atom<any>>(self: A) => WithoutSerializable<A>
1821 <A extends Atom<any>>(self: A, duration: Duration.Input): WithoutSerializable<A>
1822} = dual(
1823 2,
1824 <A>(self: Atom<A>, duration: Duration.Input): Atom<A> => {
1825 const millis = Duration.toMillis(Duration.fromInputUnsafe(duration))
1826 return transform(self, function(get) {
1827 let timeout: number | undefined
1828 let value = get.once(self)
1829 function update() {
1830 timeout = undefined
1831 get.setSelf(value)
1832 }
1833 get.addFinalizer(function() {
1834 if (timeout) clearTimeout(timeout)
1835 })
1836 get.subscribe(self, function(val) {
1837 value = val
1838 if (timeout) clearTimeout(timeout)
1839 timeout = setTimeout(update, millis) as any
1840 })
1841 return value
1842 }, { initialValueTarget: self })
1843 }
1844)
1845
1846/**
1847 * Creates a derived atom that reads the source and schedules a refresh after the
1848 * specified duration.
1849 *
1850 * **Details**
1851 *
1852 * The scheduled refresh is canceled when the derived atom's lifetime is disposed.
1853 *
1854 * @stability unstable
1855 * @category combinators
1856 * @since 4.0.0
1857 */
1858export const withRefresh: {
1859 (duration: Duration.Input): <A extends Atom<any>>(self: A) => WithoutSerializable<A>
1860 <A extends Atom<any>>(self: A, duration: Duration.Input): WithoutSerializable<A>
1861} = dual(
1862 2,
1863 <A>(self: Atom<A>, duration: Duration.Input): Atom<A> => {
1864 const millis = Duration.toMillis(Duration.fromInputUnsafe(duration))
1865 return transform(self, function(get) {
1866 const handle = setTimeout(() => get.refresh(self), millis) as any
1867 get.addFinalizer(() => clearTimeout(handle))
1868 return get(self)
1869 }, { initialValueTarget: self })
1870 }
1871)
1872
1873/**
1874 * Adds stale-while-revalidate refresh behavior to an async result atom.
1875 *
1876 * **Details**
1877 *
1878 * Returns the current result and defers stale-source revalidation until after the read.
1879 * Skips revalidation if the source becomes fresh or the atom is disposed.
1880 * Manual refreshes still forward to the source. `revalidateOnMount` controls
1881 * initial stale refreshes; `revalidateOnFocus` controls focus refreshes
1882 * (`true` respects `staleTime`, `"always"` forces refresh).
1883 *
1884 * @stability unstable
1885 * @category combinators
1886 * @since 4.0.0
1887 */
1888export const swr: {
1889 (
1890 options: {
1891 readonly staleTime: Duration.Input
1892 readonly revalidateOnMount?: boolean | undefined
1893 readonly revalidateOnFocus?: boolean | "always" | undefined
1894 readonly focusSignal?: Atom<any> | undefined
1895 }
1896 ): <R extends Atom<AsyncResult.AsyncResult<any, any>>>(self: R) => WithoutSerializable<R>
1897 <R extends Atom<AsyncResult.AsyncResult<any, any>>>(
1898 self: R,
1899 options: {
1900 readonly staleTime: Duration.Input
1901 readonly revalidateOnMount?: boolean | undefined
1902 readonly revalidateOnFocus?: boolean | "always" | undefined
1903 readonly focusSignal?: Atom<any> | undefined
1904 }
1905 ): WithoutSerializable<R>
1906} = dual(
1907 2,
1908 <A, E>(
1909 self: Atom<AsyncResult.AsyncResult<A, E>>,
1910 options: {
1911 readonly staleTime: Duration.Input
1912 readonly revalidateOnMount?: boolean | undefined
1913 readonly revalidateOnFocus?: boolean | "always" | undefined
1914 readonly focusSignal?: Atom<any> | undefined
1915 }
1916 ): Atom<AsyncResult.AsyncResult<A, E>> => {
1917 const staleTime = Duration.toMillis(Duration.fromInputUnsafe(options.staleTime))
1918 return transform(self, (get) => {
1919 const current = get.once(self)
1920 get.subscribe(self, (value) => {
1921 get.setSelf(value)
1922 })
1923 if (options.revalidateOnFocus && options.focusSignal) {
1924 get.once(options.focusSignal)
1925 get.subscribe(
1926 options.focusSignal,
1927 options.revalidateOnFocus === "always" ? () => get.refresh(self) : () => {
1928 const current = get.once(self)
1929 if (shouldRevalidateSWR(current, staleTime)) {
1930 get.refresh(self)
1931 }
1932 }
1933 )
1934 }
1935 const firstRead = Option.isNone(get.self<AsyncResult.AsyncResult<A, E>>())
1936 if (firstRead && options.revalidateOnMount === false) {
1937 return current
1938 }
1939 if (shouldRevalidateSWR(current, staleTime)) {
1940 let active = true
1941 get.addFinalizer(() => {
1942 active = false
1943 })
1944 get.registry.schedulerAsync.makeDispatcher().scheduleTask(() => {
1945 if (active && shouldRevalidateSWR(get.once(self), staleTime)) {
1946 get.refresh(self)
1947 }
1948 }, 0)
1949 }
1950 return current
1951 }, { initialValueTarget: self })
1952 }
1953) as any
1954
1955const swrTimestamp = <A, E>(result: AsyncResult.AsyncResult<A, E>): Option.Option<number> => {
1956 if (result._tag === "Success") {
1957 return Option.some(result.timestamp)
1958 }
1959 if (result._tag === "Failure") {
1960 return Option.map(result.previousSuccess, (success) => success.timestamp)
1961 }
1962 return Option.none()
1963}
1964
1965const isFreshWithin = (timestamp: number, staleTime: number, now: number): boolean => now - timestamp < staleTime
1966
1967const shouldRevalidateSWR = <A, E>(result: AsyncResult.AsyncResult<A, E>, staleTime: number): boolean => {
1968 if (result.waiting) {
1969 return false
1970 }
1971 const timestamp = Option.getOrUndefined(swrTimestamp(result))
1972 if (timestamp === undefined) {
1973 return result._tag !== "Initial"
1974 }
1975 return !isFreshWithin(timestamp, staleTime, Date.now())
1976}
1977
1978/**
1979 * Wraps an atom in a writable optimistic atom.
1980 *
1981 * **Details**
1982 *
1983 * Writes accept transition atoms containing `AsyncResult` values. Waiting
1984 * successes are shown optimistically while transitions run; when successful
1985 * transitions finish, the source atom is refreshed, and failures roll the value
1986 * back to the latest source value.
1987 *
1988 * @stability unstable
1989 * @category constructors
1990 * @since 4.0.0
1991 */
1992export const optimistic = <A>(self: Atom<A>): Writable<A, Atom<AsyncResult.AsyncResult<A, unknown>>> => {
1993 let counter = 0
1994 const writeAtom = removeTtl(state(
1995 [
1996 counter,
1997 undefined as any as Atom<AsyncResult.AsyncResult<A, unknown>>
1998 ] as const
1999 ))
2000 return writable(
2001 (get) => {
2002 let lastValue = get.once(self)
2003 let needsRefresh = false
2004 get.subscribe(self, (value) => {
2005 lastValue = value
2006 if (transitions.size > 0) {
2007 return
2008 }
2009 needsRefresh = false
2010 if (!AsyncResult.isAsyncResult(value)) {
2011 return get.setSelf(value)
2012 }
2013 const current = Option.getOrUndefined(get.self<AsyncResult.AsyncResult<any, any>>())!
2014 switch (value._tag) {
2015 case "Initial": {
2016 if (AsyncResult.isInitial(current)) {
2017 get.setSelf(value)
2018 }
2019 return
2020 }
2021 case "Success": {
2022 if (AsyncResult.isSuccess(current)) {
2023 if (!value.waiting && value.timestamp >= current.timestamp) {
2024 get.setSelf(value)
2025 }
2026 } else {
2027 get.setSelf(value)
2028 }
2029 return
2030 }
2031 case "Failure": {
2032 return get.setSelf(value)
2033 }
2034 }
2035 })
2036 const transitions = new Set<Atom<AsyncResult.AsyncResult<A, unknown>>>()
2037 const cancels = new Set<() => void>()
2038 get.subscribe(writeAtom, ([, atom]) => {
2039 if (transitions.has(atom)) return
2040 transitions.add(atom)
2041 let cancel: (() => void) | undefined
2042 // eslint-disable-next-line prefer-const
2043 cancel = get.registry.subscribe(atom, (result) => {
2044 if (AsyncResult.isSuccess(result) && result.waiting) {
2045 return get.setSelf(result.value)
2046 }
2047 transitions.delete(atom)
2048 if (cancel) {
2049 cancels.delete(cancel)
2050 cancel()
2051 }
2052 if (!needsRefresh && !AsyncResult.isFailure(result)) {
2053 needsRefresh = true
2054 }
2055 if (transitions.size === 0) {
2056 if (needsRefresh) {
2057 needsRefresh = false
2058 get.refresh(self)
2059 } else {
2060 get.setSelf(lastValue)
2061 }
2062 }
2063 }, { immediate: true })
2064 if (transitions.has(atom)) {
2065 cancels.add(cancel)
2066 } else {
2067 cancel()
2068 }
2069 })
2070 get.addFinalizer(() => {
2071 for (const cancel of cancels) cancel()
2072 transitions.clear()
2073 cancels.clear()
2074 })
2075 return lastValue
2076 },
2077 (ctx, atom) => ctx.set(writeAtom, [++counter, atom]),
2078 (refresh) => refresh(self)
2079 )
2080}
2081
2082/**
2083 * Creates an `AtomResultFn` that applies an optimistic update before running the
2084 * underlying mutation.
2085 *
2086 * **Details**
2087 *
2088 * The reducer computes the provisional value from the current value and mutation
2089 * input. The wrapped function result then completes the transition or updates the
2090 * optimistic value through the provided setter callback.
2091 *
2092 * @stability unstable
2093 * @category combinators
2094 * @since 4.0.0
2095 */
2096export const optimisticFn: {
2097 <A, W, XA, XE, OW = void>(
2098 options: {
2099 readonly reducer: (current: NoInfer<A>, update: OW) => NoInfer<W>
2100 readonly fn:
2101 | AtomResultFn<OW, XA, XE>
2102 | ((set: (result: NoInfer<W>) => void) => AtomResultFn<OW, XA, XE>)
2103 }
2104 ): (
2105 self: Writable<A, Atom<AsyncResult.AsyncResult<W, unknown>>>
2106 ) => AtomResultFn<OW, XA, XE>
2107 <A, W, XA, XE, OW = void>(
2108 self: Writable<A, Atom<AsyncResult.AsyncResult<W, unknown>>>,
2109 options: {
2110 readonly reducer: (current: NoInfer<A>, update: OW) => NoInfer<W>
2111 readonly fn:
2112 | AtomResultFn<OW, XA, XE>
2113 | ((set: (result: NoInfer<W>) => void) => AtomResultFn<OW, XA, XE>)
2114 }
2115 ): AtomResultFn<OW, XA, XE>
2116} = dual(2, <A, W, XA, XE, OW = void>(
2117 self: Writable<A, Atom<AsyncResult.AsyncResult<W, unknown>>>,
2118 options: {
2119 readonly reducer: (current: NoInfer<A>, update: OW) => NoInfer<W>
2120 readonly fn:
2121 | AtomResultFn<OW, XA, XE>
2122 | ((set: (result: NoInfer<W>) => void) => AtomResultFn<OW, XA, XE>)
2123 }
2124): AtomResultFn<OW, XA, XE> => {
2125 const transition = removeTtl(state<AsyncResult.AsyncResult<W, unknown>>(AsyncResult.initial()))
2126 return fn((arg: OW, get) => {
2127 let value = options.reducer(get(self), arg)
2128 if (AsyncResult.isAsyncResult(value)) {
2129 value = AsyncResult.waiting(value, { touch: true })
2130 }
2131 get.set(transition, AsyncResult.success(value, { waiting: true }))
2132 get.set(self, transition)
2133 const fn = typeof options.fn === "function"
2134 ? autoDispose(options.fn((value) =>
2135 get.set(
2136 transition,
2137 AsyncResult.success(AsyncResult.isAsyncResult(value) ? AsyncResult.waiting(value) : value, { waiting: true })
2138 )
2139 ))
2140 : options.fn
2141 get.set(fn, arg)
2142 return Effect.callback<XA, XE>((resume) => {
2143 get.subscribe(fn, (result) => {
2144 if (result._tag === "Initial" || result.waiting) return
2145 get.set(transition, AsyncResult.map(result, () => value))
2146 resume(AsyncResult.toExit(result) as any)
2147 }, { immediate: true })
2148 })
2149 })
2150})
2151
2152/**
2153 * Runs synchronous atom updates as a batch.
2154 *
2155 * **Details**
2156 *
2157 * Stale nodes are rebuilt and listeners are notified after the callback completes,
2158 * so dependent updates observe the final batched state.
2159 *
2160 * @stability unstable
2161 * @category batching
2162 * @since 4.0.0
2163 */
2164export const batch: (f: () => void) => void = Registry.batch
2165
2166// -----------------------------------------------------------------------------
2167// Focus
2168// -----------------------------------------------------------------------------
2169
2170/**
2171 * Creates a browser-only signal atom that increments when the document becomes visible.
2172 *
2173 * **Details**
2174 *
2175 * It listens for `visibilitychange` events on `window` and removes the listener
2176 * when the atom is disposed.
2177 *
2178 * @stability unstable
2179 * @category constants
2180 * @since 4.0.0
2181 */
2182export const windowFocusSignal: Atom<number> = readable((get) => {
2183 let count = 0
2184 function update() {
2185 if (document.visibilityState === "visible") {
2186 get.setSelf(++count)
2187 }
2188 }
2189 window.addEventListener("visibilitychange", update)
2190 get.addFinalizer(() => {
2191 window.removeEventListener("visibilitychange", update)
2192 })
2193 return count
2194})
2195
2196/**
2197 * Creates a combinator that refreshes an atom whenever the supplied signal atom
2198 * changes.
2199 *
2200 * **Details**
2201 *
2202 * The derived atom also subscribes to the source atom so normal source updates are
2203 * forwarded to its own value.
2204 *
2205 * @stability unstable
2206 * @category constructors
2207 * @since 4.0.0
2208 */
2209export const makeRefreshOnSignal = <_>(signal: Atom<_>) => <A extends Atom<any>>(self: A): WithoutSerializable<A> =>
2210 transform(self, (get) => {
2211 get.once(signal)
2212 get.subscribe(signal, (_) => get.refresh(self))
2213 get.subscribe(self, (value) => get.setSelf(value))
2214 return get.once(self)
2215 }, { initialValueTarget: self }) as any
2216
2217/**
2218 * Refreshes an atom whenever `windowFocusSignal` changes.
2219 *
2220 * **Details**
2221 *
2222 * This helper is browser-only because `windowFocusSignal` depends on `window` and
2223 * `document.visibilityState`.
2224 *
2225 * @stability unstable
2226 * @category combinators
2227 * @since 4.0.0
2228 */
2229export const refreshOnWindowFocus: <A extends Atom<any>>(self: A) => WithoutSerializable<A> = makeRefreshOnSignal(
2230 windowFocusSignal
2231)
2232
2233// -----------------------------------------------------------------------------
2234// KeyValueStore
2235// -----------------------------------------------------------------------------
2236
2237/**
2238 * Creates a writable atom backed by a `KeyValueStore` entry.
2239 *
2240 * **Details**
2241 *
2242 * Values are encoded and decoded with the supplied schema. In sync mode the atom
2243 * exposes the decoded value and writes the default value when the key is missing;
2244 * in async mode it exposes an `AsyncResult` of the decoded value.
2245 *
2246 * @stability unstable
2247 * @category constructors
2248 * @since 4.0.0
2249 */
2250export const kvs = <S extends Schema.ConstraintCodec<any, any>, const Mode extends "sync" | "async" = never>(options: {
2251 readonly runtime: AtomRuntime<KeyValueStore.KeyValueStore, any>
2252 readonly key: string
2253 readonly schema: S
2254 readonly defaultValue: LazyArg<S["Type"]>
2255 readonly mode?: Mode | undefined
2256}): Writable<"async" extends Mode ? AsyncResult.AsyncResult<S["Type"]> : S["Type"], S["Type"]> => {
2257 const setAtom = options.runtime.fn(
2258 (value: S["Type"]) =>
2259 KeyValueStore.KeyValueStore.use((store) =>
2260 KeyValueStore.toSchemaStore(store, options.schema).set(options.key, value)
2261 )
2262 )
2263 const resultAtom = options.runtime.atom(
2264 KeyValueStore.KeyValueStore.use((store) => KeyValueStore.toSchemaStore(store, options.schema).get(options.key))
2265 )
2266 return writable(
2267 options.mode === "async" ?
2268 (get) => {
2269 get.mount(setAtom)
2270 const mapper = AsyncResult.map<Option.Option<S["Type"]>, S["Type"]>(
2271 Option.getOrElse(() => {
2272 const value = options.defaultValue()
2273 get.set(setAtom, value)
2274 return value
2275 })
2276 )
2277 get.subscribe(resultAtom, (result) => get.setSelf(mapper(result)))
2278 return mapper(get.once(resultAtom))
2279 } :
2280 (get) => {
2281 get.mount(setAtom)
2282 get.subscribe(resultAtom, (result) => {
2283 if (!AsyncResult.isSuccess(result)) return
2284 if (Option.isSome(result.value)) {
2285 get.setSelf(result.value.value)
2286 } else {
2287 const value = Option.getOrElse(get.self<S["Type"]>(), options.defaultValue)
2288 get.setSelf(value)
2289 get.set(setAtom, value)
2290 }
2291 }, { immediate: true })
2292 return Option.getOrElse(get.self<S["Type"]>(), options.defaultValue)
2293 },
2294 (ctx, value: S["Type"]) => {
2295 ctx.set(setAtom, value as any)
2296 ctx.setSelf(options.mode === "async" ? AsyncResult.success(value) : value)
2297 }
2298 ) as any
2299}
2300
2301// -----------------------------------------------------------------------------
2302// URL search params
2303// -----------------------------------------------------------------------------
2304
2305/**
2306 * Creates an atom that reads and writes a URL search parameter.
2307 *
2308 * **Gotchas**
2309 *
2310 * If you pass a schema, it has to be synchronous and have no context.
2311 *
2312 * @stability unstable
2313 * @category constructors
2314 * @since 4.0.0
2315 */
2316export const searchParam = <S extends Schema.ConstraintCodec<any, string> = never>(
2317 name: string,
2318 options?: {
2319 readonly schema?: S | undefined
2320 }
2321): Writable<[S] extends [never] ? string : Option.Option<S["Type"]>> => {
2322 const decode = options?.schema && Schema.decodeExit(options.schema)
2323 const encode = options?.schema && Schema.encodeExit(options.schema)
2324 return writable(
2325 (get) => {
2326 if (typeof window === "undefined") {
2327 return decode ? Option.none() : ""
2328 }
2329 const handleUpdate = () => {
2330 if (searchParamState.updating) return
2331 const searchParams = new URLSearchParams(window.location.search)
2332 const newValue = searchParams.get(name) || ""
2333 if (decode) {
2334 get.setSelf(Exit.getSuccess(decode(newValue)))
2335 } else if (newValue !== Option.getOrUndefined(get.self())) {
2336 get.setSelf(newValue)
2337 }
2338 }
2339 window.addEventListener("popstate", handleUpdate)
2340 window.addEventListener("pushstate", handleUpdate)
2341 get.addFinalizer(() => {
2342 window.removeEventListener("popstate", handleUpdate)
2343 window.removeEventListener("pushstate", handleUpdate)
2344 })
2345 const value = new URLSearchParams(window.location.search).get(name) || ""
2346 return decode ? Exit.getSuccess(decode(value)) : value as any
2347 },
2348 (ctx, value: any) => {
2349 if (typeof window === "undefined") {
2350 ctx.setSelf(value)
2351 return
2352 }
2353
2354 if (encode) {
2355 const encoded = Option.flatMap(value, (v) => Exit.getSuccess(encode(v as S["Type"])))
2356 searchParamState.updates.set(name, Option.getOrElse(encoded, () => ""))
2357 value = Option.zipRight(encoded, value)
2358 } else {
2359 searchParamState.updates.set(name, value)
2360 }
2361 ctx.setSelf(value)
2362 if (searchParamState.timeout) {
2363 clearTimeout(searchParamState.timeout)
2364 }
2365 searchParamState.timeout = setTimeout(updateSearchParams, 500) as any
2366 }
2367 )
2368}
2369
2370const searchParamState = {
2371 timeout: undefined as number | undefined,
2372 updates: new Map<string, string>(),
2373 updating: false
2374}
2375
2376function updateSearchParams() {
2377 searchParamState.timeout = undefined
2378 searchParamState.updating = true
2379 const searchParams = new URLSearchParams(window.location.search)
2380 for (const [key, value] of searchParamState.updates.entries()) {
2381 if (value.length > 0) {
2382 searchParams.set(key, value)
2383 } else {
2384 searchParams.delete(key)
2385 }
2386 }
2387 searchParamState.updates.clear()
2388 const newUrl = `${window.location.pathname}?${searchParams.toString()}`
2389 window.history.pushState({}, "", newUrl)
2390 searchParamState.updating = false
2391}
2392
2393// -----------------------------------------------------------------------------
2394// conversions
2395// -----------------------------------------------------------------------------
2396
2397/**
2398 * Converts an atom into a stream using the `AtomRegistry` service.
2399 *
2400 * **Details**
2401 *
2402 * The stream emits the atom's current value immediately and then emits subsequent
2403 * changes until the stream scope is closed.
2404 *
2405 * @stability unstable
2406 * @category converting
2407 * @since 4.0.0
2408 */
2409export const toStream = <A>(self: Atom<A>): Stream.Stream<A, never, AtomRegistry> =>
2410 Stream.unwrap(AtomRegistry.use((r) => Effect.succeed(Registry.toStream(r, self))))
2411
2412/**
2413 * Converts an `AsyncResult` atom into a stream using the `AtomRegistry` service.
2414 *
2415 * **Details**
2416 *
2417 * Initial results are skipped, successes are emitted as stream values, and
2418 * failures fail the stream with the result cause.
2419 *
2420 * @stability unstable
2421 * @category converting
2422 * @since 4.0.0
2423 */
2424export const toStreamResult = <A, E>(self: Atom<AsyncResult.AsyncResult<A, E>>): Stream.Stream<A, E, AtomRegistry> =>
2425 Stream.unwrap(AtomRegistry.use((r) => Effect.succeed(Registry.toStreamResult(r, self))))
2426
2427/**
2428 * Reads an atom's current value from the `AtomRegistry` service.
2429 *
2430 * @stability unstable
2431 * @category converting
2432 * @since 4.0.0
2433 */
2434export const get = <A>(self: Atom<A>): Effect.Effect<A, never, AtomRegistry> =>
2435 AtomRegistry.use((r) => Effect.succeed(r.get(self)))
2436
2437/**
2438 * Reads a writable atom, computes a return value and next write value, writes the
2439 * next value, and returns the computed result.
2440 *
2441 * @stability unstable
2442 * @category converting
2443 * @since 4.0.0
2444 */
2445export const modify: {
2446 <R, W, A>(
2447 f: (_: R) => [returnValue: A, nextValue: W]
2448 ): (self: Writable<R, W>) => Effect.Effect<A, never, AtomRegistry>
2449 <R, W, A>(self: Writable<R, W>, f: (_: R) => [returnValue: A, nextValue: W]): Effect.Effect<A, never, AtomRegistry>
2450} = dual(
2451 2,
2452 <R, W, A>(self: Writable<R, W>, f: (_: R) => [returnValue: A, nextValue: W]): Effect.Effect<A, never, AtomRegistry> =>
2453 Effect.map(AtomRegistry, (_) => _.modify(self, f))
2454)
2455
2456/**
2457 * Writes a value to a writable atom through the `AtomRegistry` service.
2458 *
2459 * @stability unstable
2460 * @category converting
2461 * @since 4.0.0
2462 */
2463export const set: {
2464 <W>(value: W): <R>(self: Writable<R, W>) => Effect.Effect<void, never, AtomRegistry>
2465 <R, W>(self: Writable<R, W>, value: W): Effect.Effect<void, never, AtomRegistry>
2466} = dual(
2467 2,
2468 <R, W>(self: Writable<R, W>, value: W): Effect.Effect<void, never, AtomRegistry> =>
2469 Effect.map(AtomRegistry, (_) => _.set(self, value))
2470)
2471
2472/**
2473 * Updates a writable atom by reading its current value from the registry and
2474 * writing the value returned by the update function.
2475 *
2476 * @stability unstable
2477 * @category converting
2478 * @since 4.0.0
2479 */
2480export const update: {
2481 <R, W>(f: (_: R) => W): (self: Writable<R, W>) => Effect.Effect<void, never, AtomRegistry>
2482 <R, W>(self: Writable<R, W>, f: (_: R) => W): Effect.Effect<void, never, AtomRegistry>
2483} = dual(
2484 2,
2485 <R, W>(self: Writable<R, W>, f: (_: R) => W): Effect.Effect<void, never, AtomRegistry> =>
2486 Effect.map(AtomRegistry, (_) => _.update(self, f))
2487)
2488
2489/**
2490 * Reads an `AsyncResult` atom as an effect through the `AtomRegistry` service.
2491 *
2492 * **Details**
2493 *
2494 * The effect waits while the result is `Initial`, and also while it is waiting
2495 * when `suspendOnWaiting` is enabled. Successes succeed with the value and
2496 * failures fail with the result cause.
2497 *
2498 * @stability unstable
2499 * @category converting
2500 * @since 4.0.0
2501 */
2502export const getResult = <A, E>(
2503 self: Atom<AsyncResult.AsyncResult<A, E>>,
2504 options?: { readonly suspendOnWaiting?: boolean | undefined }
2505): Effect.Effect<A, E, AtomRegistry> => AtomRegistry.use(Registry.getResult(self, options))
2506
2507/**
2508 * Runs a refresh request for an atom through the `AtomRegistry` service.
2509 *
2510 * **When to use**
2511 *
2512 * Use to invalidate and recompute an atom from an Effect that has access to the
2513 * active registry.
2514 *
2515 * @stability unstable
2516 * @category converting
2517 * @since 4.0.0
2518 */
2519export const refresh = <A>(self: Atom<A>): Effect.Effect<void, never, AtomRegistry> =>
2520 Effect.map(AtomRegistry, (_) => _.refresh(self))
2521
2522/**
2523 * Mounts an atom in the `AtomRegistry` for the lifetime of the current scope.
2524 *
2525 * **Details**
2526 *
2527 * Mounting keeps the atom subscribed with a no-op listener until the scope
2528 * finalizer releases it.
2529 *
2530 * @stability unstable
2531 * @category converting
2532 * @since 4.0.0
2533 */
2534export const mount = <A>(self: Atom<A>): Effect.Effect<void, never, AtomRegistry | Scope.Scope> =>
2535 AtomRegistry.use((r) => Registry.mount(r, self))
2536
2537// -----------------------------------------------------------------------------
2538// Serializable
2539// -----------------------------------------------------------------------------
2540
2541/**
2542 * The type id used to mark atoms that carry serialization metadata.
2543 *
2544 * @stability unstable
2545 * @category type IDs
2546 * @since 4.0.0
2547 */
2548export const SerializableTypeId: SerializableTypeId = "~effect-atom/atom/Atom/Serializable"
2549
2550/**
2551 * The literal type of the serializable atom marker.
2552 *
2553 * @stability unstable
2554 * @category type IDs
2555 * @since 4.0.0
2556 */
2557export type SerializableTypeId = "~effect-atom/atom/Atom/Serializable"
2558
2559/**
2560 * Serialization metadata attached to an atom.
2561 *
2562 * **Details**
2563 *
2564 * The key identifies the atom in dehydrated state, and the encode/decode
2565 * functions convert between the atom value and the schema encoded value.
2566 *
2567 * @stability unstable
2568 * @category models
2569 * @since 4.0.0
2570 */
2571export interface Serializable<S extends Schema.Constraint> {
2572 readonly [SerializableTypeId]: {
2573 readonly key: string
2574 readonly encode: (value: S["Type"]) => S["Encoded"]
2575 readonly decode: (value: S["Encoded"]) => S["Type"]
2576 }
2577}
2578
2579/**
2580 * Returns `true` when an atom carries `Serializable` metadata.
2581 *
2582 * @stability unstable
2583 * @category guards
2584 * @since 4.0.0
2585 */
2586export const isSerializable = (self: Atom<any>): self is Atom<any> & Serializable<any> => SerializableTypeId in self
2587
2588/**
2589 * Attaches serialization metadata to an atom using a schema and stable key.
2590 *
2591 * **Details**
2592 *
2593 * The schema is converted to a JSON codec for synchronous encode/decode, and the
2594 * key is also used as the atom label when the atom does not already have one.
2595 *
2596 * @stability unstable
2597 * @category combinators
2598 * @since 4.0.0
2599 */
2600export const serializable: {
2601 <R extends Atom<any>, S extends Schema.ConstraintCodec<Type<R>, any>>(options: {
2602 readonly key: string
2603 readonly schema: S
2604 }): (self: R) => R & Serializable<S>
2605 <R extends Atom<any>, S extends Schema.ConstraintCodec<Type<R>, any>>(self: R, options: {
2606 readonly key: string
2607 readonly schema: S
2608 }): R & Serializable<S>
2609} = dual(2, <R extends Atom<any>, A, I>(self: R, options: {
2610 readonly key: string
2611 readonly schema: Schema.ConstraintCodec<A, I>
2612}): R & Serializable<any> => {
2613 const codecJson = Schema.toCodecJson(options.schema)
2614 return Object.assign(Object.create(Object.getPrototypeOf(self)), {
2615 ...self,
2616 label: self.label ?? [options.key, getStackTraceLimit() === 0 ? "" : new Error().stack?.split("\n")[5] ?? ""],
2617 [SerializableTypeId]: {
2618 key: options.key,
2619 encode: Schema.encodeSync(codecJson),
2620 decode: Schema.decodeSync(codecJson)
2621 }
2622 })
2623})
2624
2625/**
2626 * The type id used to mark atoms with a server-side read override.
2627 *
2628 * @stability unstable
2629 * @category type IDs
2630 * @since 4.0.0
2631 */
2632export const ServerValueTypeId = "~effect-atom/atom/Atom/ServerValue" as const
2633
2634/**
2635 * Sets the value of an Atom when read on the server.
2636 *
2637 * @stability unstable
2638 * @category transforming
2639 * @since 4.0.0
2640 */
2641export const withServerValue: {
2642 <A extends Atom<any>>(read: (get: <A>(atom: Atom<A>) => A) => Type<A>): (self: A) => A
2643 <A extends Atom<any>>(self: A, read: (get: <A>(atom: Atom<A>) => A) => Type<A>): A
2644} = dual(
2645 2,
2646 <A extends Atom<any>>(self: A, read: (get: <A>(atom: Atom<A>) => A) => Type<A>): A =>
2647 Object.assign(Object.create(Object.getPrototypeOf(self)), {
2648 ...self,
2649 [ServerValueTypeId]: read
2650 })
2651)
2652
2653/**
2654 * Sets an `AsyncResult` atom's server-side value to
2655 * `AsyncResult.initial(true)`.
2656 *
2657 * @stability unstable
2658 * @category transforming
2659 * @since 4.0.0
2660 */
2661export const withServerValueInitial = <A extends Atom<AsyncResult.AsyncResult<any, any>>>(self: A): A =>
2662 withServerValue(self, constant(AsyncResult.initial(true)) as any)
2663
2664/**
2665 * Reads an atom from a registry, using its server-side read override when one is
2666 * present.
2667 *
2668 * **Details**
2669 *
2670 * Nested reads performed by the override are resolved against the same registry.
2671 *
2672 * @stability unstable
2673 * @category getters
2674 * @since 4.0.0
2675 */
2676export const getServerValue: {
2677 (registry: Registry.AtomRegistry): <A>(self: Atom<A>) => A
2678 <A>(self: Atom<A>, registry: Registry.AtomRegistry): A
2679} = dual(
2680 2,
2681 <A>(self: Atom<A>, registry: Registry.AtomRegistry): A =>
2682 ServerValueTypeId in self
2683 ? (self as any)[ServerValueTypeId]((atom: Atom<any>) => registry.get(atom))
2684 : registry.get(self)
2685)