1import type { NonEmptyArray } from "../Array.ts"
2import * as Context from "../Context.ts"
3import type { Effect } from "../Effect.ts"
4import type { Exit } from "../Exit.ts"
5import type { Fiber } from "../Fiber.ts"
6import { dual } from "../Function.ts"
7import type * as Request from "../Request.ts"
8import { makeEntry } from "../Request.ts"
9import type { RequestResolver } from "../RequestResolver.ts"
10import { Scheduler } from "../Scheduler.ts"
11import { exitDie, isEffect } from "./core.ts"
12import * as effect from "./effect.ts"
13
14/** @internal */
15export const request: {
16 <A extends Request.Any, EX = never, RX = never>(
17 resolver: RequestResolver<A> | Effect<RequestResolver<A>, EX, RX>
18 ): (self: A) => Effect<
19 Request.Success<A>,
20 Request.Error<A> | EX,
21 Request.Services<A> | RX
22 >
23 <A extends Request.Any, EX = never, RX = never>(
24 self: A,
25 resolver: RequestResolver<A> | Effect<RequestResolver<A>, EX, RX>
26 ): Effect<
27 Request.Success<A>,
28 Request.Error<A> | EX,
29 Request.Services<A> | RX
30 >
31} = dual(
32 2,
33 <A extends Request.Any, EX = never, RX = never>(
34 self: A,
35 resolver: RequestResolver<A> | Effect<RequestResolver<A>, EX, RX>
36 ): Effect<
37 Request.Success<A>,
38 Request.Error<A> | EX,
39 Request.Services<A> | RX
40 > => {
41 const withResolver = (resolver: RequestResolver<A>) =>
42 effect.callback<
43 Request.Success<A>,
44 Request.Error<A>,
45 Request.Services<A>
46 >((resume) => {
47 const entry = addEntry(resolver, self, resume, effect.getCurrentFiber()!)
48 return maybeRemoveEntry(resolver, entry)
49 })
50 return isEffect(resolver) ? effect.flatMap(resolver, withResolver) : withResolver(resolver)
51 }
52)
53
54/** @internal */
55export const requestUnsafe = <A extends Request.Any>(
56 self: A,
57 options: {
58 readonly resolver: RequestResolver<A>
59 readonly onExit: (exit: Exit<Request.Success<A>, Request.Error<A>>) => void
60 readonly context: Context.Context<never>
61 }
62): () => void => {
63 const entry = addEntry(options.resolver, self, options.onExit, {
64 context: options.context,
65 cache: { scheduler: Context.get(options.context, Scheduler) }
66 })
67 return () => removeEntryUnsafe(options.resolver, entry)
68}
69
70interface Batch {
71 key: unknown
72 resolver: RequestResolver<any>
73 map: Map<unknown, Batch>
74 readonly entrySet: Set<Request.Entry<any>>
75 readonly entries: Set<Request.Entry<any>>
76 readonly delayEffect: Effect<void>
77 readonly run: Effect<void, unknown>
78 fiber?: Fiber<void, unknown> | undefined
79}
80
81const batchPool: Array<Batch> = []
82const pendingBatches = new WeakMap<RequestResolver<any>, Map<unknown, Batch>>()
83
84const addEntry = <A extends Request.Any>(
85 resolver: RequestResolver<A>,
86 request: A,
87 resume: (exit: Exit<any, any>) => void,
88 fiber: {
89 readonly context: Context.Context<never>
90 readonly cache: { readonly scheduler: Scheduler }
91 readonly id?: number
92 }
93) => {
94 let batchMap = pendingBatches.get(resolver)
95 if (!batchMap) {
96 batchMap = new Map<object, Batch>()
97 pendingBatches.set(resolver, batchMap)
98 }
99 let batch: Batch | undefined
100 let completed = false
101 const entry = makeEntry({
102 request,
103 context: fiber.context as any,
104 uninterruptible: false,
105 completeUnsafe(effect) {
106 if (completed) return
107 completed = true
108 // Removed entries still notify resolver hooks, but not their cancelled callers.
109 if (batch && !batch.entrySet.delete(entry)) return
110 resume(effect)
111 }
112 })
113 if (resolver.preCheck !== undefined && !resolver.preCheck(entry)) {
114 return entry
115 }
116 const key = resolver.batchKey(entry)
117 batch = batchMap.get(key)
118 if (!batch) {
119 if (batchPool.length > 0) {
120 batch = batchPool.pop()!
121 batch.key = key
122 batch.resolver = resolver
123 batch.map = batchMap
124 } else {
125 const newBatch: Batch = {
126 key,
127 resolver,
128 map: batchMap,
129 entrySet: new Set(),
130 entries: new Set(),
131 delayEffect: effect.flatMap(
132 effect.suspend(() => newBatch.resolver.delay),
133 (_) => runBatch(newBatch)
134 ) as Effect<void>,
135 run: effect.onExit(
136 effect.suspend(() =>
137 newBatch.resolver.runAll(Array.from(newBatch.entries) as NonEmptyArray<Request.Entry<any>>, newBatch.key)
138 ),
139 (exit) => {
140 for (const entry of newBatch.entrySet) {
141 entry.completeUnsafe(
142 exit._tag === "Success"
143 ? exitDie(
144 new Error("Effect.request: RequestResolver did not complete request", { cause: entry.request })
145 )
146 : exit
147 )
148 }
149 newBatch.entries.clear()
150 if (batchPool.length < 128) {
151 newBatch.entrySet.clear()
152 newBatch.key = undefined
153 newBatch.fiber = undefined
154 newBatch.resolver = undefined as any
155 newBatch.map = undefined as any
156 batchPool.push(newBatch)
157 }
158 return effect.void
159 }
160 )
161 }
162 batch = newBatch
163 }
164 batchMap.set(key, batch)
165 batch.fiber = effect.runForkWith(fiber.context)(batch.delayEffect, { scheduler: fiber.cache.scheduler })
166 }
167
168 batch.entrySet.add(entry)
169 batch.entries.add(entry)
170 if (batch.resolver.collectWhile(batch.entries)) return entry
171
172 batch.fiber!.interruptUnsafe(fiber.id)
173 batch.fiber = effect.runForkWith(fiber.context)(runBatch(batch), { scheduler: fiber.cache.scheduler })
174 return entry
175}
176
177const removeEntryUnsafe = <A extends Request.Any>(
178 resolver: RequestResolver<A>,
179 entry: Request.Entry<A>
180) => {
181 if (entry.uninterruptible) return
182 const batchMap = pendingBatches.get(resolver)
183 if (!batchMap) return
184 const key = resolver.batchKey(entry)
185 const batch = batchMap.get(key)
186 if (!batch) return
187
188 if (!batch.entries.delete(entry)) return
189 batch.entrySet.delete(entry)
190
191 let fiber: Fiber<void, unknown> | undefined
192 if (batch.entries.size === 0) {
193 batchMap.delete(key)
194 fiber = batch.fiber
195 }
196 // Delay finalizers may enqueue new requests, so complete the removed entry first.
197 entry.completeUnsafe(effect.exitInterrupt())
198 fiber?.interruptUnsafe()
199}
200
201const maybeRemoveEntry = <A extends Request.Any>(
202 resolver: RequestResolver<A>,
203 entry: Request.Entry<A>
204) => effect.sync(() => removeEntryUnsafe(resolver, entry))
205
206function runBatch(batch: Batch) {
207 if (!batch.map.has(batch.key)) return effect.void
208 batch.map.delete(batch.key)
209 return batch.run
210}