Function foldWeightedDecomposeEffectFold

packages/effect/src/internal/sink.ts:1276–1328

(
  s: S,
  max: number,
  costFn: (s: S, input: In) => Effect.Effect<number, E, R>,
  decompose: (input: In) => Effect.Effect<Chunk.Chunk<In>, E2, R2>,
  f: (s: S, input: In) => Effect.Effect<S, E3, R3>,
  input: Chunk.Chunk<In>,
  dirty: boolean,
  cost: number,
  index: number
)

/** @internal */
const foldWeightedDecomposeEffectFold = <S, In, E, R, E2, R2, E3, R3>(
  s: S,
  max: number,
  costFn: (s: S, input: In) => Effect.Effect<number, E, R>,
  decompose: (input: In) => Effect.Effect<Chunk.Chunk<In>, E2, R2>,
  f: (s: S, input: In) => Effect.Effect<S, E3, R3>,
  input: Chunk.Chunk<In>,
  dirty: boolean,
  cost: number,
  index: number
): Effect.Effect<[S, number, boolean, Chunk.Chunk<In>], E | E2 | E3, R | R2 | R3> => {
  if (index === input.length) {
    return Effect.succeed([s, cost, dirty, Chunk.empty<In>()])
  }
  const elem = pipe(input, Chunk.unsafeGet(index))
  return pipe(
    costFn(s, elem),
    Effect.map((newCost) => cost + newCost),
    Effect.flatMap((total) => {
      if (total <= max) {
        return pipe(
          f(s, elem),
          Effect.flatMap((s) =>
            foldWeightedDecomposeEffectFold(s, max, costFn, decompose, f, input, true, total, index + 1)
          )
        )
      }
      return pipe(
        decompose(elem),
        Effect.flatMap((decomposed) => {
          if (decomposed.length <= 1 && !dirty) {
            // If `elem` cannot be decomposed, we need to cross the `max` threshold. To
            // minimize "injury", we only allow this when we haven't added anything else
            // to the aggregate (dirty = false).
            return pipe(
              f(s, elem),
              Effect.map((s) => [s, total, true, pipe(input, Chunk.drop(index + 1))])
            )
          }
          if (decomposed.length <= 1 && dirty) {
            // If the state is dirty and `elem` cannot be decomposed, we stop folding
            // and include `elem` in th leftovers.
            return Effect.succeed([s, cost, dirty, pipe(input, Chunk.drop(index))])
          }
          // `elem` got decomposed, so we will recurse with the decomposed elements pushed
          // into the chunk we're processing and see if we can aggregate further.
          const next = pipe(decomposed, Chunk.appendAll(pipe(input, Chunk.drop(index + 1))))
          return foldWeightedDecomposeEffectFold(s, max, costFn, decompose, f, next, dirty, cost, 0)
        })
      )
    })
  )
}

/** @internal */
export const flatMap = dual<