1/**
2 * Transactional priority queues whose state is stored in a `TxRef`. Elements
3 * are kept in the order defined by the `Order` supplied at construction time,
4 * and dequeue operations return the first element according to that ordering.
5 *
6 * Use `TxPriorityQueue` when multiple fibers coordinate through a shared queue
7 * and queue operations need to compose with other transactional state changes.
8 * The retrying `peek` and `take` operations wait transactionally when the queue
9 * is empty, so they can be combined with other transactional reads and writes in
10 * one atomic workflow.
11 *
12 * @since 4.0.0
13 */
14
15import type { Chunk } from "./Chunk.ts"
16import * as C from "./Chunk.ts"
17import * as Effect from "./Effect.ts"
18import { dual } from "./Function.ts"
19import type { Inspectable } from "./Inspectable.ts"
20import { NodeInspectSymbol, toJson } from "./Inspectable.ts"
21import type { Option } from "./Option.ts"
22import * as O from "./Option.ts"
23import type { Order } from "./Order.ts"
24import type { Pipeable } from "./Pipeable.ts"
25import { pipeArguments } from "./Pipeable.ts"
26import { hasProperty, type Predicate } from "./Predicate.ts"
27import * as TxRef from "./TxRef.ts"
28
29const TypeId = "~effect/TxPriorityQueue"
30
31/**
32 * A transactional priority queue backed by a sorted `Chunk`.
33 *
34 * **Details**
35 *
36 * Elements are stored in ascending order according to the `Order` provided at
37 * construction time. `take` returns the smallest element, `peek` observes it
38 * without removing.
39 *
40 * **Example** (Dequeuing values by priority)
41 *
42 * ```ts import.meta.vitest
43 * import { Effect, Order, TxPriorityQueue } from "effect"
44 *
45 * const program = Effect.gen(function*() {
46 * const pq = yield* TxPriorityQueue.empty<number>(Order.Number)
47 * yield* TxPriorityQueue.offer(pq, 3)
48 * yield* TxPriorityQueue.offer(pq, 1)
49 * yield* TxPriorityQueue.offer(pq, 2)
50 * return yield* TxPriorityQueue.take(pq)
51 * })
52 *
53 * await Effect.runPromise(program) // => 1
54 * ```
55 *
56 * @category models
57 * @since 4.0.0
58 */
59export interface TxPriorityQueue<in out A> extends Inspectable, Pipeable {
60 readonly [TypeId]: typeof TypeId
61 readonly ref: TxRef.TxRef<Chunk<A>>
62 readonly ord: Order<A>
63}
64
65const TxPriorityQueueProto: Omit<TxPriorityQueue<unknown>, typeof TypeId | "ref" | "ord"> = {
66 [NodeInspectSymbol](this: TxPriorityQueue<unknown>) {
67 return toJson(this)
68 },
69 toJSON(this: TxPriorityQueue<unknown>) {
70 return {
71 _id: "TxPriorityQueue"
72 }
73 },
74 pipe() {
75 return pipeArguments(this, arguments)
76 }
77}
78
79const makeTxPriorityQueue = <A>(ref: TxRef.TxRef<Chunk<A>>, ord: Order<A>): TxPriorityQueue<A> => {
80 const self = Object.create(TxPriorityQueueProto)
81 self[TypeId] = TypeId
82 self.ref = ref
83 self.ord = ord
84 return self
85}
86
87/**
88 * Merges the sorted `values` into the sorted `chunk`. Existing elements stay
89 * ahead of equal incoming ones, and incoming ties keep their input order.
90 *
91 * Each insertion point is found by galloping forward from the previous one and
92 * then bisecting, so a value that lands close to the last one costs only a few
93 * comparisons.
94 */
95const mergeSorted = <A>(chunk: Chunk<A>, values: ReadonlyArray<A>, ord: Order<A>): Chunk<A> => {
96 const arr = C.toReadonlyArray(chunk)
97 const out: Array<A> = Array(arr.length + values.length)
98 let i = 0
99 let k = 0
100 for (const value of values) {
101 let lo = i
102 let hi = i
103 for (let step = 1; hi < arr.length && ord(arr[hi], value) <= 0; step *= 2) {
104 lo = hi + 1
105 hi += step
106 }
107 hi = Math.min(hi, arr.length)
108 while (lo < hi) {
109 const mid = (lo + hi) >>> 1
110 if (ord(arr[mid], value) <= 0) {
111 lo = mid + 1
112 } else {
113 hi = mid
114 }
115 }
116 for (; i < lo; i++) out[k++] = arr[i]
117 out[k++] = value
118 }
119 for (; i < arr.length; i++) out[k++] = arr[i]
120 return C.fromIterable(out)
121}
122
123/**
124 * Creates an empty `TxPriorityQueue` with the given ordering.
125 *
126 * **Example** (Creating an empty priority queue)
127 *
128 * ```ts import.meta.vitest
129 * import { Effect, Order, TxPriorityQueue } from "effect"
130 *
131 * const program = Effect.gen(function*() {
132 * const pq = yield* TxPriorityQueue.empty<number>(Order.Number)
133 * return yield* TxPriorityQueue.isEmpty(pq)
134 * })
135 *
136 * await Effect.runPromise(program) // => true
137 * ```
138 *
139 * @category constructors
140 * @since 2.0.0
141 */
142export const empty = <A>(order: Order<A>): Effect.Effect<TxPriorityQueue<A>> =>
143 Effect.map(TxRef.make<Chunk<A>>(C.empty()), (ref) => makeTxPriorityQueue(ref, order))
144
145/**
146 * Creates a `TxPriorityQueue` from an iterable of elements.
147 *
148 * **Example** (Creating a priority queue from an iterable)
149 *
150 * ```ts import.meta.vitest
151 * import { Effect, Order, TxPriorityQueue } from "effect"
152 *
153 * const program = Effect.gen(function*() {
154 * const pq = yield* TxPriorityQueue.fromIterable(Order.Number, [3, 1, 2])
155 * return yield* TxPriorityQueue.take(pq)
156 * })
157 *
158 * await Effect.runPromise(program) // => 1
159 * ```
160 *
161 * @category constructors
162 * @since 2.0.0
163 */
164export function fromIterable<A>(
165 order: Order<A>
166): (iterable: Iterable<A>) => Effect.Effect<TxPriorityQueue<A>>
167export function fromIterable<A>(
168 order: Order<A>,
169 iterable: Iterable<A>
170): Effect.Effect<TxPriorityQueue<A>>
171export function fromIterable<A>(
172 order: Order<A>,
173 iterable?: Iterable<A>
174): Effect.Effect<TxPriorityQueue<A>> | ((iterable: Iterable<A>) => Effect.Effect<TxPriorityQueue<A>>) {
175 if (iterable === undefined) {
176 return (iterable) => fromIterable(order, iterable)
177 }
178 const arr = Array.from(iterable).sort((a, b) => order(a, b))
179 return Effect.map(
180 TxRef.make<Chunk<A>>(C.fromIterable(arr)),
181 (ref) => makeTxPriorityQueue(ref, order)
182 )
183}
184
185/**
186 * Creates a `TxPriorityQueue` from variadic elements.
187 *
188 * **Example** (Creating a priority queue from variadic values)
189 *
190 * ```ts import.meta.vitest
191 * import { Effect, Order, TxPriorityQueue } from "effect"
192 *
193 * const program = Effect.gen(function*() {
194 * const pq = yield* TxPriorityQueue.make(Order.Number)(3, 1, 2)
195 * return yield* TxPriorityQueue.take(pq)
196 * })
197 *
198 * await Effect.runPromise(program) // => 1
199 * ```
200 *
201 * @category constructors
202 * @since 2.0.0
203 */
204export const make = <A>(order: Order<A>) => (...elements: Array<A>): Effect.Effect<TxPriorityQueue<A>> =>
205 fromIterable(order, elements)
206
207/**
208 * Returns the number of elements in the queue.
209 *
210 * **Example** (Getting the queue size)
211 *
212 * ```ts import.meta.vitest
213 * import { Effect, Order, TxPriorityQueue } from "effect"
214 *
215 * const program = Effect.gen(function*() {
216 * const pq = yield* TxPriorityQueue.fromIterable(Order.Number, [1, 2, 3])
217 * return yield* TxPriorityQueue.size(pq)
218 * })
219 *
220 * await Effect.runPromise(program) // => 3
221 * ```
222 *
223 * @category getters
224 * @since 2.0.0
225 */
226export const size = <A>(self: TxPriorityQueue<A>): Effect.Effect<number> => Effect.map(TxRef.get(self.ref), C.size)
227
228/**
229 * Returns `true` if the queue is empty.
230 *
231 * **Example** (Checking whether a queue is empty)
232 *
233 * ```ts import.meta.vitest
234 * import { Effect, Order, TxPriorityQueue } from "effect"
235 *
236 * const program = Effect.gen(function*() {
237 * const pq = yield* TxPriorityQueue.empty<number>(Order.Number)
238 * return yield* TxPriorityQueue.isEmpty(pq)
239 * })
240 *
241 * await Effect.runPromise(program) // => true
242 * ```
243 *
244 * @category predicates
245 * @since 2.0.0
246 */
247export const isEmpty = <A>(self: TxPriorityQueue<A>): Effect.Effect<boolean> => Effect.map(size(self), (n) => n === 0)
248
249/**
250 * Returns `true` if the queue has at least one element.
251 *
252 * **Example** (Checking whether a queue has elements)
253 *
254 * ```ts import.meta.vitest
255 * import { Effect, Order, TxPriorityQueue } from "effect"
256 *
257 * const program = Effect.gen(function*() {
258 * const pq = yield* TxPriorityQueue.fromIterable(Order.Number, [1])
259 * return yield* TxPriorityQueue.isNonEmpty(pq)
260 * })
261 *
262 * await Effect.runPromise(program) // => true
263 * ```
264 *
265 * @category predicates
266 * @since 2.0.0
267 */
268export const isNonEmpty = <A>(self: TxPriorityQueue<A>): Effect.Effect<boolean> => Effect.map(size(self), (n) => n > 0)
269
270/**
271 * Observes the smallest element without removing it.
272 *
273 * **When to use**
274 *
275 * Use to inspect the next prioritized value and retry transactionally while
276 * the queue is empty.
277 *
278 * **Example** (Peeking at the next value)
279 *
280 * ```ts import.meta.vitest
281 * import { Effect, Order, TxPriorityQueue } from "effect"
282 *
283 * const program = Effect.gen(function*() {
284 * const pq = yield* TxPriorityQueue.fromIterable(Order.Number, [3, 1, 2])
285 * return yield* TxPriorityQueue.peek(pq)
286 * })
287 *
288 * await Effect.runPromise(program) // => 1
289 * ```
290 *
291 * @category getters
292 * @since 2.0.0
293 */
294export const peek = <A>(self: TxPriorityQueue<A>): Effect.Effect<A> =>
295 Effect.gen(function*() {
296 const chunk = yield* TxRef.get(self.ref)
297 const head = C.head(chunk)
298 if (O.isNone(head)) {
299 return yield* Effect.txRetry
300 }
301 return head.value
302 }).pipe(Effect.tx)
303
304/**
305 * Observes the smallest element without removing it, returning `None` when the
306 * queue is empty.
307 *
308 * **When to use**
309 *
310 * Use to inspect the next prioritized value without retrying on an empty queue.
311 *
312 * **Example** (Peeking without retrying)
313 *
314 * ```ts import.meta.vitest
315 * import { Effect, Option, Order, TxPriorityQueue } from "effect"
316 *
317 * const program = Effect.gen(function*() {
318 * const pq = yield* TxPriorityQueue.empty<number>(Order.Number)
319 * return yield* TxPriorityQueue.peekOption(pq)
320 * })
321 *
322 * await Effect.runPromise(program) // => Option.none()
323 * ```
324 *
325 * @category getters
326 * @since 2.0.0
327 */
328export const peekOption = <A>(self: TxPriorityQueue<A>): Effect.Effect<Option<A>> =>
329 Effect.map(TxRef.get(self.ref), C.head)
330
331/**
332 * Inserts an element into the queue in sorted position.
333 *
334 * **Example** (Offering a value)
335 *
336 * ```ts import.meta.vitest
337 * import { Effect, Order, TxPriorityQueue } from "effect"
338 *
339 * const program = Effect.gen(function*() {
340 * const pq = yield* TxPriorityQueue.empty<number>(Order.Number)
341 * yield* TxPriorityQueue.offer(pq, 2)
342 * yield* TxPriorityQueue.offer(pq, 1)
343 * return yield* TxPriorityQueue.take(pq)
344 * })
345 *
346 * await Effect.runPromise(program) // => 1
347 * ```
348 *
349 * @category mutations
350 * @since 2.0.0
351 */
352export const offer: {
353 <A>(value: A): (self: TxPriorityQueue<A>) => Effect.Effect<void>
354 <A>(self: TxPriorityQueue<A>, value: A): Effect.Effect<void>
355} = dual(
356 2,
357 <A>(self: TxPriorityQueue<A>, value: A): Effect.Effect<void> =>
358 TxRef.update(self.ref, (chunk) => mergeSorted(chunk, [value], self.ord))
359)
360
361/**
362 * Inserts all elements from an iterable into the queue.
363 *
364 * **Example** (Offering multiple values)
365 *
366 * ```ts import.meta.vitest
367 * import { Effect, Order, TxPriorityQueue } from "effect"
368 *
369 * const program = Effect.gen(function*() {
370 * const pq = yield* TxPriorityQueue.empty<number>(Order.Number)
371 * yield* TxPriorityQueue.offerAll(pq, [3, 1, 2])
372 * return yield* TxPriorityQueue.take(pq)
373 * })
374 *
375 * await Effect.runPromise(program) // => 1
376 * ```
377 *
378 * @category mutations
379 * @since 2.0.0
380 */
381export const offerAll: {
382 <A>(values: Iterable<A>): (self: TxPriorityQueue<A>) => Effect.Effect<void>
383 <A>(self: TxPriorityQueue<A>, values: Iterable<A>): Effect.Effect<void>
384} = dual(
385 2,
386 <A>(self: TxPriorityQueue<A>, values: Iterable<A>): Effect.Effect<void> =>
387 TxRef.update(self.ref, (chunk) => mergeSorted(chunk, Array.from(values).sort(self.ord), self.ord))
388)
389
390/**
391 * Takes the smallest element from the queue. Retries if the queue is empty.
392 *
393 * **Example** (Taking the next value)
394 *
395 * ```ts import.meta.vitest
396 * import { Effect, Order, TxPriorityQueue } from "effect"
397 *
398 * const program = Effect.gen(function*() {
399 * const pq = yield* TxPriorityQueue.fromIterable(Order.Number, [3, 1, 2])
400 * return yield* TxPriorityQueue.take(pq)
401 * })
402 *
403 * await Effect.runPromise(program) // => 1
404 * ```
405 *
406 * @category mutations
407 * @since 2.0.0
408 */
409export const take = <A>(self: TxPriorityQueue<A>): Effect.Effect<A> =>
410 Effect.gen(function*() {
411 const chunk = yield* TxRef.get(self.ref)
412 const head = C.head(chunk)
413 if (O.isNone(head)) {
414 return yield* Effect.txRetry
415 }
416 yield* TxRef.set(self.ref, C.drop(chunk, 1))
417 return head.value
418 }).pipe(Effect.tx)
419
420/**
421 * Takes all elements from the queue, returning them in priority order.
422 *
423 * **Example** (Taking all values in priority order)
424 *
425 * ```ts import.meta.vitest
426 * import { Effect, Order, TxPriorityQueue } from "effect"
427 *
428 * const program = Effect.gen(function*() {
429 * const pq = yield* TxPriorityQueue.fromIterable(Order.Number, [3, 1, 2])
430 * return yield* TxPriorityQueue.takeAll(pq)
431 * })
432 *
433 * await Effect.runPromise(program) // => [1, 2, 3]
434 * ```
435 *
436 * @category mutations
437 * @since 2.0.0
438 */
439export const takeAll = <A>(self: TxPriorityQueue<A>): Effect.Effect<Array<A>> =>
440 Effect.map(
441 TxRef.modify(self.ref, (chunk) => [chunk, C.empty()]),
442 C.toArray
443 )
444
445/**
446 * Tries to take the smallest element. Returns `None` if the queue is empty.
447 *
448 * **Example** (Taking without retrying)
449 *
450 * ```ts import.meta.vitest
451 * import { Effect, Option, Order, TxPriorityQueue } from "effect"
452 *
453 * const program = Effect.gen(function*() {
454 * const pq = yield* TxPriorityQueue.empty<number>(Order.Number)
455 * return yield* TxPriorityQueue.takeOption(pq)
456 * })
457 *
458 * await Effect.runPromise(program) // => Option.none()
459 * ```
460 *
461 * @category mutations
462 * @since 2.0.0
463 */
464export const takeOption = <A>(self: TxPriorityQueue<A>): Effect.Effect<Option<A>> =>
465 TxRef.modify(self.ref, (chunk) => {
466 const head = C.head(chunk)
467 if (O.isNone(head)) {
468 return [O.none<A>(), chunk]
469 }
470 return [O.some(head.value), C.drop(chunk, 1)]
471 })
472
473/**
474 * Takes up to `n` elements from the queue in priority order.
475 *
476 * **Example** (Taking up to a limit)
477 *
478 * ```ts import.meta.vitest
479 * import { Effect, Order, TxPriorityQueue } from "effect"
480 *
481 * const program = Effect.gen(function*() {
482 * const pq = yield* TxPriorityQueue.fromIterable(Order.Number, [5, 3, 1, 4, 2])
483 * return yield* TxPriorityQueue.takeUpTo(pq, 2)
484 * })
485 *
486 * await Effect.runPromise(program) // => [1, 2]
487 * ```
488 *
489 * @category mutations
490 * @since 2.0.0
491 */
492export const takeUpTo: {
493 (n: number): <A>(self: TxPriorityQueue<A>) => Effect.Effect<Array<A>>
494 <A>(self: TxPriorityQueue<A>, n: number): Effect.Effect<Array<A>>
495} = dual(
496 2,
497 <A>(self: TxPriorityQueue<A>, n: number): Effect.Effect<Array<A>> =>
498 Effect.map(
499 TxRef.modify(self.ref, (chunk) => {
500 const taken = C.take(chunk, n)
501 const rest = C.drop(chunk, n)
502 return [taken, rest]
503 }),
504 C.toArray
505 )
506)
507
508/**
509 * Removes elements matching the predicate.
510 *
511 * **Example** (Removing matching values)
512 *
513 * ```ts import.meta.vitest
514 * import { Effect, Order, TxPriorityQueue } from "effect"
515 *
516 * const program = Effect.gen(function*() {
517 * const pq = yield* TxPriorityQueue.fromIterable(Order.Number, [1, 2, 3, 4, 5])
518 * yield* TxPriorityQueue.removeIf(pq, (n) => n % 2 === 0)
519 * return yield* TxPriorityQueue.takeAll(pq)
520 * })
521 *
522 * await Effect.runPromise(program) // => [1, 3, 5]
523 * ```
524 *
525 * @category filtering
526 * @since 2.0.0
527 */
528export const removeIf: {
529 <A>(predicate: Predicate<A>): (self: TxPriorityQueue<A>) => Effect.Effect<void>
530 <A>(self: TxPriorityQueue<A>, predicate: Predicate<A>): Effect.Effect<void>
531} = dual(
532 2,
533 <A>(self: TxPriorityQueue<A>, predicate: Predicate<A>): Effect.Effect<void> =>
534 TxRef.update(self.ref, (chunk) => C.filter(chunk, (a) => !predicate(a)))
535)
536
537/**
538 * Keeps only elements matching the predicate.
539 *
540 * **Example** (Retaining matching values)
541 *
542 * ```ts import.meta.vitest
543 * import { Effect, Order, TxPriorityQueue } from "effect"
544 *
545 * const program = Effect.gen(function*() {
546 * const pq = yield* TxPriorityQueue.fromIterable(Order.Number, [1, 2, 3, 4, 5])
547 * yield* TxPriorityQueue.retainIf(pq, (n) => n % 2 === 0)
548 * return yield* TxPriorityQueue.takeAll(pq)
549 * })
550 *
551 * await Effect.runPromise(program) // => [2, 4]
552 * ```
553 *
554 * @category filtering
555 * @since 2.0.0
556 */
557export const retainIf: {
558 <A>(predicate: Predicate<A>): (self: TxPriorityQueue<A>) => Effect.Effect<void>
559 <A>(self: TxPriorityQueue<A>, predicate: Predicate<A>): Effect.Effect<void>
560} = dual(
561 2,
562 <A>(self: TxPriorityQueue<A>, predicate: Predicate<A>): Effect.Effect<void> =>
563 TxRef.update(self.ref, (chunk) => C.filter(chunk, predicate))
564)
565
566/**
567 * Returns all elements in priority order without removing them.
568 *
569 * **Example** (Reading values in priority order)
570 *
571 * ```ts import.meta.vitest
572 * import { Effect, Order, TxPriorityQueue } from "effect"
573 *
574 * const program = Effect.gen(function*() {
575 * const pq = yield* TxPriorityQueue.fromIterable(Order.Number, [3, 1, 2])
576 * return yield* TxPriorityQueue.toArray(pq)
577 * })
578 *
579 * await Effect.runPromise(program) // => [1, 2, 3]
580 * ```
581 *
582 * @category converting
583 * @since 2.0.0
584 */
585export const toArray = <A>(self: TxPriorityQueue<A>): Effect.Effect<Array<A>> =>
586 Effect.map(TxRef.get(self.ref), C.toArray)
587
588/**
589 * Determines if the provided value is a `TxPriorityQueue`.
590 *
591 * **Example** (Checking for a TxPriorityQueue)
592 *
593 * ```ts import.meta.vitest
594 * import { Effect, Order, TxPriorityQueue } from "effect"
595 *
596 * const program = Effect.gen(function*() {
597 * const pq = yield* TxPriorityQueue.empty<number>(Order.Number)
598 * return [TxPriorityQueue.isTxPriorityQueue(pq), TxPriorityQueue.isTxPriorityQueue("nope")]
599 * })
600 *
601 * await Effect.runPromise(program) // => [true, false]
602 * ```
603 *
604 * @category guards
605 * @since 4.0.0
606 */
607export const isTxPriorityQueue = (u: unknown): u is TxPriorityQueue<unknown> => hasProperty(u, TypeId)