export const aggregateWithin: {
  <B, A, A2, E2, R2, C, E3, R3>(
    sink: Sink.Sink<B, A | A2, A2, E2, R2>,
    schedule: Schedule.Schedule<C, Option.Option<B>, E3, R3>
  ): <E, R>(self: Stream<A, E, R>) => Stream<B, E2 | E | E3, R2 | R3 | R>
  <A, E, R, B, A2, E2, R2, C, E3, R3>(
    self: Stream<A, E, R>,
    sink: Sink.Sink<B, A | A2, A2, E2, R2>,
    schedule: Schedule.Schedule<C, Option.Option<B>, E3, R3>
  ): Stream<B, E | E2 | E3, R | R2 | R3>
} = dual(3, <A, E, R, B, A2, E2, R2, C, E3, R3>(
  self: Stream<A, E, R>,
  sink: Sink.Sink<B, A | A2, A2, E2, R2>,
  schedule: Schedule.Schedule<C, Option.Option<B>, E3, R3>
): Stream<B, E | E2 | E3, R | R2 | R3> =>
  fromChannel(Channel.fromTransformBracket(Effect.fnUntraced(function*(_upstream, _, scope) {
    const pull = yield* Channel.toPullScoped(self.channel, _)

    const pullLatch = Latch.makeUnsafe(false)
    const scheduleStep = Symbol()
    const buffer = yield* Queue.make<Arr.NonEmptyReadonlyArray<A> | typeof scheduleStep, E | Cause.Done<void>>({
      capacity: 0
    })

    // upstream -> buffer
    yield* pull.pipe(
      pullLatch.whenOpen,
      Effect.flatMap((arr) => {
        pullLatch.closeUnsafe()
        return Queue.offer(buffer, arr)
      }),
      Effect.forever, // don't disable autoYield to prevent choking the schedule
      Effect.catchCause((cause) => Queue.failCause(buffer, cause)),
      Effect.forkIn(scope)
    )

    // schedule -> buffer
    let lastOutput = Option.none<B>()
    let leftover: Arr.NonEmptyReadonlyArray<A2> | undefined
    let sinkHasInput = false
    const step = yield* Schedule.toStepWithSleep(schedule)
    const stepToBuffer = Effect.suspend(function loop(): Pull.Pull<never, E3, void, R3> {
      return step(lastOutput).pipe(
        Effect.flatMap(() => !sinkHasInput ? loop() : Queue.offer(buffer, scheduleStep)),
        Effect.flatMap(() => Effect.never),
        Pull.catchDone(() => Cause.done())
      )
    })

    // buffer -> sink
    const pullFromBuffer: Pull.Pull<
      Arr.NonEmptyReadonlyArray<A>,
      E
    > = Queue.take(buffer).pipe(
      Effect.flatMap((arr) => {
        if (arr === scheduleStep) {
          return Cause.done()
        }
        sinkHasInput = true
        return Effect.succeed(arr)
      })
    )

    const sinkUpstream = Effect.suspend((): Pull.Pull<Arr.NonEmptyReadonlyArray<A | A2>, E> => {
      if (leftover !== undefined) {
        const chunk = leftover
        leftover = undefined
        sinkHasInput = true
        return Effect.succeed(chunk)
      }
      pullLatch.openUnsafe()
      return pullFromBuffer
    })
    const catchSinkHalt = Effect.flatMap(([value, leftover_]: Sink.End<B, A2>) => {
      // ignore the last output if the upstream only pulled a halt
      if (!sinkHasInput && buffer.state._tag === "Done") return Cause.done()
      lastOutput = Option.some(value)
      leftover = leftover_
      return Effect.succeed(Arr.of(value))
    })

    return Effect.suspend(() => {
      // if the buffer has exited and there is no more data to process
      if (buffer.state._tag === "Done" && leftover === undefined) {
        return buffer.state.exit as Exit.Exit<never, Cause.Done<void> | E>
      }
      sinkHasInput = leftover !== undefined
      return Effect.succeed(Effect.suspend(() => sink.transform(sinkUpstream as any, scope)))
    }).pipe(
      Effect.flatMap((pull) => Effect.raceFirst(catchSinkHalt(pull), stepToBuffer))
    )
  }))))