1/**
2 * Process-local invalidation for connecting writes to dependent reads.
3 *
4 * This module does not cache values itself. It lets callers register handlers
5 * for keys, invalidate those keys, wrap successful mutations so they invalidate
6 * keys, and expose effects as queues or streams that rerun when matching keys
7 * change. The service can also batch invalidations so handlers run after the
8 * batch completes.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import * as Context from "../Context.ts"
14import * as Effect from "../Effect.ts"
15import type * as Exit from "../Exit.ts"
16import * as Fiber from "../Fiber.ts"
17import { dual, flow } from "../Function.ts"
18import * as Hash from "../Hash.ts"
19import * as Layer from "../Layer.ts"
20import * as Queue from "../Queue.ts"
21import type { ReadonlyRecord } from "../Record.ts"
22import * as Scope from "../Scope.ts"
23import * as Stream from "../Stream.ts"
24
25/**
26 * Brand type for `Reactivity`.
27 *
28 * @stability unstable
29 * @category type IDs
30 * @since 4.0.0
31 */
32export type TypeId = "~effect/reactivity/Reactivity"
33
34/**
35 * Brand for `Reactivity` implementations.
36 *
37 * @stability unstable
38 * @category type IDs
39 * @since 4.0.0
40 */
41export const TypeId: TypeId = "~effect/reactivity/Reactivity"
42
43/**
44 * Registers handlers and reruns queries when their keys are invalidated.
45 *
46 * @stability unstable
47 * @category models
48 * @since 4.0.0
49 */
50export interface Reactivity {
51 readonly [TypeId]: TypeId
52 readonly invalidateUnsafe: (keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>) => void
53 readonly registerUnsafe: (
54 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>,
55 handler: () => void
56 ) => () => void
57 readonly invalidate: (
58 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>
59 ) => Effect.Effect<void>
60 readonly mutation: <A, E, R>(
61 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>,
62 effect: Effect.Effect<A, E, R>
63 ) => Effect.Effect<A, E, R>
64 readonly query: <A, E, R>(
65 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>,
66 effect: Effect.Effect<A, E, R>
67 ) => Effect.Effect<Queue.Dequeue<A, E>, never, R | Scope.Scope>
68 readonly stream: <A, E, R>(
69 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>,
70 effect: Effect.Effect<A, E, R>
71 ) => Stream.Stream<A, E, Exclude<R, Scope.Scope>>
72 readonly withBatch: <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
73}
74
75/**
76 * Service key for reactive invalidation.
77 *
78 * @stability unstable
79 * @category services
80 * @since 4.0.0
81 */
82export const Reactivity: Context.Service<Reactivity, Reactivity> = Context.Service("effect/reactivity/Reactivity")
83
84/**
85 * Creates an in-memory `Reactivity` service.
86 *
87 * **Details**
88 *
89 * The service tracks handlers by hashed keys and runs the registered handlers when
90 * matching keys are invalidated.
91 *
92 * @stability unstable
93 * @category constructors
94 * @since 4.0.0
95 */
96export const make = Effect.sync(() => {
97 const handlers = new Map<number | string, Set<() => void>>()
98
99 const invalidateUnsafe = (keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>): void => {
100 keysToHashes(keys, (hash) => {
101 const set = handlers.get(hash)
102 if (set === undefined) return
103 set.forEach((run) => run())
104 })
105 }
106
107 const invalidate = (
108 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>
109 ): Effect.Effect<void> =>
110 Effect.contextWith((services) => {
111 const pending = Context.getOrUndefined(services, PendingInvalidation)
112 if (pending) {
113 keysToHashes(keys, (hash) => {
114 pending.add(hash)
115 })
116 } else {
117 invalidateUnsafe(keys)
118 }
119 return Effect.void
120 })
121
122 const mutation = <A, E, R>(
123 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>,
124 effect: Effect.Effect<A, E, R>
125 ): Effect.Effect<A, E, R> => Effect.tap(effect, invalidate(keys))
126
127 const registerUnsafe = (
128 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>,
129 handler: () => void
130 ): () => void => {
131 const resolvedKeys: Array<string | number> = []
132 keysToHashes(keys, (hash) => {
133 resolvedKeys.push(hash)
134 let set = handlers.get(hash)
135 if (set === undefined) {
136 set = new Set()
137 handlers.set(hash, set)
138 }
139 set.add(handler)
140 })
141 return () => {
142 for (let i = 0; i < resolvedKeys.length; i++) {
143 const set = handlers.get(resolvedKeys[i])
144 if (set === undefined) continue
145 set.delete(handler)
146 if (set.size === 0) {
147 handlers.delete(resolvedKeys[i])
148 }
149 }
150 }
151 }
152
153 const query = <A, E, R>(
154 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>,
155 effect: Effect.Effect<A, E, R>
156 ): Effect.Effect<Queue.Dequeue<A, E>, never, R | Scope.Scope> =>
157 Effect.gen(function*() {
158 const services = yield* Effect.context<Scope.Scope | R>()
159 const scope = Context.get(services, Scope.Scope)
160 const results = yield* Queue.make<A, E>()
161 const runFork = flow(Effect.runForkWith(services), Fiber.runIn(scope))
162
163 let running = false
164 let pending = false
165 const handleExit = (exit: Exit.Exit<A, E>) => {
166 if (exit._tag === "Failure") {
167 Queue.failCauseUnsafe(results, exit.cause)
168 } else {
169 Queue.offerUnsafe(results, exit.value)
170 }
171 if (pending) {
172 pending = false
173 runFork(effect).addObserver(handleExit)
174 } else {
175 running = false
176 }
177 }
178
179 function run() {
180 if (running) {
181 pending = true
182 return
183 }
184 running = true
185 runFork(effect).addObserver(handleExit)
186 }
187
188 const cancel = registerUnsafe(keys, run)
189 yield* Scope.addFinalizer(scope, Effect.sync(cancel))
190 run()
191
192 return results as Queue.Dequeue<A, E>
193 })
194
195 const stream = <A, E, R>(
196 tables: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>,
197 effect: Effect.Effect<A, E, R>
198 ): Stream.Stream<A, E, Exclude<R, Scope.Scope>> =>
199 query(tables, effect).pipe(
200 Effect.map(Stream.fromQueue),
201 Stream.unwrap
202 )
203
204 const withBatch = <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
205 Effect.suspend(() => {
206 const pending = new Set<string | number>()
207 return effect.pipe(
208 Effect.provideService(PendingInvalidation, pending),
209 Effect.onExit((_) =>
210 Effect.sync(() => {
211 pending.forEach((hash) => {
212 const set = handlers.get(hash)
213 if (set === undefined) return
214 set.forEach((run) => run())
215 })
216 })
217 )
218 )
219 })
220
221 return Reactivity.of({
222 [TypeId]: TypeId,
223 mutation,
224 query,
225 stream,
226 invalidateUnsafe,
227 invalidate,
228 registerUnsafe,
229 withBatch
230 })
231})
232
233class PendingInvalidation extends Context.Service<PendingInvalidation, Set<string | number>>()(
234 "effect/reactivity/Reactivity/PendingInvalidation"
235) {}
236
237/**
238 * Wraps an effect so the supplied keys are invalidated after the effect succeeds.
239 *
240 * **Gotchas**
241 *
242 * If the effect fails, the keys are not invalidated.
243 *
244 * @stability unstable
245 * @category accessors
246 * @since 4.0.0
247 */
248export const mutation: {
249 (
250 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>
251 ): <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R | Reactivity>
252 <A, E, R>(
253 effect: Effect.Effect<A, E, R>,
254 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>
255 ): Effect.Effect<A, E, R | Reactivity>
256} = dual(2, <A, E, R>(
257 effect: Effect.Effect<A, E, R>,
258 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>
259): Effect.Effect<A, E, R | Reactivity> => Reactivity.use((_) => _.mutation(keys, effect)))
260
261/**
262 * Runs an effect as a query tied to the supplied invalidation keys.
263 *
264 * **Details**
265 *
266 * The returned queue receives the initial result and each later result after the
267 * keys are invalidated. The registration is removed when the current scope closes.
268 *
269 * @stability unstable
270 * @category accessors
271 * @since 4.0.0
272 */
273export const query: {
274 (
275 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>
276 ): <A, E, R>(
277 effect: Effect.Effect<A, E, R>
278 ) => Effect.Effect<Queue.Dequeue<A, E>, never, R | Scope.Scope | Reactivity>
279 <A, E, R>(
280 effect: Effect.Effect<A, E, R>,
281 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>
282 ): Effect.Effect<Queue.Dequeue<A, E>, never, R | Scope.Scope | Reactivity>
283} = dual(2, <A, E, R>(
284 effect: Effect.Effect<A, E, R>,
285 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>
286): Effect.Effect<Queue.Dequeue<A, E>, never, R | Scope.Scope | Reactivity> =>
287 Reactivity.use((r) => r.query(keys, effect)))
288
289/**
290 * Runs an effect as a stream of query results tied to the supplied invalidation
291 * keys.
292 *
293 * **Details**
294 *
295 * The effect runs initially and reruns whenever the keys are invalidated.
296 *
297 * @stability unstable
298 * @category accessors
299 * @since 4.0.0
300 */
301export const stream: {
302 (
303 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>
304 ): <A, E, R>(effect: Effect.Effect<A, E, R>) => Stream.Stream<A, E, Exclude<R, Scope.Scope> | Reactivity>
305 <A, E, R>(
306 effect: Effect.Effect<A, E, R>,
307 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>
308 ): Stream.Stream<A, E, Exclude<R, Scope.Scope> | Reactivity>
309} = dual(2, <A, E, R>(
310 effect: Effect.Effect<A, E, R>,
311 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>
312): Stream.Stream<A, E, Exclude<R, Scope.Scope> | Reactivity> =>
313 Reactivity.use((r) => r.query(keys, effect)).pipe(
314 Effect.map(Stream.fromQueue),
315 Stream.unwrap
316 ))
317
318/**
319 * Invalidates the supplied keys through the `Reactivity` service.
320 *
321 * **Details**
322 *
323 * Registered queries for matching keys are rerun immediately, or collected until
324 * the enclosing reactivity batch completes.
325 *
326 * @stability unstable
327 * @category accessors
328 * @since 4.0.0
329 */
330export const invalidate = (
331 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>
332): Effect.Effect<void, never, Reactivity> => Reactivity.use((r) => r.invalidate(keys))
333
334/**
335 * The default layer that provides an in-memory `Reactivity` service.
336 *
337 * @stability unstable
338 * @category layers
339 * @since 4.0.0
340 */
341export const layer: Layer.Layer<Reactivity> = Layer.effect(Reactivity)(make)
342
343function stringOrHash(u: unknown): string | number {
344 switch (typeof u) {
345 case "string":
346 case "number":
347 case "bigint":
348 case "boolean":
349 return String(u)
350 default:
351 return Hash.hash(u)
352 }
353}
354
355const keysToHashes = (
356 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>,
357 f: (hash: string | number) => void
358): void => {
359 if (Array.isArray(keys)) {
360 for (let i = 0; i < keys.length; i++) {
361 f(stringOrHash(keys[i]))
362 }
363 return
364 }
365 for (const key in keys) {
366 f(key)
367 const ids = (keys as ReadonlyRecord<string, ReadonlyArray<unknown>>)[key]
368 for (let i = 0; i < ids.length; i++) {
369 f(`${key}:${stringOrHash(ids[i])}`)
370 }
371 }
372}