<S extends StoreClassWithShapes, Row extends Schema.Schema<unknown>>(
store: S,
select: (shapes: ShapeRefs<ShapesOfStore<S>>) => ShapeRef<Row>
): Effect.Effect<
Stream.Stream<SchemaDecoded<Row>, StoreJournalDecodeError>,
never,
Storage | Scope.Scope
>
<S extends StoreClassWithShapes>(store: S): Effect.Effect<
Stream.Stream<AllShapeRows<ShapesOfStore<S>>, StoreJournalDecodeError>,
never,
Storage | Scope.Scope
>
(scope: string | StoreScopeTag): Effect.Effect<
Stream.Stream<StoreChangeEvent, StoreJournalDecodeError>,
StoreScopeNotRegistered,
Storage | Scope.Scope
>Stream store changes. Three forms:
changes(scope)— coarse firehose of StoreChangeEvents for a scope (string or tag).changes(store)— decoded rows of every shape on the store (discriminated union).changes(store, select)— decoded rows of the one shape the selector navigates to, e.g.changes(store, (shapes) => shapes.sensors.temperature).
Requires a Service.layer / store.layer that installed the scope bridge.
export function function changes<S extends StoreClassWithShapes, Row extends Schema.Schema<unknown>>(store: S, select: (shapes: ShapeRefs<ShapesOfStore<S>>) => ShapeRef<Row>): Effect.Effect<Stream.Stream<SchemaDecoded<Row>, StoreJournalDecodeError>, never, Storage | Scope.Scope> (+2 overloads)Stream store changes. Three forms:
changes(scope) — coarse firehose of
StoreChangeEvent
s for a scope (string or tag).changes(store) — decoded rows of every shape on the store (discriminated union).changes(store, select) — decoded rows of the one shape the selector navigates to, e.g.
changes(store, (shapes) => shapes.sensors.temperature).
Requires a
Service.layer
/
store.layer
that installed the scope bridge.
changes<function (type parameter) S in changes<S extends StoreClassWithShapes, Row extends Schema.Schema<unknown>>(store: S, select: (shapes: ShapeRefs<ShapesOfStore<S>>) => ShapeRef<Row>): Effect.Effect<Stream.Stream<SchemaDecoded<Row>, StoreJournalDecodeError>, never, Storage | Scope.Scope>S extends import StoreClassWithShapesStoreClassWithShapes, function (type parameter) Row in changes<S extends StoreClassWithShapes, Row extends Schema.Schema<unknown>>(store: S, select: (shapes: ShapeRefs<ShapesOfStore<S>>) => ShapeRef<Row>): Effect.Effect<Stream.Stream<SchemaDecoded<Row>, StoreJournalDecodeError>, never, Storage | Scope.Scope>Row extends import SchemaSchema.interface Schema<out T>Namespace of type-level helpers for
Schema
.
A typed view of a schema that tracks only the decoded (output) type T.
Details
Use Schema<T> as a constraint when you want to accept "any schema that
decodes to T" and do not need to know or constrain the encoded
representation, required services, or any other type parameters.
This is a structural interface — concrete schema values are produced by the
constructors in this module (e.g.
Struct
,
String
,
Number
).
When you also need the encoded type or service requirements, use
Codec
.
Example (Accepting any schema decoding to string)
import { Schema } from "effect"
declare function print(schema: Schema.Schema<string>): void
print(Schema.String) // ok
print(Schema.NonEmptyString) // ok
Schema<unknown>>(
store: S extends StoreClassWithShapesstore: function (type parameter) S in changes<S extends StoreClassWithShapes, Row extends Schema.Schema<unknown>>(store: S, select: (shapes: ShapeRefs<ShapesOfStore<S>>) => ShapeRef<Row>): Effect.Effect<Stream.Stream<SchemaDecoded<Row>, StoreJournalDecodeError>, never, Storage | Scope.Scope>S,
select: (
shapes: ShapeRefs<ShapesOfStore<S>>
) => ShapeRef<Row>
shapes: ShapeRefs<ShapesOfStore<S>>shapes: import ShapeRefsShapeRefs<import ShapesOfStoreShapesOfStore<function (type parameter) S in changes<S extends StoreClassWithShapes, Row extends Schema.Schema<unknown>>(store: S, select: (shapes: ShapeRefs<ShapesOfStore<S>>) => ShapeRef<Row>): Effect.Effect<Stream.Stream<SchemaDecoded<Row>, StoreJournalDecodeError>, never, Storage | Scope.Scope>S>>) => import ShapeRefShapeRef<function (type parameter) Row in changes<S extends StoreClassWithShapes, Row extends Schema.Schema<unknown>>(store: S, select: (shapes: ShapeRefs<ShapesOfStore<S>>) => ShapeRef<Row>): Effect.Effect<Stream.Stream<SchemaDecoded<Row>, StoreJournalDecodeError>, never, Storage | Scope.Scope>Row>,
): import EffectEffect.interface Effect<out A, out E = never, out R = never>The Effect interface defines a value that lazily describes a workflow or
job. The workflow requires some context R, and may fail with an error of
type E, or succeed with a value of type A.
When to use
Use when you need to represent a lazy, composable workflow that can require
services, fail with a typed error, or succeed with a typed value.
Details
Effect values model resourceful interaction with the outside world,
including synchronous, asynchronous, concurrent, and parallel interaction.
They use a fiber-based concurrency model, with built-in support for
scheduling, fine-grained interruption, structured concurrency, and high
scalability.
To run an Effect value, you need a Runtime, which is a type that is
capable of executing Effect values.
Effect<
import StreamStream.interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<import SchemaDecodedSchemaDecoded<function (type parameter) Row in changes<S extends StoreClassWithShapes, Row extends Schema.Schema<unknown>>(store: S, select: (shapes: ShapeRefs<ShapesOfStore<S>>) => ShapeRef<Row>): Effect.Effect<Stream.Stream<SchemaDecoded<Row>, StoreJournalDecodeError>, never, Storage | Scope.Scope>Row>, import StoreJournalDecodeErrorStoreJournalDecodeError>,
never,
class Storageclass Storage {
Service: Service;
key: Identifier;
}
Scope bridge every store handle resolves through — provided by an app
Service
layer or
the soft-default
layerDefaultMemory
(via
withDefaultStorage
). Engines resolve
handles via
withDefault
/
withStorage
(preferred) or bridge.at.
Storage | import ScopeScope.Scope
>;
export function function changes<S extends StoreClassWithShapes>(store: S): Effect.Effect<Stream.Stream<AllShapeRows<ShapesOfStore<S>>, StoreJournalDecodeError>, never, Storage | Scope.Scope> (+2 overloads)Stream store changes. Three forms:
changes(scope) — coarse firehose of
StoreChangeEvent
s for a scope (string or tag).changes(store) — decoded rows of every shape on the store (discriminated union).changes(store, select) — decoded rows of the one shape the selector navigates to, e.g.
changes(store, (shapes) => shapes.sensors.temperature).
Requires a
Service.layer
/
store.layer
that installed the scope bridge.
changes<function (type parameter) S in changes<S extends StoreClassWithShapes>(store: S): Effect.Effect<Stream.Stream<AllShapeRows<ShapesOfStore<S>>, StoreJournalDecodeError>, never, Storage | Scope.Scope>S extends import StoreClassWithShapesStoreClassWithShapes>(
store: S extends StoreClassWithShapesstore: function (type parameter) S in changes<S extends StoreClassWithShapes>(store: S): Effect.Effect<Stream.Stream<AllShapeRows<ShapesOfStore<S>>, StoreJournalDecodeError>, never, Storage | Scope.Scope>S,
): import EffectEffect.interface Effect<out A, out E = never, out R = never>The Effect interface defines a value that lazily describes a workflow or
job. The workflow requires some context R, and may fail with an error of
type E, or succeed with a value of type A.
When to use
Use when you need to represent a lazy, composable workflow that can require
services, fail with a typed error, or succeed with a typed value.
Details
Effect values model resourceful interaction with the outside world,
including synchronous, asynchronous, concurrent, and parallel interaction.
They use a fiber-based concurrency model, with built-in support for
scheduling, fine-grained interruption, structured concurrency, and high
scalability.
To run an Effect value, you need a Runtime, which is a type that is
capable of executing Effect values.
Effect<
import StreamStream.interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<import AllShapeRowsAllShapeRows<import ShapesOfStoreShapesOfStore<function (type parameter) S in changes<S extends StoreClassWithShapes>(store: S): Effect.Effect<Stream.Stream<AllShapeRows<ShapesOfStore<S>>, StoreJournalDecodeError>, never, Storage | Scope.Scope>S>>, import StoreJournalDecodeErrorStoreJournalDecodeError>,
never,
class Storageclass Storage {
Service: Service;
key: Identifier;
}
Scope bridge every store handle resolves through — provided by an app
Service
layer or
the soft-default
layerDefaultMemory
(via
withDefaultStorage
). Engines resolve
handles via
withDefault
/
withStorage
(preferred) or bridge.at.
Storage | import ScopeScope.Scope
>;
export function function changes(scope: string | StoreScopeTag): Effect.Effect<Stream.Stream<StoreChangeEvent, StoreJournalDecodeError>, StoreScopeNotRegistered, Storage | Scope.Scope> (+2 overloads)Stream store changes. Three forms:
changes(scope) — coarse firehose of
StoreChangeEvent
s for a scope (string or tag).changes(store) — decoded rows of every shape on the store (discriminated union).changes(store, select) — decoded rows of the one shape the selector navigates to, e.g.
changes(store, (shapes) => shapes.sensors.temperature).
Requires a
Service.layer
/
store.layer
that installed the scope bridge.
changes(
scope: anyscope: string | import StoreScopeTagStoreScopeTag,
): import EffectEffect.interface Effect<out A, out E = never, out R = never>The Effect interface defines a value that lazily describes a workflow or
job. The workflow requires some context R, and may fail with an error of
type E, or succeed with a value of type A.
When to use
Use when you need to represent a lazy, composable workflow that can require
services, fail with a typed error, or succeed with a typed value.
Details
Effect values model resourceful interaction with the outside world,
including synchronous, asynchronous, concurrent, and parallel interaction.
They use a fiber-based concurrency model, with built-in support for
scheduling, fine-grained interruption, structured concurrency, and high
scalability.
To run an Effect value, you need a Runtime, which is a type that is
capable of executing Effect values.
Effect<
import StreamStream.interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<import StoreChangeEventStoreChangeEvent, import StoreJournalDecodeErrorStoreJournalDecodeError>,
import StoreScopeNotRegisteredStoreScopeNotRegistered,
class Storageclass Storage {
Service: Service;
key: Identifier;
}
Scope bridge every store handle resolves through — provided by an app
Service
layer or
the soft-default
layerDefaultMemory
(via
withDefaultStorage
). Engines resolve
handles via
withDefault
/
withStorage
(preferred) or bridge.at.
Storage | import ScopeScope.Scope
>;
export function function changes<S extends StoreClassWithShapes, Row extends Schema.Schema<unknown>>(store: S, select: (shapes: ShapeRefs<ShapesOfStore<S>>) => ShapeRef<Row>): Effect.Effect<Stream.Stream<SchemaDecoded<Row>, StoreJournalDecodeError>, never, Storage | Scope.Scope> (+2 overloads)Stream store changes. Three forms:
changes(scope) — coarse firehose of
StoreChangeEvent
s for a scope (string or tag).changes(store) — decoded rows of every shape on the store (discriminated union).changes(store, select) — decoded rows of the one shape the selector navigates to, e.g.
changes(store, (shapes) => shapes.sensors.temperature).
Requires a
Service.layer
/
store.layer
that installed the scope bridge.
changes(
storeOrScope: anystoreOrScope: string | import StoreScopeTagStoreScopeTag | import StoreClassWithShapesStoreClassWithShapes,
select: | ((
shapes: ShapeRefs<StoreShapes>
) => ShapeRef<Schema.Schema<unknown>>)
| undefined
shapes: ShapeRefs<StoreShapes>shapes: import ShapeRefsShapeRefs<import StoreShapesStoreShapes>) => import ShapeRefShapeRef<import SchemaSchema.interface Schema<out T>Namespace of type-level helpers for
Schema
.
A typed view of a schema that tracks only the decoded (output) type T.
Details
Use Schema<T> as a constraint when you want to accept "any schema that
decodes to T" and do not need to know or constrain the encoded
representation, required services, or any other type parameters.
This is a structural interface — concrete schema values are produced by the
constructors in this module (e.g.
Struct
,
String
,
Number
).
When you also need the encoded type or service requirements, use
Codec
.
Example (Accepting any schema decoding to string)
import { Schema } from "effect"
declare function print(schema: Schema.Schema<string>): void
print(Schema.String) // ok
print(Schema.NonEmptyString) // ok
Schema<unknown>>,
): import EffectEffect.interface Effect<out A, out E = never, out R = never>The Effect interface defines a value that lazily describes a workflow or
job. The workflow requires some context R, and may fail with an error of
type E, or succeed with a value of type A.
When to use
Use when you need to represent a lazy, composable workflow that can require
services, fail with a typed error, or succeed with a typed value.
Details
Effect values model resourceful interaction with the outside world,
including synchronous, asynchronous, concurrent, and parallel interaction.
They use a fiber-based concurrency model, with built-in support for
scheduling, fine-grained interruption, structured concurrency, and high
scalability.
To run an Effect value, you need a Runtime, which is a type that is
capable of executing Effect values.
Effect<
import StreamStream.interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<unknown, import StoreJournalDecodeErrorStoreJournalDecodeError, unknown>,
import StoreScopeNotRegisteredStoreScopeNotRegistered,
class Storageclass Storage {
Service: Service;
key: Identifier;
}
Scope bridge every store handle resolves through — provided by an app
Service
layer or
the soft-default
layerDefaultMemory
(via
withDefaultStorage
). Engines resolve
handles via
withDefault
/
withStorage
(preferred) or bridge.at.
Storage | import ScopeScope.Scope
> {
if (const isStoreClassWithShapes: (
value: unknown
) => value is StoreClassWithShapes
True for a single-scope store class carrying its contract — the typed
changes
forms.
isStoreClassWithShapes(storeOrScope: anystoreOrScope)) {
const const store: StoreClassWithShapesstore = storeOrScope: StoreClassWithShapesstoreOrScope;
if (select: | ((
shapes: ShapeRefs<StoreShapes>
) => ShapeRef<Schema.Schema<unknown>>)
| undefined
var undefinedundefined) {
const const ref: anyref = import resolveShapeRefresolveShapeRef(select: (
shapes: ShapeRefs<StoreShapes>
) => ShapeRef<Schema.Schema<unknown>>
import makeShapeRefsmakeShapeRefs(const store: StoreClassWithShapesstore.contract.shapes)));
return const storeChangesStream: (
store: StoreClassWithShapes
) => Effect.Effect<
Stream.Stream<
StoreChangeEvent,
StoreJournalDecodeError
>,
never,
Storage | Scope.Scope
>
Resolve the store's scope changes stream, dying if the scope is unregistered (wiring error).
storeChangesStream(const store: StoreClassWithShapesstore).Pipeable.pipe<Effect.Effect<Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>, never, Storage | Scope.Scope>, Effect.Effect<Stream.Stream<unknown, any, never>, never, Storage | Scope.Scope>>(this: Effect.Effect<Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>, never, Storage | Scope.Scope>, ab: (_: Effect.Effect<Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>, never, Storage | Scope.Scope>) => Effect.Effect<Stream.Stream<unknown, any, never>, never, Storage | Scope.Scope>): Effect.Effect<...> (+21 overloads)pipe(
import EffectEffect.const map: <Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>, Stream.Stream<unknown, any, never>>(f: (a: Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>) => Stream.Stream<unknown, any, never>) => <E, R>(self: Effect.Effect<Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>, E, R>) => Effect.Effect<Stream.Stream<unknown, any, never>, E, R> (+1 overload)Transforms the value inside an effect by applying a function to it.
When to use
Use to transform an effect's success value with a function that returns a
plain value, producing a new effect without changing the original effect's
typed error or context requirements.
Details
map takes a function and applies it to the value contained within an
effect, creating a new effect with the transformed value.
It's important to note that effects are immutable, meaning that the original
effect is not modified. Instead, a new effect is returned with the updated
value.
Example (Choosing map syntax variants)
import { Effect, pipe } from "effect"
const myEffect = Effect.succeed(1)
const transformation = (n: number) => n + 1
const mappedWithPipe = pipe(myEffect, Effect.map(transformation))
const mappedWithDataFirst = Effect.map(myEffect, transformation)
const mappedWithMethod = myEffect.pipe(Effect.map(transformation))
Example (Adding a service charge)
import { Effect, pipe } from "effect"
const addServiceCharge = (amount: number) => amount + 1
const fetchTransactionAmount = Effect.promise(() => Promise.resolve(100))
const finalAmount = pipe(
fetchTransactionAmount,
Effect.map(addServiceCharge)
)
Effect.runPromise(finalAmount).then(console.log)
// Output: 101
map((stream: Stream.Stream<
StoreChangeEvent,
StoreJournalDecodeError,
never
>
(parameter) stream: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
stream: Stream.Stream<
StoreChangeEvent,
StoreJournalDecodeError,
never
>
(parameter) stream: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
Pipeable.pipe<Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>, Stream.Stream<any, StoreJournalDecodeError, never>, Stream.Stream<unknown, any, never>>(this: Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>, ab: (_: Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>) => Stream.Stream<any, StoreJournalDecodeError, never>, bc: (_: Stream.Stream<any, StoreJournalDecodeError, never>) => Stream.Stream<unknown, any, never>): Stream.Stream<unknown, any, never> (+21 overloads)pipe(
import StreamStream.const filter: <any>(predicate: Predicate.Predicate<any>) => <E, R>(self: Stream.Stream<any, E, R>) => Stream.Stream<any, E, R> (+3 overloads)Filters a stream to the elements that satisfy a predicate.
Example (Filtering stream values)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
const stream = Stream.make(1, 2, 3, 4).pipe(
Stream.filter((n) => n % 2 === 0)
)
const values = yield* Stream.runCollect(stream)
yield* Console.log(values)
})
Effect.runPromise(program)
// Output: [ 2, 4 ]
filter((event: anyevent) => event: anyevent.method === const ref: anyref.shapeKey),
import StreamStream.const mapEffect: <any, unknown, StoreJournalDecodeError, never>(f: (a: any, i: number) => Effect.Effect<unknown, StoreJournalDecodeError, never>, options?: {
readonly concurrency?: number | "unbounded" | undefined;
readonly unordered?: boolean | undefined;
} | undefined) => <E, R>(self: Stream.Stream<any, E, R>) => Stream.Stream<unknown, any, R> (+1 overload)
Maps over elements of the stream with the specified effectful function.
When to use
Use when each stream element transformation needs an Effect, service
dependency, failure channel, or configured concurrency.
Example (Effectfully mapping stream values)
import { Console, Effect, Stream } from "effect"
const stream = Stream.make(1, 2, 3)
const mappedStream = stream.pipe(
Stream.mapEffect((n) =>
Effect.gen(function*() {
yield* Console.log(`Processing: ${n}`)
return n * 2
})
)
)
const program = Effect.gen(function*() {
const result = yield* Stream.runCollect(mappedStream)
yield* Console.log(result)
})
Effect.runPromise(program)
// Output:
// Processing: 1
// Processing: 2
// Processing: 3
// [2, 4, 6]
mapEffect((event: anyevent) => const decodeChangeRow: (
row: Schema.Schema<unknown>,
payload: unknown
) => Effect.Effect<
unknown,
StoreJournalDecodeError,
never
>
Decode a change-event payload against a shape's row schema, re-tagging failures. A bare
Schema.Schema<unknown> carries DecodingServices: unknown, so the decode requirement is
unknown here; the public
changes
overloads pin the stream requirement to never
independently, so callers never see it.
decodeChangeRow(const ref: anyref.row, event: anyevent.payload)),
),
),
) as any;
}
const const rowByKey: anyrowByKey = import shapeRowsByKeyshapeRowsByKey(const store: StoreClassWithShapesstore.contract.shapes);
return const storeChangesStream: (
store: StoreClassWithShapes
) => Effect.Effect<
Stream.Stream<
StoreChangeEvent,
StoreJournalDecodeError
>,
never,
Storage | Scope.Scope
>
Resolve the store's scope changes stream, dying if the scope is unregistered (wiring error).
storeChangesStream(const store: StoreClassWithShapesstore).Pipeable.pipe<Effect.Effect<Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>, never, Storage | Scope.Scope>, Effect.Effect<Stream.Stream<unknown, any, never>, never, Storage | Scope.Scope>>(this: Effect.Effect<Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>, never, Storage | Scope.Scope>, ab: (_: Effect.Effect<Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>, never, Storage | Scope.Scope>) => Effect.Effect<Stream.Stream<unknown, any, never>, never, Storage | Scope.Scope>): Effect.Effect<...> (+21 overloads)pipe(
import EffectEffect.const map: <Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>, Stream.Stream<unknown, any, never>>(f: (a: Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>) => Stream.Stream<unknown, any, never>) => <E, R>(self: Effect.Effect<Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>, E, R>) => Effect.Effect<Stream.Stream<unknown, any, never>, E, R> (+1 overload)Transforms the value inside an effect by applying a function to it.
When to use
Use to transform an effect's success value with a function that returns a
plain value, producing a new effect without changing the original effect's
typed error or context requirements.
Details
map takes a function and applies it to the value contained within an
effect, creating a new effect with the transformed value.
It's important to note that effects are immutable, meaning that the original
effect is not modified. Instead, a new effect is returned with the updated
value.
Example (Choosing map syntax variants)
import { Effect, pipe } from "effect"
const myEffect = Effect.succeed(1)
const transformation = (n: number) => n + 1
const mappedWithPipe = pipe(myEffect, Effect.map(transformation))
const mappedWithDataFirst = Effect.map(myEffect, transformation)
const mappedWithMethod = myEffect.pipe(Effect.map(transformation))
Example (Adding a service charge)
import { Effect, pipe } from "effect"
const addServiceCharge = (amount: number) => amount + 1
const fetchTransactionAmount = Effect.promise(() => Promise.resolve(100))
const finalAmount = pipe(
fetchTransactionAmount,
Effect.map(addServiceCharge)
)
Effect.runPromise(finalAmount).then(console.log)
// Output: 101
map((stream: Stream.Stream<
StoreChangeEvent,
StoreJournalDecodeError,
never
>
(parameter) stream: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
stream: Stream.Stream<
StoreChangeEvent,
StoreJournalDecodeError,
never
>
(parameter) stream: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
Pipeable.pipe<Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>, Stream.Stream<unknown, any, never>>(this: Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>, ab: (_: Stream.Stream<StoreChangeEvent, StoreJournalDecodeError, never>) => Stream.Stream<unknown, any, never>): Stream.Stream<unknown, any, never> (+21 overloads)pipe(
import StreamStream.const mapEffect: <any, unknown, StoreJournalDecodeError, never>(f: (a: any, i: number) => Effect.Effect<unknown, StoreJournalDecodeError, never>, options?: {
readonly concurrency?: number | "unbounded" | undefined;
readonly unordered?: boolean | undefined;
} | undefined) => <E, R>(self: Stream.Stream<any, E, R>) => Stream.Stream<unknown, any, R> (+1 overload)
Maps over elements of the stream with the specified effectful function.
When to use
Use when each stream element transformation needs an Effect, service
dependency, failure channel, or configured concurrency.
Example (Effectfully mapping stream values)
import { Console, Effect, Stream } from "effect"
const stream = Stream.make(1, 2, 3)
const mappedStream = stream.pipe(
Stream.mapEffect((n) =>
Effect.gen(function*() {
yield* Console.log(`Processing: ${n}`)
return n * 2
})
)
)
const program = Effect.gen(function*() {
const result = yield* Stream.runCollect(mappedStream)
yield* Console.log(result)
})
Effect.runPromise(program)
// Output:
// Processing: 1
// Processing: 2
// Processing: 3
// [2, 4, 6]
mapEffect((event: anyevent) => {
const const row: anyrow = const rowByKey: anyrowByKey.get(event: anyevent.method);
return const row: anyrow === var undefinedundefined
? import EffectEffect.const die: (
defect: unknown
) => Effect.Effect<never>
Creates an effect that terminates a fiber with a specified error.
When to use
Use when you need an Effect to report an unrecoverable defect instead of a
typed error.
Details
The die function is used to signal a defect, which represents a critical
and unexpected error in the code. When invoked, it produces an effect that
does not handle the error and instead terminates the fiber.
The error channel of the resulting effect is of type never, indicating that
it cannot recover from this failure.
Example (Failing on division by zero)
import { Effect } from "effect"
const divide = (a: number, b: number) =>
b === 0
? Effect.die(new Error("Cannot divide by zero"))
: Effect.succeed(a / b)
// ┌─── Effect<number, never, never>
// ▼
const program = divide(1, 0)
Effect.runPromise(program).catch(console.error)
// Output:
// (FiberFailure) Error: Cannot divide by zero
// ...stack trace...
die(
new import StoreJournalDecodeErrorStoreJournalDecodeError({
cause: stringcause: `no shape row schema registered for change method "${event: anyevent.method}"`,
detail: stringdetail: "Store.changes(store)",
}),
)
: const decodeChangeRow: (
row: Schema.Schema<unknown>,
payload: unknown
) => Effect.Effect<
unknown,
StoreJournalDecodeError,
never
>
Decode a change-event payload against a shape's row schema, re-tagging failures. A bare
Schema.Schema<unknown> carries DecodingServices: unknown, so the decode requirement is
unknown here; the public
changes
overloads pin the stream requirement to never
independently, so callers never see it.
decodeChangeRow(const row: anyrow, event: anyevent.payload);
}),
),
),
) as any;
}
const const key: anykey = typeof storeOrScope: anystoreOrScope === "string" ? storeOrScope: stringstoreOrScope : storeOrScope: anystoreOrScope.key;
return import EffectEffect.const flatMap: <StorageApi, never, Storage, Stream.Stream<unknown, StoreJournalDecodeError, unknown>, StoreScopeNotRegistered, Storage | Scope.Scope>(self: Effect.Effect<StorageApi, never, Storage>, f: (a: StorageApi) => Effect.Effect<Stream.Stream<unknown, StoreJournalDecodeError, unknown>, StoreScopeNotRegistered, Storage | Scope.Scope>) => Effect.Effect<Stream.Stream<unknown, StoreJournalDecodeError, unknown>, any, Storage | Scope.Scope> (+1 overload)Chains effects to produce new Effect instances, useful for combining
operations that depend on previous results.
When to use
Use when you need to chain multiple effects, ensuring that each
step produces a new Effect while flattening any nested effects that may
occur.
Details
flatMap lets you sequence effects so that the result of one effect can be
used in the next step. It is similar to flatMap used with arrays but works
specifically with Effect instances, allowing you to avoid deeply nested
effect structures.
Since effects are immutable, flatMap always returns a new effect instead of
changing the original one.
Example (Choosing flatMap syntax variants)
import { Effect, pipe } from "effect"
const myEffect = Effect.succeed(1)
const transformation = (n: number) => Effect.succeed(n + 1)
const flatMappedWithPipe = pipe(myEffect, Effect.flatMap(transformation))
const flatMappedWithDataFirst = Effect.flatMap(myEffect, transformation)
const flatMappedWithMethod = myEffect.pipe(Effect.flatMap(transformation))
Example (Sequencing dependent effects)
import { Data, Effect, pipe } from "effect"
class DiscountRateError extends Data.TaggedError("DiscountRateError")<{}> {}
// Function to apply a discount safely to a transaction amount
const applyDiscount = (
total: number,
discountRate: number
): Effect.Effect<number, DiscountRateError> =>
discountRate === 0
? Effect.fail(new DiscountRateError())
: Effect.succeed(total - (total * discountRate) / 100)
// Simulated asynchronous task to fetch a transaction amount from database
const fetchTransactionAmount = Effect.promise(() => Promise.resolve(100))
// Chaining the fetch and discount application using `flatMap`
const finalAmount = pipe(
fetchTransactionAmount,
Effect.flatMap((amount) => applyDiscount(amount, 5))
)
Effect.runPromise(finalAmount).then(console.log)
// Output: 95
flatMap(class Storageclass Storage {
key: Identifier;
of: (this: void, self: StorageApi) => StorageApi;
context: (self: StorageApi) => Context.Context<Storage>;
use: (f: (service: StorageApi) => Effect.Effect<A, E, R>) => Effect.Effect<A, E, Storage | R>;
useSync: (f: (service: StorageApi) => A) => Effect.Effect<A, never, Storage>;
Identifier: Identifier;
Service: Shape;
stack: string | undefined;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
toString: () => string;
toJSON: () => unknown;
}
Scope bridge every store handle resolves through — provided by an app
Service
layer or
the soft-default
layerDefaultMemory
(via
withDefaultStorage
). Engines resolve
handles via
withDefault
/
withStorage
(preferred) or bridge.at.
Storage, (bridge: StorageApibridge) => bridge: StorageApibridge.changes(const key: anykey));
}