openAgent

1export * as KeyedMutex from "./keyed-mutex"
2 
3import { Effect, Semaphore } from "effect"
4 
5export interface KeyedMutex<in Key> {
6  readonly size: Effect.Effect<number>
7  readonly withLock: (key: Key) => <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
8}
9 
10/**
11 * Creates an in-memory mutex with one lock per key. Entries are removed when no
12 * holder or waiter remains.
13 *
14 *   same key      -> queue
15 *   different key -> run independently
16 *
17 * `users` counts holders and waiters so an entry is not removed while a waiter
18 * will reuse it.
19 */
20export const makeUnsafe = <Key>(): KeyedMutex<Key> => {
21  const locks = new Map<Key, { readonly semaphore: Semaphore.Semaphore; users: number }>()
22 
23  const withLock =
24    (key: Key) =>
25    <A, E, R>(effect: Effect.Effect<A, E, R>) =>
26      Effect.suspend(() => {
27        const current = locks.get(key)
28        const entry = current ?? { semaphore: Semaphore.makeUnsafe(1), users: 0 }
29        if (!current) locks.set(key, entry)
30        entry.users++
31        return entry.semaphore.withPermit(effect).pipe(
32          Effect.ensuring(
33            Effect.sync(() => {
34              entry.users--
35              if (entry.users === 0) locks.delete(key)
36            }),
37          ),
38        )
39      })
40 
41  return { size: Effect.sync(() => locks.size), withLock }
42}
43 
44/** Creates an in-memory keyed mutex inside an Effect workflow. */
45export const make = <Key>(): Effect.Effect<KeyedMutex<Key>> => Effect.sync(makeUnsafe<Key>)
46