# Sink

A `Sink<A, In, L, E, R>` is the *consumer* side of streaming. Where a `Stream`
describes how values are produced, a `Sink` describes how they are consumed: it
pulls `In` elements, may fail with `E`, requires services `R`, and produces a
result `A` once it is done — optionally returning `L` *leftover* elements it did
not use. You run a stream through a sink with `Stream.run`, and every `run*`
destructor you have already seen (`runCollect`, `runFold`, …) is really a
specialised sink underneath.

```text
     ┌─── result produced by the Sink
     │  ┌─── elements consumed by the Sink
     │  │   ┌─── leftover elements
     │  │   │  ┌─── possible errors
     │  │   │  │  ┌─── required services
     ▼  ▼   ▼  ▼  ▼
  Sink<A, In, L, E, R>
```

```ts
import { Effect, Sink, Stream } from "effect"

const numbers = Stream.make(1, 2, 3, 4, 5)

// `Sink.take(2)` consumes exactly two elements and returns them as an array.
// Its type is `Sink<Array<number>, number, number>` — it produces an array,
// consumes numbers, and reports the unused elements ([3, 4, 5]) as leftovers.
const firstTwo = Sink.take<number>(2)

// `Stream.run` feeds the stream into the sink and yields the sink's result.
export const result: Effect.Effect<Array<number>> = numbers.pipe(
  Stream.run(firstTwo)
)
// => [1, 2]
```

## Built-in sinks

The `Sink` module ships the consumers you reach for most often. Each is a value
you can pass to `Stream.run`:

```ts
import { Effect, Sink, Stream } from "effect"

const numbers = Stream.make(1, 2, 3, 4, 5)

// `Sink.collect()` gathers every element into an array.
export const all = numbers.pipe(Stream.run(Sink.collect<number>()))
// => [1, 2, 3, 4, 5]

// `Sink.sum` and `Sink.count` aggregate without retaining elements.
export const total = numbers.pipe(Stream.run(Sink.sum)) // => 15
export const howMany = numbers.pipe(Stream.run(Sink.count)) // => 5

// `Sink.head()` / `Sink.last()` capture an edge element as an Option.
export const first = numbers.pipe(Stream.run(Sink.head<number>()))

// `Sink.forEach` runs an effect per element and produces void — a streaming
// equivalent of `Stream.runForEach`.
export const logged = numbers.pipe(
  Stream.run(Sink.forEach((n) => Effect.logInfo(`got ${n}`)))
)

// `Sink.drain` consumes and discards everything, producing void.
export const drained = numbers.pipe(Stream.run(Sink.drain))
```

## Folding and early termination

Like a stream, a sink can fold elements into a single accumulated value — and,
crucially, it can *stop early*. `Sink.fold` takes an initial state, a `while`
predicate, and a step function; it keeps consuming only while the predicate holds.
Elements pulled past the stopping point come back as leftovers.

```ts
import { Effect, Sink, Stream } from "effect"

const numbers = Stream.make(1, 2, 3, 4, 5, 6)

// Sum elements while the running total stays below 10. As soon as it reaches 10
// the sink completes; remaining input becomes leftover.
const sumUntilTen = Sink.fold(
  () => 0, // initial state (a thunk so it can be re-evaluated safely)
  (total) => total < 10, // keep going while this holds
  (total, n: number) => Effect.succeed(total + n) // effectful step
)

export const partialSum = numbers.pipe(Stream.run(sumUntilTen))
```

`Sink.reduce` is the simpler, non-effectful cousin for folding without a
continuation predicate, and `Sink.takeWhile` collects the longest prefix matching
a predicate. For batch-level folding over each pulled array there is
`Sink.foldArray`.

## Transforming sinks

Sinks compose. You can map their result, adapt their input, recover from
failures, and even sequence one after another.

```ts
import { Sink, Stream } from "effect"

const numbers = Stream.make(1, 2, 3, 4, 5)

// `Sink.map` transforms the *result* once the sink completes.
const average = Sink.collect<number>().pipe(
  Sink.map((arr) => arr.length === 0 ? 0 : arr.reduce((a, b) => a + b, 0) / arr.length)
)

// `Sink.mapInput` adapts the *input* side, letting a `Sink<_, number>` consume
// a stream of strings by parsing first.
const sumOfParsed = Sink.sum.pipe(
  Sink.mapInput((s: string) => Number(s))
)

// `Sink.flatMap` runs a second sink after the first, using the first result to
// decide what to do next. Here we take 2 elements, then drain the rest.
const takeThenDrain = Sink.take<number>(2).pipe(
  Sink.flatMap((firstTwo) =>
    Sink.drain.pipe(Sink.map(() => firstTwo))
  )
)

export const avg = numbers.pipe(Stream.run(average))
export const head2 = numbers.pipe(Stream.run(takeThenDrain))
```

`Sink.orElse` switches to a fallback sink if the first one fails, and
`Sink.catchCause` recovers based on the full `Cause` — the same error-handling
shape you use for [streams](https://effect.plants.sh/streaming/error-handling/) and
[effects](https://effect.plants.sh/error-management/).

## Building a custom sink

When the built-ins do not fit, `Sink.make<In>()` lets you describe a consumer as
a pipeline over a `Stream<In>`. The final step must return an `Effect`; its
success value becomes the sink's result. This is how you turn any stream-shaped
computation into a reusable consumer.

```ts
import { Stream, Sink } from "effect"

// A custom sink that collects elements and returns both the count and the sum,
// expressed as a small stream pipeline ending in an Effect.
const countAndSum = Sink.make<number>()((stream: Stream.Stream<number>) =>
  stream.pipe(
    Stream.runFold(
      () => ({ count: 0, sum: 0 }),
      (acc, n) => ({ count: acc.count + 1, sum: acc.sum + n })
    )
  )
)

export const summary = Stream.make(10, 20, 30).pipe(Stream.run(countAndSum))
// => { count: 3, sum: 60 }
```

Because a sink is a first-class value, you can name it, test it in isolation, and
reuse it across many streams — the same way you reuse an `Effect`. Reach for an
explicit sink whenever a consumption strategy is worth a name; for one-off
consumption, the [run\* destructors](https://effect.plants.sh/streaming/consuming-streams/) are the
shorter path.

---

## API reference

Everything below is run with `Stream.run(stream, sink)` (or
`stream.pipe(Stream.run(sink))`). Sinks pair naturally with the grouping
combinators on the [transforming streams](https://effect.plants.sh/streaming/transforming-streams/) page
(`Stream.transduce` / `Stream.aggregate` repeatedly feed a stream through a sink)
and with the [run\* destructors](https://effect.plants.sh/streaming/consuming-streams/).

### Leftovers and the `End` tuple

When a sink finishes it produces an `End<A, L>` — a tuple `[value, leftover?]`.
The optional second element is a *non-empty* array of input that was pulled from
upstream but not consumed. `Stream.run` returns only the `value`; leftovers
matter when one sink hands off to another (`flatMap`, `orElse`) or when you build
parsers. The `In` type parameter is contravariant, so a sink accepting broader
input can be used where narrower input is expected.

## Constructors

### make

Creates a pipe-style sink builder over input type `In`. The first pipeline step
receives the input as a `Stream<In>`; the final step must return an `Effect`
whose success becomes the result.

```ts
import { Sink, Stream } from "effect"

const lastTwo = Sink.make<number>()((s: Stream.Stream<number>) =>
  s.pipe(Stream.takeRight(2), Stream.runCollect)
)
// Stream.run(Stream.make(1, 2, 3), lastTwo) // => [2, 3]
```

### fromChannel

Builds a sink from a `Channel` that processes non-empty arrays of input and
emits an `End<A, L>`. Low-level; pairs with `toChannel`.

```ts
import { Sink } from "effect"

// const sink = Sink.fromChannel(channel)
// channel: Channel<never, E, End<A, L>, NonEmptyReadonlyArray<In>, never, void, R>
```

### fromTransform

The lowest-level constructor: you receive the upstream pull and the active scope
and return an effect completing with the sink's `End` value. Every other
constructor is built on this.

```ts
import { Effect, Sink, Stream } from "effect"

// Always succeed with 0, consuming nothing.
const zero = Sink.fromTransform(() => Effect.succeed([0] as const))
export const r = Stream.make(1, 2, 3).pipe(Stream.run(zero))
// => 0
```

### fromEffect

Ignores all upstream input and completes with the success value of an effect.

```ts
import { Effect, Sink, Stream } from "effect"

const sink = Sink.fromEffect(Effect.succeed("done"))
export const r = Stream.make(1, 2, 3).pipe(Stream.run(sink))
// => "done"
```

### fromEffectEnd

Like `fromEffect`, but the effect returns an `End<A, L>` so it can supply
both a result and explicit leftovers.

```ts
import { Effect, Sink, Stream } from "effect"

const sink = Sink.fromEffectEnd(Effect.succeed([42] as const))
export const r = Stream.make(1, 2, 3).pipe(Stream.run(sink))
// => 42
```

### fromQueue

Offers every consumed element to a `Queue`, then ends the queue when upstream
finishes. Completes with `void`.

```ts
import { Cause, Effect, Queue, Sink, Stream } from "effect"

export const program = Effect.gen(function* () {
  // `fromQueue` ends the queue when upstream finishes, so the queue's error
  // channel must include `Cause.Done`.
  const queue = yield* Queue.make<number, Cause.Done>()
  yield* Stream.make(1, 2, 3).pipe(Stream.run(Sink.fromQueue(queue)))
  return yield* Queue.takeAll(queue)
})
// => [1, 2, 3]
```

### fromPubSub

Publishes every consumed element to a `PubSub`. Completes with `void` when
upstream ends.

```ts
import { PubSub, Sink, Stream } from "effect"

declare const pubsub: PubSub.PubSub<number>
const sink = Sink.fromPubSub(pubsub)
// Stream.make(1, 2, 3).pipe(Stream.run(sink)) // => void, publishing each element
```

### succeed

Immediately ends with a constant value (and optional leftovers), ignoring input.

```ts
import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3).pipe(Stream.run(Sink.succeed(42)))
// => 42
```

### sync

Immediately ends with a lazily evaluated value.

```ts
import { Sink } from "effect"

const sink = Sink.sync(() => 42)
// Stream.run(stream, sink) // => 42, computed when the sink runs
```

### suspend

Defers construction of a sink to when it is run — useful for per-run mutable
state.

```ts
import { Sink, Stream } from "effect"

const counter = Sink.suspend(() => Sink.count)
export const r = Stream.make(1, 2, 3).pipe(Stream.run(counter))
// => 3
```

### fail

A sink that always fails with the given typed error.

```ts
import { Effect, Sink, Stream } from "effect"

const sink = Sink.fail("boom" as const)
Effect.runPromiseExit(Stream.make(1).pipe(Stream.run(sink)))
// => Exit.fail("boom")
```

### failSync

A sink that fails with a lazily evaluated error.

```ts
import { Sink } from "effect"

const sink = Sink.failSync(() => new Error("late"))
// Stream.run(stream, sink) // => fails with Error: late
```

### failCause

A sink that halts with a specified `Cause` (e.g. a typed failure or a defect).

```ts
import { Cause, Sink } from "effect"

const sink = Sink.failCause(Cause.fail("oops"))
// Stream.run(stream, sink) // => fails with "oops"
```

### failCauseSync

A sink that halts with a lazily evaluated `Cause`.

```ts
import { Cause, Sink } from "effect"

const sink = Sink.failCauseSync(() => Cause.die("defect"))
// Stream.run(stream, sink) // => dies with "defect"
```

### die

A sink that halts with an unrecoverable defect.

```ts
import { Sink } from "effect"

const sink = Sink.die(new Error("unexpected"))
// Stream.run(stream, sink) // => dies with Error: unexpected
```

### never

A sink that never completes. Useful as a placeholder or to keep a fiber alive.

```ts
import { Sink } from "effect"

const sink = Sink.never
// Stream.run(stream, sink) // => never resolves
```

### drain

Consumes and discards all input, completing with `void`.

```ts
import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3).pipe(Stream.run(Sink.drain))
// => undefined
```

## Folding

### fold

Folds input element by element with an effectful step, continuing while the
`while` predicate holds. Stopping mid-array yields the remaining elements as
leftovers.

```ts
import { Effect, Sink, Stream } from "effect"

const sumUnder10 = Sink.fold(
  () => 0,
  (total) => total < 10,
  (total, n: number) => Effect.succeed(total + n)
)
export const r = Stream.make(4, 5, 6, 7).pipe(Stream.run(sumUnder10))
// => 15 (4 + 5 + 6, stops once >= 10)
```

### foldArray

Folds each pulled non-empty array at once with an effectful step, continuing
while the predicate holds. Batch-level; produces no leftovers.

```ts
import { Effect, Sink } from "effect"

const totalLengths = Sink.foldArray(
  () => 0,
  () => true,
  (acc, chunk: ReadonlyArray<string>) => Effect.succeed(acc + chunk.length)
)
// Stream.run(stream, totalLengths) // => total number of elements seen
```

### foldUntil

Folds elements until a fixed maximum number have been consumed (or the stream
ends). Extra elements from the final array become leftovers.

```ts
import { Effect, Sink, Stream } from "effect"

// Sum at most 3 elements.
const sumFirst3 = Sink.foldUntil(
  () => 0,
  3,
  (acc, n: number) => Effect.succeed(acc + n)
)
export const r = Stream.make(1, 2, 3, 4, 5).pipe(Stream.run(sumFirst3))
// => 6 (1 + 2 + 3)
```

## Reducing

### reduce

Reduces every input element into state with a pure step. Consumes the whole
stream.

```ts
import { Sink, Stream } from "effect"

const product = Sink.reduce(() => 1, (acc, n: number) => acc * n)
export const r = Stream.make(1, 2, 3, 4).pipe(Stream.run(product))
// => 24
```

### reduceArray

Reduces each pulled non-empty array into state with a pure step.

```ts
import { Sink } from "effect"

const concat = Sink.reduceArray(
  () => "",
  (acc, chunk: ReadonlyArray<string>) => acc + chunk.join("")
)
// Stream.run(stream, concat) // => all chunks concatenated
```

### reduceEffect

Reduces every element into state with an *effectful* step (can fail, can require
services).

```ts
import { Effect, Sink } from "effect"

const sumLogged = Sink.reduceEffect(
  () => 0,
  (acc, n: number) => Effect.as(Effect.logInfo(`+${n}`), acc + n)
)
// Stream.run(stream, sumLogged) // => the sum, logging each element
```

### reduceWhile

Reduces with a pure step while a predicate on the state holds; stops otherwise,
returning unconsumed elements as leftovers.

```ts
import { Sink } from "effect"

const untilNegative = Sink.reduceWhile(
  () => 0,
  (sum) => sum >= 0,
  (sum, n: number) => sum + n
)
// Stream.run(stream, untilNegative) // => sum, stopping once it goes negative
```

### reduceWhileEffect

Effectful variant of `reduceWhile`: the step is an `Effect`.

```ts
import { Effect, Sink } from "effect"

const sink = Sink.reduceWhileEffect(
  () => 0,
  (sum) => sum < 100,
  (sum, n: number) => Effect.succeed(sum + n)
)
// Stream.run(stream, sink) // => running sum, stops at >= 100
```

### reduceWhileArray

Reduces non-empty input arrays with a pure step while a predicate holds. No
leftovers.

```ts
import { Sink } from "effect"

const sink = Sink.reduceWhileArray(
  () => 0,
  (n) => n < 10,
  (n, chunk: ReadonlyArray<number>) => n + chunk.length
)
// Stream.run(stream, sink) // => count, stopping once it reaches 10
```

### reduceWhileArrayEffect

Effectful array-level reduce that checks the predicate before consuming the next
array. No leftovers.

```ts
import { Effect, Sink } from "effect"

const sink = Sink.reduceWhileArrayEffect(
  () => 0,
  (n) => n < 10,
  (n, chunk: ReadonlyArray<number>) => Effect.succeed(n + chunk.length)
)
// Stream.run(stream, sink) // => count, stopping once it reaches 10
```

## Aggregates

### sum

Sums all numeric input.

```ts
import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3, 4).pipe(Stream.run(Sink.sum))
// => 10
```

### count

Counts the number of elements consumed.

```ts
import { Sink, Stream } from "effect"

export const r = Stream.make("a", "b", "c").pipe(Stream.run(Sink.count))
// => 3
```

### collect

Accumulates every element into an array.

```ts
import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3).pipe(Stream.run(Sink.collect<number>()))
// => [1, 2, 3]
```

### head

The first element as `Option`, or `Option.none()` if the stream is empty. Later
elements from the same pulled array become leftovers.

```ts
import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3).pipe(Stream.run(Sink.head<number>()))
// => Option.some(1)
```

### last

The final element as `Option`, or `Option.none()` if empty. Only completes when
upstream ends.

```ts
import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3).pipe(Stream.run(Sink.last<number>()))
// => Option.some(3)
```

### find

The first element matching a predicate (or `Refinement`, which narrows the
result type), as `Option`. Returns `Option.none()` if none match.

```ts
import { Sink, Stream } from "effect"

const firstEven = Sink.find((n: number) => n % 2 === 0)
export const r = Stream.make(1, 3, 4, 5).pipe(Stream.run(firstEven))
// => Option.some(4)
```

### findEffect

Like `find` but the predicate is an `Effect` (can fail / require services).

```ts
import { Effect, Sink } from "effect"

const sink = Sink.findEffect((n: number) => Effect.succeed(n > 10))
// Stream.run(stream, sink) // => Option of the first element > 10
```

### every

`true` only if *all* elements satisfy the predicate.

```ts
import { Sink, Stream } from "effect"

const allPositive = Sink.every((n: number) => n > 0)
export const r = Stream.make(1, 2, 3).pipe(Stream.run(allPositive))
// => true
```

### some

`true` if *any* element satisfies the predicate.

```ts
import { Sink, Stream } from "effect"

const anyNegative = Sink.some((n: number) => n < 0)
export const r = Stream.make(1, 2, 3).pipe(Stream.run(anyNegative))
// => false
```

## Taking

### take

Collects exactly up to `n` elements into an array. Extra elements from the final
array become leftovers; `n <= 0` yields an empty array.

```ts
import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3, 4, 5).pipe(Stream.run(Sink.take<number>(3)))
// => [1, 2, 3]
```

### takeWhile

Collects the longest prefix satisfying a predicate (or `Refinement`). The first
failing element is consumed and excluded; later elements become leftovers.

```ts
import { Sink, Stream } from "effect"

const ascendingPrefix = Sink.takeWhile((n: number) => n < 4)
export const r = Stream.make(1, 2, 3, 4, 1).pipe(Stream.run(ascendingPrefix))
// => [1, 2, 3]
```

### takeWhileFilter

Applies a `Filter` while it succeeds, collecting each transformed output. The
first failure stops collection.

```ts
import { Filter, Result, Sink, Stream } from "effect"

// Keep numbers below 5, mapping each to its double; stop at the first failure.
const doubleWhileSmall = Sink.takeWhileFilter(
  Filter.make((n: number) => (n < 5 ? Result.succeed(n * 2) : Result.fail(n)))
)
export const r = Stream.make(1, 2, 5, 3).pipe(Stream.run(doubleWhileSmall))
// => [2, 4]
```

### takeWhileEffect

Collects elements while an *effectful* predicate returns `true`. The first
`false` stops collection; later elements become leftovers.

```ts
import { Effect, Sink, Stream } from "effect"

const sink = Sink.takeWhileEffect((n: number) => Effect.succeed(n < 4))
export const r = Stream.make(1, 2, 3, 4, 5).pipe(Stream.run(sink))
// => [1, 2, 3]
```

### takeWhileFilterEffect

Applies a `FilterEffect` while it succeeds, collecting each transformed output.

```ts
import { Effect, Filter, Result, Sink } from "effect"

const sink = Sink.takeWhileFilterEffect(
  Filter.makeEffect((n: number) =>
    Effect.succeed(n < 5 ? Result.succeed(n) : Result.fail(n))
  )
)
// Stream.run(Stream.make(1, 2, 9), sink) // => [1, 2]
```

### takeUntil

Collects elements until the predicate returns `true`, *including* the matching
element.

```ts
import { Sink, Stream } from "effect"

const upToFirstEven = Sink.takeUntil((n: number) => n % 2 === 0)
export const r = Stream.make(1, 3, 4, 5).pipe(Stream.run(upToFirstEven))
// => [1, 3, 4]
```

### takeUntilEffect

Like `takeUntil` but the predicate is an `Effect`.

```ts
import { Effect, Sink, Stream } from "effect"

const sink = Sink.takeUntilEffect((n: number) => Effect.succeed(n > 3))
export const r = Stream.make(1, 2, 4, 5).pipe(Stream.run(sink))
// => [1, 2, 4]
```

## ForEach

### forEach

Runs an effect for every element, discarding the results, producing `void`.

```ts
import { Effect, Sink, Stream } from "effect"

export const program = Stream.make(1, 2, 3).pipe(
  Stream.run(Sink.forEach((n) => Effect.logInfo(`item ${n}`)))
)
// => void, logging "item 1", "item 2", "item 3"
```

### forEachArray

Runs an effect once per pulled non-empty array, producing `void`.

```ts
import { Effect, Sink, Stream } from "effect"

export const program = Stream.make(1, 2, 3).pipe(
  Stream.run(
    Sink.forEachArray((chunk) => Effect.logInfo(`chunk of ${chunk.length}`))
  )
)
// => void
```

### forEachWhile

Runs an effect per element while it returns `true`; stops when it returns
`false` or upstream ends.

```ts
import { Effect, Sink } from "effect"

const sink = Sink.forEachWhile((n: number) =>
  Effect.as(Effect.logInfo(`saw ${n}`), n < 3)
)
// Stream.run(Stream.make(1, 2, 3, 4), sink) // => void, logs 1, 2, 3 then stops
```

### forEachWhileArray

Runs an effect per non-empty array while it returns `true`.

```ts
import { Effect, Sink } from "effect"

const sink = Sink.forEachWhileArray((chunk: ReadonlyArray<number>) =>
  Effect.succeed(chunk.length > 0)
)
// Stream.run(stream, sink) // => void
```

## Transforming a sink

### map

Transforms the result once the sink completes.

```ts
import { Sink, Stream } from "effect"

const countAsString = Sink.count.pipe(Sink.map((n) => `count=${n}`))
export const r = Stream.make(1, 2).pipe(Stream.run(countAsString))
// => "count=2"
```

### as

Replaces the result with a constant, preserving input consumption.

```ts
import { Sink, Stream } from "effect"

const done = Sink.drain.pipe(Sink.as("done" as const))
export const r = Stream.make(1, 2, 3).pipe(Stream.run(done))
// => "done"
```

### mapInput

Adapts the input side with a pure function so the sink can consume a different
element type.

```ts
import { Sink, Stream } from "effect"

const sumOfLengths = Sink.sum.pipe(Sink.mapInput((s: string) => s.length))
export const r = Stream.make("ab", "cde").pipe(Stream.run(sumOfLengths))
// => 5
```

### mapInputEffect

Adapts the input side with an *effectful* function (can fail / require services).

```ts
import { Effect, Sink, Stream } from "effect"

const sink = Sink.sum.pipe(
  Sink.mapInputEffect((s: string) => Effect.succeed(Number(s)))
)
export const r = Stream.make("1", "2", "3").pipe(Stream.run(sink))
// => 6
```

### mapInputArray

Transforms each pulled non-empty input array before it reaches the sink.

```ts
import { Array, Sink, Stream } from "effect"

const sink = Sink.sum.pipe(
  Sink.mapInputArray((arr) => Array.map(arr, (s: string) => s.length))
)
export const r = Stream.make("a", "bb").pipe(Stream.run(sink))
// => 3
```

### mapInputArrayEffect

Effectful variant of `mapInputArray`, transforming each input array via an
`Effect`.

```ts
import { Array, Effect, Sink } from "effect"

const sink = Sink.sum.pipe(
  Sink.mapInputArrayEffect((arr) =>
    Effect.succeed(Array.map(arr, (s: string) => Number(s)))
  )
)
// Stream.run(Stream.make("1", "2"), sink) // => 3
```

### mapEnd

Transforms the full `End` tuple, so you can change both the result and the
leftovers.

```ts
import { Sink } from "effect"

// Take 2 elements but drop the reported leftovers.
const noLeftover = Sink.take<number>(2).pipe(Sink.mapEnd(([value]) => [value]))
// Stream.run(stream, noLeftover) // => [1, 2], leftover discarded
```

### mapEffectEnd

Effectfully transforms the full `End` tuple (result + leftovers).

```ts
import { Effect, Sink } from "effect"

const sink = Sink.take<number>(2).pipe(
  Sink.mapEffectEnd(([value, leftover]) =>
    Effect.succeed([value.length, leftover] as const)
  )
)
// Stream.run(stream, sink) // => 2
```

### mapEffect

Transforms the result with an `Effect` (can fail / require services).

```ts
import { Effect, Sink } from "effect"

const parsed = Sink.head<string>().pipe(
  Sink.mapEffect((opt) => Effect.succeed(opt))
)
// Stream.run(stream, parsed) // => Option of the first element
```

### mapError

Transforms the error channel.

```ts
import { Sink } from "effect"

const sink = Sink.fail("boom" as const).pipe(
  Sink.mapError((e) => new Error(e))
)
// Stream.run(stream, sink) // => fails with Error: boom
```

### mapLeftover

Transforms each leftover element.

```ts
import { Sink } from "effect"

const sink = Sink.take<number>(2).pipe(Sink.mapLeftover((n) => n * 10))
// leftovers (e.g. [3, 4, 5]) become [30, 40, 50]
```

### flatMap

Runs a second sink after the first, choosing it from the first result. Leftovers
of the first sink are fed to the second before pulling more upstream input.

```ts
import { Sink, Stream } from "effect"

const takeThenCount = Sink.take<number>(2).pipe(
  Sink.flatMap((first) => Sink.count.pipe(Sink.map((rest) => ({ first, rest }))))
)
export const r = Stream.make(1, 2, 3, 4, 5).pipe(Stream.run(takeThenCount))
// => { first: [1, 2], rest: 3 }
```

### ignoreLeftover

Drops any leftovers a sink would produce, keeping the result. Does not pull
additional input.

```ts
import { Sink, Stream } from "effect"

const firstTwo = Sink.take<number>(2).pipe(Sink.ignoreLeftover)
export const r = Stream.make(1, 2, 3).pipe(Stream.run(firstTwo))
// => [1, 2], leftover [3] discarded
```

## Combinators and error handling

### orElse

Switches to a fallback sink (built from the error) if the first sink fails,
continuing to consume from the same upstream.

```ts
import { Sink, Stream } from "effect"

const sink = Sink.fail("nope" as const).pipe(
  Sink.orElse(() => Sink.succeed("recovered" as const))
)
export const r = Stream.make(1, 2, 3).pipe(Stream.run(sink))
// => "recovered"
```

### catch

Handles typed errors with an effectful fallback value (exported as `Sink.catch`).

```ts
import { Effect, Sink, Stream } from "effect"

const sink = Sink.fail("nope" as const).pipe(
  Sink.catch((e) => Effect.succeed(`handled: ${e}`))
)
export const r = Stream.make(1).pipe(Stream.run(sink))
// => "handled: nope"
```

### catchCause

Recovers based on the full `Cause` (defects, interruption, typed errors).

```ts
import { Cause, Effect, Sink, Stream } from "effect"

const sink = Sink.fail("nope" as const).pipe(
  Sink.catchCause((cause) => Effect.succeed(Cause.squash(cause)))
)
export const r = Stream.make(1).pipe(Stream.run(sink))
// => "nope"
```

## Timing

### summarized

Runs a summary effect before and after the sink, pairing the result with a value
computed from the two summaries.

```ts
import { Clock, Duration, Sink } from "effect"

const timedDrain = Sink.drain.pipe(
  Sink.summarized(
    Clock.currentTimeMillis,
    (start, end) => Duration.millis(end - start)
  )
)
// Stream.run(stream, timedDrain) // => [void, Duration]
```

### withDuration

Pairs the sink's result with the time it took to run.

```ts
import { Sink } from "effect"

const sink = Sink.withDuration(Sink.collect<number>())
// Stream.run(stream, sink) // => [Array<number>, Duration]
```

### timed

Drains all input and returns the elapsed duration.

```ts
import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3).pipe(Stream.run(Sink.timed))
// => Duration (elapsed while draining)
```

## Lifecycle and context

### onExit

Runs an effect after the sink completes, fails, or is interrupted, receiving the
result `Exit`. The original result and leftovers are preserved.

```ts
import { Effect, Sink } from "effect"

const sink = Sink.collect<number>().pipe(
  Sink.onExit((exit) => Effect.logInfo(`done: ${exit._tag}`))
)
// Stream.run(stream, sink) // => the collected array, logging on exit
```

### ensuring

Runs a finalizer effect after the sink completes, fails, or is interrupted.

```ts
import { Effect, Sink } from "effect"

const sink = Sink.drain.pipe(
  Sink.ensuring(Effect.logInfo("sink finished"))
)
// Stream.run(stream, sink) // => void, always logs at the end
```

### provideContext

Provides a `Context` to the sink, removing those services from its requirements.

```ts
import { Context, Sink } from "effect"

declare const context: Context.Context<never>
const sink = Sink.drain.pipe(Sink.provideContext(context))
```

### provideService

Provides a single service implementation, removing it from the requirements.

```ts
import { Context, Effect, Sink, Stream } from "effect"

class Config extends Context.Service<Config, { readonly factor: number }>()(
  "Config"
) {}

const sink = Sink.reduceEffect(
  () => 0,
  (acc, n: number) => Config.pipe(Effect.map((c) => acc + n * c.factor))
).pipe(Sink.provideService(Config, { factor: 2 }))

export const r = Stream.make(1, 2, 3).pipe(Stream.run(sink))
// => 12
```

### unwrap

Builds a sink from a (possibly scoped) effect that produces a sink.

```ts
import { Effect, Sink } from "effect"

const sink = Sink.unwrap(
  Effect.succeed(Sink.forEach((n: number) => Effect.logInfo(`${n}`)))
)
// Stream.run(stream, sink) // => void
```

## Interop and guards

### toChannel

Converts a sink back into the underlying `Channel`.

```ts
import { Sink } from "effect"

const channel = Sink.toChannel(Sink.succeed(42))
```

### isSink

Type guard that checks whether a value is a `Sink`.

```ts
import { Sink } from "effect"

console.log(Sink.isSink(Sink.never)) // => true
console.log(Sink.isSink({ data: [1, 2, 3] })) // => false
```