# groupedWithin

Partitions the stream into arrays, emitting when `n` is reached or `duration`
elapses after the first element of the current group.
The size must be a positive safe integer. A group can retain up to `n`
values, so callers own the memory policy for valid sizes. Invalid sizes
fail with `Cause.IllegalArgumentError`.

Matches Effect `Stream.groupedWithin`.

## Signatures

```ts
export declare const groupedWithin: {
    (n: number, duration: Duration.Input): <A, E, R>(self: Fx<A, E, R>) => Fx<NonEmptyReadonlyArray<A>, E | Cause.IllegalArgumentError, R | Scope.Scope>;
    <A, E, R>(self: Fx<A, E, R>, n: number, duration: Duration.Input): Fx<NonEmptyReadonlyArray<A>, E | Cause.IllegalArgumentError, R | Scope.Scope>;
};
```

## Why

`groupedWithin` bounds a batch by both cardinality and time. The timer starts with the first value,
and whichever boundary wins emits the ordered non-empty group and resets both buffer and timer.
A partial group is also flushed when the source run returns, including after delivered failure.

## Ownership and lifetime

Each run retains at most `n` values and at most one timer fiber in the required Scope. A source
failure is Sink delivery, so `flushNow` still runs afterward unless the consumer interrupts the
run. Flushing or interruption cancels the timer; invalid sizes fail before subscription.

## Examples

```ts
import { Cause, Effect } from "effect"
import { Fx, Sink } from "@typed/fx"
const source = Fx.make<number, string>((sink) =>
  sink.onSuccess(1).pipe(Effect.andThen(sink.onFailure(Cause.fail("boom"))))
)
const program = Fx.groupedWithin(source, 2, "1 hour").run(
  Sink.make(Effect.logError, (group) => Effect.log(group))
) // logs the failure, then flushes [1]
```