> **Can't find what you're looking for?** Use `search_docs` on the docs MCP server at `https://docs.ampdb.io/api/mcp` to find what you need.

# Effect Basics

The TypeScript SDK is built on [Effect](https://effect.website), a library for writing programs that can fail, run
concurrently, and use resources like network connections. Its functions return Effect values instead of promises, and
you set it up with Effect's layers. This page explains the parts of Effect you need to use the SDK. If you'd rather not
write Effect code, see [Use the SDK from async code](/sdks/typescript/effect#use-the-sdk-from-async-code).

The SDK uses Effect 4. Effect's own [documentation](https://effect.website/docs) goes further than this page.

## Effects

An Effect is a description of work: running a query, say. Creating one does nothing. Work happens when you run it, and
you can run the same Effect many times. Its type, `Effect<A, E, R>`, says three things:

| Parameter | What it is                                                                   | For `flight.query(sql)`       |
| --------- | ---------------------------------------------------------------------------- | ----------------------------- |
| `A`       | What it succeeds with                                                        | The result batches            |
| `E`       | The errors it can fail with, which you can handle (`never` if it can't fail) | `ArrowFlightError`            |
| `R`       | The services it needs to run (`never` if none)                               | `never`: it has what it needs |

`Effect.runPromise` runs an Effect and returns a promise of its result. The promise rejects with the error if the Effect
fails:

```ts
const batches = await Effect.runPromise(flight.query(`SELECT 1 AS one`))
```

`Effect.runPromise` only takes Effects whose `R` is `never`: TypeScript won't compile a program that's missing a
service.

## Write programs with `Effect.gen`

`Effect.gen` builds an Effect from a generator function, which reads like an `async` function. Where you'd `await` a
promise, you `yield*` an Effect, which gives you its result:

```ts
const program = Effect.gen(function* () {
  const flight = yield* ArrowFlight.ArrowFlight
  const batches = yield* flight.query(`SELECT count(*) AS blocks FROM "acme/eth_mainnet".blocks`)
  return batches.flatMap((batch) => batch.data)
})
```

If an Effect you `yield*` fails, the rest of the function doesn't run, and `program` fails with its error, like a
rejected `await`. `program`'s `E` is the union of the errors of everything it yields.

`.pipe(f, g)` applies functions in turn: `program.pipe(Effect.provide(AmpLayer))` is
`Effect.provide(program, AmpLayer)`. Effect's functions take their main argument last for this, so you can chain them.

## Services and layers

The SDK's features are services: `ArrowFlight` runs queries, `AdminApi` manages datasets, and so on. A service is an
object with methods, identified by a class, its tag. `yield* ArrowFlight.ArrowFlight` gets the `ArrowFlight` service,
and adds it to the program's `R`.

A layer builds services. `ArrowFlight.layer` builds `ArrowFlight`, but needs a `Transport`, the service that connects
to the query server. `layerTransportGrpc(…)` builds one. `Layer.provide` gives a layer what it needs from another, and
the result needs nothing:

```ts
// Layer<Transport>
const TransportLayer = layerTransportGrpc({ baseUrl: "http://localhost:1602" })
// Layer<ArrowFlight, ArrowFlightError, Transport> becomes Layer<ArrowFlight, ArrowFlightError>
const AmpLayer = ArrowFlight.layer.pipe(Layer.provide(TransportLayer))
```

Then `Effect.provide` gives the program its services, which removes them from its `R`:

```ts
await Effect.runPromise(program.pipe(Effect.provide(AmpLayer)))
```

The streaming services build on each other, so their layers stack: a `CdcStream` needs a `TransactionalStream`, which
needs a `ProtocolStream`, which needs `ArrowFlight`. Each page shows the stack its examples use. To give a program
services that don't depend on each other, merge their layers:

```ts
const AppLayer = Layer.mergeAll(AmpLayer, AdminLayer)
```

A layer builds its services once each time you provide it, and closes their resources, like connections, when the
program ends.

## Errors

The errors in an Effect's `E` are expected failures, which you can handle: a query with a typo, a dataset that doesn't
exist. Each has a `_tag` property naming its type, like `"RpcError"` or `"DatasetNotFoundError"`. `Effect.catchTag`
handles one:

```ts [errors.ts]
import { Domain } from "@edgeandnode/amp"
import { AdminApi, layer } from "@edgeandnode/amp/admin/service"
import { Effect, Layer } from "effect"
import { FetchHttpClient } from "effect/http"

const program = Effect.gen(function* () {
  const admin = yield* AdminApi
  const version = yield* admin.getDatasetVersion(
    Domain.DatasetNamespace.make("acme"),
    Domain.DatasetName.make("eth_testnet"),
    Domain.DatasetVersion.make("1.0.0"),
  )
  console.log(version.manifestHash)
}).pipe(
  Effect.catchTag("DatasetNotFoundError", (error) => Effect.sync(() => console.log(`${error.code}: ${error.message}`))),
)

const AdminLayer = layer({ url: "http://localhost:1610" }).pipe(Layer.provide(FetchHttpClient.layer))

await Effect.runPromise(program.pipe(Effect.provide(AdminLayer)))
```

:::terminal
```bash
node errors.ts
```

```
DATASET_NOT_FOUND: Dataset 'acme/eth_testnet' at revision '1.0.0' not found
```
:::

Once handled, `DatasetNotFoundError` is no longer in the program's `E`. To handle several, pass `Effect.catchTags` an
object with a handler for each tag. `Effect.catch` handles every error.

Failures that aren't in `E` are defects: bugs and broken assumptions, like a response the SDK can't read. Effect doesn't
expect you to handle them, but `Effect.catchDefect` can. [Errors](/sdks/typescript/errors) lists what each service fails
with, and which failures the SDK turns into defects.

## Streams

A `Stream<A, E, R>` is like an Effect that produces many values over time, like a streaming query's results. As with an
Effect, nothing happens until you run it:

* `Stream.runForEach(f)` runs the stream, and calls `f` on each value. `f` returns an Effect, which the stream waits for
  before it reads the next value. To run plain code, wrap it in `Effect.sync(() => …)`.
* `Stream.runCollect` runs a stream that ends, and returns all its values.
* `Stream.take(n)` ends a stream after its first `n` values.

```ts
yield* protocol.stream(sql).pipe(
    Stream.take(10),
    Stream.runForEach((message) => Effect.sync(() => console.log(message._tag))),
  )
```

## Schemas and branded types

A schema describes a type of data, and checks or converts values of it. The SDK uses Effect's `Schema` module to decode
what Amp sends, and you can use it to give query rows a type (see
[Querying](/sdks/typescript/querying#give-rows-a-type)).

Some of the SDK's types are branded: values of a base type, like `string`, that passed a check. A
`Domain.DatasetName` is a string of lowercase letters, digits, and underscores, for example. Create one with its
schema's `make`, which throws `Schema validation failed` when the value isn't valid:

```ts
const name = Domain.DatasetName.make("eth_mainnet")
```

To check a value without throwing, decode it: `Schema.decodeUnknownOption(Domain.DatasetName)(input)` returns an
`Option`, `Some` with the value or `None`. [Domain Types](/sdks/typescript/domain) lists each type and its check.

## Redacted values

Tokens are wrapped in `Redacted` values, which print as `<redacted>`, so they don't end up in logs.
`Redacted.make(token)` wraps one, and `Redacted.value(redacted)` reads it.

## Use the SDK from async code

You can call the SDK from `async` functions with three tools from Effect:

* A `ManagedRuntime` builds a layer's services once and keeps them, until you dispose of it. `runtime.runPromise` runs
  an Effect with them.
* Once you have a service, its methods return Effects and Streams that need nothing else: `Effect.runPromise` runs them.
* `Stream.toAsyncIterable` turns a Stream into something you read with `for await`.

```ts [async.ts]
import { ArrowFlight } from "@edgeandnode/amp"
import { layerTransportGrpc } from "@edgeandnode/amp/arrow-flight/node"
import { Effect, Layer, ManagedRuntime, Stream } from "effect"

const AmpLayer = ArrowFlight.layer.pipe(Layer.provide(layerTransportGrpc({ baseUrl: "http://localhost:1602" })))

// Builds the services once, and keeps them until you dispose of it
const runtime = ManagedRuntime.make(AmpLayer)
const flight = await runtime.runPromise(ArrowFlight.ArrowFlight)

// An Effect, run as a promise
const batches = await Effect.runPromise(flight.query(`SELECT count(*) AS blocks FROM "acme/eth_mainnet".blocks`))
console.log(batches.flatMap((batch) => batch.data))

// A failure rejects the promise with the error
try {
  await Effect.runPromise(flight.query(`SELECT * FROM "acme/eth_mainnet".missing`))
} catch (error) {
  if (error instanceof ArrowFlight.RpcError) console.error(error._tag, String(error.cause))
}

// A Stream, read with for await
const results = flight.streamQuery(`SELECT block_num FROM "acme/eth_mainnet".blocks ORDER BY block_num LIMIT 3`)
for await (const batch of Stream.toAsyncIterable(results)) {
  console.log(batch.data.length, "rows")
}

await runtime.dispose()
```

:::terminal
```bash
node async.ts
```

```
[ { blocks: '100' } ]
RpcError ConnectError: [invalid_argument] failed to infer SQL output schema | Caused by: Error during planning: table 'amp.acme/eth_mainnet.missing' not found -> Error during planning: table 'amp.acme/eth_mainnet.missing' not found
3 rows
0 rows
```
:::

The last batch has no rows: it marks the end of the results (see [Querying](/sdks/typescript/querying#results)).
Breaking out of a `for await` loop stops the stream, and closes its connection.

In a server, make the runtime when the server starts, and dispose of it when it stops. Disposing closes the services'
connections.