# Encoding & Decoding

Streams of bytes or text rarely arrive as the values you actually want to work
with — they arrive as raw `Uint8Array` chunks, NDJSON lines, MessagePack frames,
or Server-Sent Events. Effect models each codec as a `Channel` (a stage that
consumes input chunks and produces output chunks) and lets you splice it into a
`Stream` with `Stream.pipeThroughChannel`, so decoding/encoding becomes just
another step in the pipeline.

```ts
import { Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

// Raw NDJSON text — one JSON object per line — as it might arrive from a file
// or socket.
const raw = Stream.make(
  '{"timestamp":"2025-06-01T00:00:00Z","level":"info","message":"start"}\n' +
    '{"timestamp":"2025-06-01T00:00:01Z","level":"error","message":"oops"}\n'
)

// `Ndjson.decodeString()` is a Channel that splits incoming strings on newlines
// and `JSON.parse`s each line. Splice it in with `Stream.pipeThroughChannel`.
export const decoded = raw.pipe(
  Stream.pipeThroughChannel(Ndjson.decodeString()),
  Stream.runCollect
)
// => [{ timestamp, level, message }, { ... }]
```
**Note:** `Ndjson`, `Msgpack`, and `Sse` live under `effect/unstable/encoding`. Unstable
  modules have stable behaviour but may see API tweaks before they graduate to
  the core namespace. The lower-level text codecs (`Stream.decodeText`,
  `Stream.splitLines`) and the schema-channel adapters (`ChannelSchema`) are part
  of the stable core `effect` package.

## Text codecs on a Stream

Before any framing, you usually need to turn raw bytes into text (or text back
into bytes). `Stream.decodeText`, `Stream.encodeText`, and `Stream.splitLines`
operate directly on a `Stream` and need no channel plumbing.

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

// Bytes in, text out. Internally uses a streaming TextDecoder, so multi-byte
// characters that straddle a chunk boundary are handled correctly.
const bytes = Stream.make(
  new TextEncoder().encode("Hello "),
  new TextEncoder().encode("World")
)

export const text = bytes.pipe(Stream.decodeText(), Stream.runCollect)
// => ["Hello ", "World"]

// Text in, UTF-8 bytes out — the inverse direction.
export const encoded = Stream.make("Hello", " ", "World").pipe(
  Stream.encodeText,
  Stream.runCollect
)
// => [Uint8Array[72,101,108,108,111], Uint8Array[32], Uint8Array[87,111,114,108,100]]

// Split a text stream into lines, handling \n, \r, and \r\n across chunk
// boundaries. Trailing delimiters do not produce a spurious empty line.
export const lines = Stream.make("a\nb\r\n", "c\n").pipe(
  Stream.splitLines,
  Stream.runCollect
)
// => ["a", "b", "c"]
```

`Stream.decodeText` accepts an optional encoding, e.g.
`Stream.decodeText({ encoding: "utf-16le" })`, forwarded to the underlying
`TextDecoder`.

## Splicing codecs in: `pipeThroughChannel`

Every codec in this page is a `Channel`. There are three ways to attach one to a
stream; the first is what you will reach for almost every time.

### `Stream.pipeThroughChannel`

Runs the stream's chunks through a channel and adopts the channel's error type.
Upstream stream errors are passed *into* the channel, so the resulting error
type is just the channel's (`E2`).

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

// A trivial channel: stringify every chunk element.
const stringify = Channel.identity<
  readonly [number, ...Array<number>],
  never,
  unknown
>().pipe(Channel.map((chunk) => chunk.map(String)))

export const out = Stream.make(1, 2, 3).pipe(
  Stream.pipeThroughChannel(stringify),
  Stream.runCollect
)
// => ["1", "2", "3"]
```

### `Stream.pipeThroughChannelOrFail`

Same splice, but upstream stream failures are *not* fed to the channel, so the
resulting stream can fail with either the original stream error or the channel
error (`E | E2`). Use this when the channel should only see successful values.

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

const stringify = Channel.identity<
  readonly [number, ...Array<number>],
  "StreamError",
  unknown
>().pipe(Channel.map((chunk) => chunk.map(String)))

export const out = Stream.make(1, 2, 3).pipe(
  Stream.rechunk(2),
  Stream.pipeThroughChannelOrFail(stringify),
  Stream.runCollect
)
// => ["1", "2", "3"]
```

### `Stream.pipeThrough`

A different tool with a confusingly similar name: it pipes the stream through a
`Sink` and emits the sink's *leftovers*. Reach for this when a sink stops partway
and you want the unconsumed tail as a stream — not for codecs.

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

export const leftovers = Stream.make(1, 2, 3, 4).pipe(
  Stream.pipeThrough(Sink.take(2)),
  Stream.runCollect
)
// => [3, 4]
```

## A realistic pipeline

Because decode and encode are both just channels, the common
*decode → transform → re-encode* shape reads as one linear pipeline. Here we
decode NDJSON into validated values, filter, then re-encode.

```ts
import { Schema, Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

class LogEntry extends Schema.Class<LogEntry>("LogEntry")({
  timestamp: Schema.String,
  level: Schema.Literals(["info", "warn", "error"]),
  message: Schema.String
}) {}

const ndjsonInput = Stream.make(
  '{"timestamp":"2025-06-01T00:00:00Z","level":"info","message":"ok"}\n' +
    '{"timestamp":"2025-06-01T00:00:01Z","level":"error","message":"fail"}\n' +
    '{"timestamp":"2025-06-01T00:00:02Z","level":"warn","message":"slow"}\n'
)

export const errorsOnly = ndjsonInput.pipe(
  // 1. Decode each line into a validated LogEntry.
  Stream.pipeThroughChannel(Ndjson.decodeSchemaString(LogEntry)()),
  // 2. Keep only error-level entries (an ordinary stream transform).
  Stream.filter((entry) => entry.level === "error"),
  // 3. Re-encode the survivors back to NDJSON strings.
  Stream.pipeThroughChannel(Ndjson.encodeSchemaString(LogEntry)()),
  Stream.runCollect
)
// => ['{"timestamp":"2025-06-01T00:00:01Z","level":"error","message":"fail"}\n']
```
**Caution:** The schema-aware NDJSON/Msgpack constructors are *curried*: you call them with
  the schema, then call the returned function with no arguments to get the
  channel — `Ndjson.decodeSchemaString(LogEntry)()`. The trailing `()` fixes the
  optional input-error and done type parameters.

---

## NDJSON — `effect/unstable/encoding/Ndjson`

Newline-delimited JSON: one complete JSON value per line. Decoders accumulate
partial lines across chunks; encoders append a trailing newline as record
framing. Byte helpers handle UTF-8 transport, string helpers handle
already-decoded text, and `Schema` helpers validate or transform each record.

### `Ndjson.decodeString`

Channel that splits incoming `string` chunks on newlines and `JSON.parse`s each
line into `unknown`. Pass `{ ignoreEmptyLines: true }` to skip blank lines
(otherwise a blank line fails as invalid JSON).

```ts
import { Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

export const out = Stream.make('{"a":1}\n', '{"a":2}\n').pipe(
  Stream.pipeThroughChannel(Ndjson.decodeString({ ignoreEmptyLines: true })),
  Stream.runCollect
)
// => [{ a: 1 }, { a: 2 }]
```

### `Ndjson.encodeString`

Channel that serializes each value with `JSON.stringify`, joins a chunk with
newlines, and appends a trailing newline. The inverse of `decodeString`.

```ts
import { Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

export const out = Stream.make({ a: 1 }, { a: 2 }).pipe(
  Stream.pipeThroughChannel(Ndjson.encodeString()),
  Stream.runCollect
)
// => ['{"a":1}\n', '{"a":2}\n']
```

### `Ndjson.decode`

Like `decodeString` but consumes UTF-8 `Uint8Array` chunks (it pipes through
`Channel.decodeText` first). Use it for raw byte transports such as sockets or
file descriptors.

```ts
import { Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

const bytes = Stream.make(new TextEncoder().encode('{"a":1}\n{"a":2}\n'))

export const out = bytes.pipe(
  Stream.pipeThroughChannel(Ndjson.decode()),
  Stream.runCollect
)
// => [{ a: 1 }, { a: 2 }]
```

### `Ndjson.encode`

Like `encodeString` but produces UTF-8 `Uint8Array` chunks, ready to write to a
byte sink.

```ts
import { Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

export const out = Stream.make({ a: 1 }).pipe(
  Stream.pipeThroughChannel(Ndjson.encode()),
  Stream.runCollect
)
// => [Uint8Array(...) /* '{"a":1}\n' as UTF-8 bytes */]
```

### `Ndjson.decodeSchemaString` / `Ndjson.decodeSchema`

Parse each NDJSON line, then decode the parsed value with a `Schema`, yielding a
validated, typed value. `decodeSchemaString` consumes strings;
`decodeSchema` consumes `Uint8Array` bytes. Validation failures surface as
`Schema.SchemaError`; malformed JSON as `NdjsonError`.

```ts
import { Schema, Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

const User = Schema.Struct({ id: Schema.Number, name: Schema.String })

export const out = Stream.make('{"id":1,"name":"Ada"}\n').pipe(
  Stream.pipeThroughChannel(Ndjson.decodeSchemaString(User)()),
  Stream.runCollect
)
// => [{ id: 1, name: "Ada" }]  (typed as { id: number; name: string })
```

### `Ndjson.encodeSchemaString` / `Ndjson.encodeSchema`

Run each value *through* the schema (applying encoding transformations) and then
serialize to NDJSON. `encodeSchemaString` emits strings; `encodeSchema` emits
UTF-8 bytes.

```ts
import { Schema, Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

const User = Schema.Struct({ id: Schema.Number, name: Schema.String })

export const out = Stream.make({ id: 1, name: "Ada" }).pipe(
  Stream.pipeThroughChannel(Ndjson.encodeSchemaString(User)()),
  Stream.runCollect
)
// => ['{"id":1,"name":"Ada"}\n']
```

### `Ndjson.duplex` / `Ndjson.duplexString`

Wrap a *bidirectional* byte (or string) channel — such as a socket — so callers
send and receive decoded values while the wrapped channel still moves NDJSON
bytes/strings. Outgoing values are encoded; incoming bytes/strings are decoded.

```ts
import { Channel } from "effect"
import { Ndjson } from "effect/unstable/encoding"

// `socket` is some byte-level duplex Channel<Uint8Array[], …, Uint8Array[], …>.
declare const socket: Channel.Channel<
  ReadonlyArray<Uint8Array> & { 0: Uint8Array },
  never,
  unknown,
  ReadonlyArray<Uint8Array> & { 0: Uint8Array },
  Ndjson.NdjsonError,
  unknown
>

// Now the channel emits decoded `unknown` values and accepts `unknown` to send.
export const jsonSocket = Ndjson.duplex(socket, { ignoreEmptyLines: true })
```

### `Ndjson.duplexSchema` / `Ndjson.duplexSchemaString`

The schema-aware duplex: values you send are encoded with `inputSchema`,
values you receive are decoded with `outputSchema`. Ideal for typed
request/response framing over a single connection.

```ts
import { Channel, Schema } from "effect"
import { Ndjson } from "effect/unstable/encoding"

const Request = Schema.Struct({ method: Schema.String })
const Response = Schema.Struct({ ok: Schema.Boolean })

declare const socket: Channel.Channel<
  ReadonlyArray<Uint8Array> & { 0: Uint8Array },
  never,
  unknown,
  ReadonlyArray<Uint8Array> & { 0: Uint8Array },
  Ndjson.NdjsonError | Schema.SchemaError,
  unknown
>

export const typed = Ndjson.duplexSchema(socket, {
  inputSchema: Request,
  outputSchema: Response
})
// send { method } values, receive { ok } values
```

### `Ndjson.NdjsonError`

Tagged error raised on codec failure. `kind` is `"Unpack"` for decode failures
and `"Pack"` for encode failures; `cause` holds the underlying exception. Catch
it with `Stream.catchTag` (see [Error handling](https://effect.plants.sh/streaming/error-handling/)).

```ts
import { Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

export const recovered = Stream.make("not-valid-json\n").pipe(
  Stream.pipeThroughChannel(Ndjson.decodeString()),
  Stream.catchTag("NdjsonError", (err) =>
    Stream.succeed({ recovered: true, kind: err.kind })
  ),
  Stream.runCollect
)
// => [{ recovered: true, kind: "Unpack" }]
```

---

## MessagePack — `effect/unstable/encoding/Msgpack`

A compact binary serialization format for protocols and storage that expect
bytes rather than JSON text (RPC transports, socket streams, caches, DB
columns). The shape mirrors `Ndjson`, but everything is `Uint8Array` and there is
no string variant.

### `Msgpack.encode`

Channel that packs each value into MessagePack bytes (one byte array per value).
Packing failures fail with `MsgPackError`.

```ts
import { Stream } from "effect"
import { Msgpack } from "effect/unstable/encoding"

export const out = Stream.make({ a: 1 }, { a: 2 }).pipe(
  Stream.pipeThroughChannel(Msgpack.encode()),
  Stream.runCollect
)
// => [Uint8Array(...), Uint8Array(...)]  // two packed frames
```

### `Msgpack.decode`

Channel that unpacks MessagePack byte chunks into values. Incomplete frames are
buffered across chunks until enough bytes arrive, so it tolerates arbitrary chunk
boundaries. Invalid data fails with `MsgPackError`.

```ts
import { Stream } from "effect"
import { Msgpack } from "effect/unstable/encoding"

// Round-trip: encode then decode.
export const out = Stream.make({ a: 1 }, { a: 2 }).pipe(
  Stream.pipeThroughChannel(Msgpack.encode()),
  Stream.pipeThroughChannel(Msgpack.decode()),
  Stream.runCollect
)
// => [{ a: 1 }, { a: 2 }]
```

### `Msgpack.encodeSchema`

Encode each value through a `Schema`, then pack as MessagePack bytes. Can fail
with either `Schema.SchemaError` or `MsgPackError`.

```ts
import { Schema, Stream } from "effect"
import { Msgpack } from "effect/unstable/encoding"

const Point = Schema.Struct({ x: Schema.Number, y: Schema.Number })

export const out = Stream.make({ x: 1, y: 2 }).pipe(
  Stream.pipeThroughChannel(Msgpack.encodeSchema(Point)()),
  Stream.runCollect
)
// => [Uint8Array(...)]  // packed { x: 1, y: 2 }
```

### `Msgpack.decodeSchema`

Unpack bytes into `unknown`, then decode each value with the schema, producing
typed, validated values.

```ts
import { Schema, Stream } from "effect"
import { Msgpack } from "effect/unstable/encoding"

const Point = Schema.Struct({ x: Schema.Number, y: Schema.Number })

export const out = Stream.make({ x: 1, y: 2 }).pipe(
  Stream.pipeThroughChannel(Msgpack.encodeSchema(Point)()),
  Stream.pipeThroughChannel(Msgpack.decodeSchema(Point)()),
  Stream.runCollect
)
// => [{ x: 1, y: 2 }]
```

### `Msgpack.duplex` / `Msgpack.duplexSchema`

Wrap a bidirectional byte channel with MessagePack framing. `duplex` exposes
`unknown` values; `duplexSchema` exposes typed input/output via `inputSchema`
and `outputSchema`. Same role as the NDJSON duplex helpers.

```ts
import { Channel, Schema } from "effect"
import { Msgpack } from "effect/unstable/encoding"

const Req = Schema.Struct({ q: Schema.String })
const Res = Schema.Struct({ hits: Schema.Number })

declare const socket: Channel.Channel<
  ReadonlyArray<Uint8Array<ArrayBuffer>> & { 0: Uint8Array<ArrayBuffer> },
  never,
  unknown,
  ReadonlyArray<Uint8Array<ArrayBuffer>> & { 0: Uint8Array<ArrayBuffer> },
  Msgpack.MsgPackError | Schema.SchemaError,
  unknown
>

export const typed = Msgpack.duplexSchema(socket, {
  inputSchema: Req,
  outputSchema: Res
})
```

### `Msgpack.schema` / `Msgpack.transformation`

`Msgpack.schema(inner)` builds a *Schema* (not a channel) that stores values as
MessagePack bytes: decoding a `Uint8Array` unpacks then runs `inner`, encoding
runs `inner` then packs. `Msgpack.transformation` is the underlying
`Transformation` it is built from. Useful for persisting a MessagePack blob as a
struct field or a DB column.

```ts
import { Schema } from "effect"
import { Msgpack } from "effect/unstable/encoding"

const Session = Schema.Struct({ userId: Schema.Number, roles: Schema.Array(Schema.String) })

// A schema whose Encoded side is Uint8Array (MessagePack bytes).
const SessionBlob = Msgpack.schema(Session)

export const session = Schema.decodeSync(SessionBlob)(
  Schema.encodeSync(SessionBlob)({ userId: 1, roles: ["admin"] })
)
// => { userId: 1, roles: ["admin"] }
```

### `Msgpack.MsgPackError`

Tagged error for codec failures. `kind` is `"Pack"` or `"Unpack"`; `cause` holds
the original error. Catch with `Stream.catchTag("MsgPackError", …)`.

```ts
import { Stream } from "effect"
import { Msgpack } from "effect/unstable/encoding"

const garbage = Stream.make(new Uint8Array([0xc1]) as Uint8Array<ArrayBuffer>)

export const recovered = garbage.pipe(
  Stream.pipeThroughChannel(Msgpack.decode()),
  Stream.catchTag("MsgPackError", (err) => Stream.succeed(err.kind)),
  Stream.runCollect
)
// => ["Unpack"]
```

---

## Server-Sent Events — `effect/unstable/encoding/Sse`

SSE is the `EventSource` wire format: line-oriented text for unidirectional
server-to-client streams (live updates, notifications, progress feeds). A blank
line dispatches an event; repeated `data:` lines join with newlines; `retry:`
directives surface as `Retry` failures.

### `Sse.decode`

Channel that parses `string` chunks into `Sse.Event` values. Chunk boundaries may
split fields; the parser buffers until a full event is available. A `retry:`
directive fails the channel with `Sse.Retry`.

```ts
import { Stream } from "effect"
import { Sse } from "effect/unstable/encoding"

const wire = Stream.make("event: ping\ndata: hello\n\n")

export const out = wire.pipe(
  Stream.pipeThroughChannel(Sse.decode()),
  Stream.runCollect
)
// => [{ _tag: "Event", event: "ping", id: undefined, data: "hello" }]
```

### `Sse.decodeSchema`

Decode each parsed event with a schema that receives the untagged
`{ id?, event, data }` shape — handy for renaming/validating fields. Fails with
`Schema.SchemaError`, `Sse.Retry`, or the input error.

```ts
import { Schema, Stream } from "effect"
import { Sse } from "effect/unstable/encoding"

// Keep only the event name and data, validating both are strings.
const Named = Schema.Struct({ event: Schema.String, data: Schema.String })

export const out = Stream.make("event: tick\ndata: 1\n\n").pipe(
  Stream.pipeThroughChannel(Sse.decodeSchema(Named)),
  Stream.runCollect
)
// => [{ event: "tick", data: "1" }]
```

### `Sse.decodeDataSchema`

The most common SSE need: JSON-decode each event's `data` field with a schema,
keeping the `event` name and optional `id`. The output `data` is the decoded
domain value.

```ts
import { Schema, Stream } from "effect"
import { Sse } from "effect/unstable/encoding"

const Progress = Schema.Struct({ pct: Schema.Number })

export const out = Stream.make(
  'event: progress\ndata: {"pct":42}\n\n'
).pipe(
  Stream.pipeThroughChannel(Sse.decodeDataSchema(Progress)),
  Stream.runCollect
)
// => [{ event: "progress", id: undefined, data: { pct: 42 } }]
```

### `Sse.encode`

Channel that renders `Sse.Event` values as SSE text. An upstream `Sse.Retry`
failure is written as a `retry:` directive and completes the encoder.

```ts
import { Stream } from "effect"
import { Sse } from "effect/unstable/encoding"

const event: Sse.Event = {
  _tag: "Event",
  event: "ping",
  id: "1",
  data: "hello"
}

export const out = Stream.make(event).pipe(
  Stream.pipeThroughChannel(Sse.encode()),
  Stream.runCollect
)
// => ["id: 1\nevent: ping\ndata: hello\n\n"]
```

### `Sse.encodeSchema`

Schema-encode domain values to the untagged SSE shape, convert to `Event`, then
render as SSE text. The inverse pairing for `decodeSchema`.

```ts
import { Schema, Stream } from "effect"
import { Sse } from "effect/unstable/encoding"

const Named = Schema.Struct({ event: Schema.String, data: Schema.String })

export const out = Stream.make({ event: "tick", data: "1" }).pipe(
  Stream.pipeThroughChannel(Sse.encodeSchema(Named)),
  Stream.runCollect
)
// => ["event: tick\ndata: 1\n\n"]
```

### `Sse.makeParser` / `Sse.Parser`

The stateful parser underneath `decode`, for manual feeding outside a stream.
Call `feed(chunk)` to push text; your callback receives `Sse.Event` or
`Sse.Retry` values. `reset()` clears buffered state. `Parser` is the returned
interface.

```ts
import { Sse } from "effect/unstable/encoding"

const events: Array<Sse.AnyEvent> = []
const parser: Sse.Parser = Sse.makeParser((event) => events.push(event))

parser.feed("data: a\n\n")
parser.feed("data: b\n\n")
// events => [{ _tag: "Event", data: "a", ... }, { _tag: "Event", data: "b", ... }]
```

### `Sse.encoder` / `Sse.Encoder`

The default `Encoder`: its `write(event)` turns a single `Sse.Event` or
`Sse.Retry` into SSE text. `Sse.encode()` uses it internally; reach for `encoder`
when you render one event at a time. The `event:` line is omitted for the default
`"message"` name.

```ts
import { Duration } from "effect"
import { Sse } from "effect/unstable/encoding"

export const line = Sse.encoder.write({
  _tag: "Event",
  event: "message",
  id: undefined,
  data: "hi"
})
// => "data: hi\n\n"   (no `event:` line for the default name)

export const retry = Sse.encoder.write(
  new Sse.Retry({ duration: Duration.seconds(3), lastEventId: undefined })
)
// => "retry: 3000\n\n"
```

### `Sse.Event` / `Sse.EventEncoded` / `Sse.AnyEvent`

`Sse.Event` is both the `interface` for a parsed event (`{ _tag: "Event", event,
id, data }`) and a `Schema` for that tagged shape. `Sse.EventEncoded` is the
untagged `{ id?, event, data }` shape (interface + schema). `Sse.AnyEvent` is the
union `Event | Retry` that an `Encoder` can render.

```ts
import { Schema } from "effect"
import { Sse } from "effect/unstable/encoding"

// `Sse.EventEncoded` is a real Schema you can decode/encode with.
export const decoded = Schema.decodeUnknownSync(Sse.EventEncoded)({
  id: "7",
  event: "ping",
  data: "x"
})
// => { id: "7", event: "ping", data: "x" }
```

### `Sse.Retry`

Tagged class representing a `retry:` directive. Decoders surface it as a *failure*
(reconnect after `duration`); encoders serialize an upstream `Retry` failure as a
`retry:` line. `Sse.Retry.is` is a guard and `Sse.Retry.filter` separates retries
from events.

```ts
import { Duration } from "effect"
import { Sse } from "effect/unstable/encoding"

const r = new Sse.Retry({
  duration: Duration.seconds(5),
  lastEventId: "42"
})

export const isRetry = Sse.Retry.is(r) // => true
export const ms = Duration.toMillis(r.duration) // => 5000
```

### `Sse.transformEvent`

The `Transformation` that maps the untagged SSE payload to the tagged `Event`
model. `encodeSchema` uses it internally; you rarely call it directly, but it is
exported for building custom schema pipelines.

```ts
import { Sse } from "effect/unstable/encoding"

// `Sse.transformEvent` is a SchemaTransformation used inside `Sse.encodeSchema`;
// reference it when composing your own Event schema.
export const t = Sse.transformEvent
```

---

## Schema-driven channel codecs — `ChannelSchema`

`ChannelSchema` (in core `effect`) is the generic machinery the NDJSON/Msgpack
schema helpers are built on: it applies a `Schema` at a `Channel` boundary so one
side works with typed values while the other works with the encoded
representation. Schemas are applied to *non-empty chunks*, and schema services
become part of the channel's requirements. Use it directly when you have a custom
transport whose encoded form is already structured data.

### `ChannelSchema.encode`

Channel that encodes typed input chunks (`S["Type"]`) into the schema's encoded
representation (`S["Encoded"]`). Failures surface as `Schema.SchemaError`.

```ts
import { ChannelSchema, Schema, Stream } from "effect"

// Type side is `number`; Encoded side is `string`.
const Count = Schema.NumberFromString

export const out = Stream.make(1, 2, 3).pipe(
  Stream.pipeThroughChannel(ChannelSchema.encode(Count)()),
  Stream.runCollect
)
// => ["1", "2", "3"]  (encoded to the schema's Encoded representation)
```

### `ChannelSchema.encodeUnknown`

Identical to `encode` but the encoded output chunks are typed as `unknown` —
for boundaries where the encoded representation is intentionally erased.

```ts
import { ChannelSchema, Schema, Stream } from "effect"

const User = Schema.Struct({ id: Schema.Number })

export const out = Stream.make({ id: 1 }).pipe(
  Stream.pipeThroughChannel(ChannelSchema.encodeUnknown(User)()),
  Stream.runCollect
)
// => [{ id: 1 }]  (typed as unknown[])
```

### `ChannelSchema.decode`

Channel that decodes encoded input chunks (`S["Encoded"]`) into typed schema
values (`S["Type"]`), validating along the way.

```ts
import { ChannelSchema, Schema, Stream } from "effect"

const User = Schema.Struct({ id: Schema.Number, name: Schema.String })

export const out = Stream.make({ id: 1, name: "Ada" }).pipe(
  Stream.pipeThroughChannel(ChannelSchema.decode(User)()),
  Stream.runCollect
)
// => [{ id: 1, name: "Ada" }]  (validated)
```

### `ChannelSchema.decodeUnknown`

`decode` for boundaries where the encoded *input* is `unknown` (e.g. freshly
`JSON.parse`d data). Only the decoded output is statically typed.

```ts
import { ChannelSchema, Schema, Stream } from "effect"

const User = Schema.Struct({ id: Schema.Number })

// Simulate untyped input from a parser.
const parsed: ReadonlyArray<unknown> = [{ id: 1 }, { id: 2 }]

export const out = Stream.fromArray(parsed).pipe(
  Stream.pipeThroughChannel(ChannelSchema.decodeUnknown(User)()),
  Stream.runCollect
)
// => [{ id: 1 }, { id: 2 }]
```

### `ChannelSchema.duplex`

Wrap a bidirectional encoded channel so callers send/receive typed chunks:
input is encoded with `inputSchema`, output is decoded with `outputSchema`. The
wrapped channel keeps operating on the schemas' *encoded* chunks.

```ts
import { ChannelSchema, Channel, Schema } from "effect"

const Req = Schema.Struct({ q: Schema.String })
const Res = Schema.Struct({ n: Schema.Number })

// Wrapped channel speaks the encoded representations.
declare const transport: Channel.Channel<
  ReadonlyArray<typeof Res.Encoded> & { 0: typeof Res.Encoded },
  never,
  unknown,
  ReadonlyArray<typeof Req.Encoded> & { 0: typeof Req.Encoded },
  Schema.SchemaError,
  unknown
>

export const typed = ChannelSchema.duplex(transport, {
  inputSchema: Req,
  outputSchema: Res
})
// send { q } values, receive { n } values
```

### `ChannelSchema.duplexUnknown`

`duplex` for a wrapped channel whose encoded chunks are typed as `unknown` on
both sides — the variant the `Ndjson`/`Msgpack` `duplexSchema` helpers build on.

```ts
import { ChannelSchema, Channel, Schema } from "effect"

const Req = Schema.Struct({ q: Schema.String })
const Res = Schema.Struct({ n: Schema.Number })

declare const transport: Channel.Channel<
  ReadonlyArray<unknown> & { 0: unknown },
  never,
  unknown,
  ReadonlyArray<unknown> & { 0: unknown },
  Schema.SchemaError,
  unknown
>

export const typed = ChannelSchema.duplexUnknown(transport, {
  inputSchema: Req,
  outputSchema: Res
})
```