test.ts

import * as Effect from "@effect/io/Effect";
import * as Chunk from "@effect/data/Chunk";
import * as Deferred from "@effect/io/Deferred";
import * as Option from "@effect/data/Option";
import { pipe } from "@effect/data/Function";

interface Request<E, A> extends Effect.Effect<never, E, A> {}

interface RequestEntry<A extends Request<any, any>> {
  request: A;
  completer: Deferred.Deferred<Effect.Effect.Error<A>, Effect.Effect.Success<A>>;
}

interface RequestDataSource<A extends Request<any, any>, R> {
  (requests: Chunk.Chunk<RequestEntry<A>>): Effect.Effect<R, never, void>;
}

interface Cache<K extends Request<any, any>> {
  get(key: K): Effect.Effect<never, never, Option.Option<Effect.Effect.Success<K>>>;
  put(key: K, value: Effect.Effect.Success<K>): Effect.Effect<never, never, void>;
}

function withCache<A extends Request<any, any>>(cache: Cache<A>) {
  return <R>(ds: RequestDataSource<A, R>): RequestDataSource<A, R> =>
    (requests) =>
      pipe(
        // do not run already in cache
        Effect.filterPar(requests, ({ request, completer }) =>
          pipe(
            cache.get(request),
            Effect.tap((result) =>
              Option.isSome(result) ? Deferred.succeed(completer, result.value) : Effect.unit()
            ),
            Effect.map(Option.isSome)
          )
        ),
        // call resolver
        Effect.flatMap((requests) => ds(requests)),
        // update cache
        Effect.tap(() =>
          pipe(
            Chunk.map(requests, ({ request, completer }) =>
              pipe(
                Deferred.await(completer),
                Effect.flatMap((value) => cache.put(request, value)),
                Effect.catchAllCause(() => Effect.unit())
              )
            ),
            Effect.collectAllDiscard
          )
        )
      );
}

// per request: Effect.request(GetUserNameById(1), withCache(myCache)(userDataSource))
// per data-source (userDataSource = withCache(myCache)(resolver)): Effect.request(GetUserNameById(1), userDataSource)