WorkPoolConfigWithItemSchema<T, E, R, A>A WorkPool config that carries an explicit itemSchema (typed items over the wire).
export type type WorkPoolConfigWithItemSchema<
T,
E,
R,
A = void
> = WorkPoolConfigBase<T> & {
readonly itemSchema: Schema.Codec<
T,
unknown,
never,
never
>
readonly effect: (
item: T,
ctx: EffectContext<T, QueueEnqueueErrors, R>
) => Effect.Effect<A, E, R>
readonly onFailure?: QueueOnFailure<T, E, R>
readonly refill?: QueueRefill<
T,
E,
QueueEnqueueErrors,
R,
A
>
readonly store?: QueueStoreWriter<T, E, A>
}
A
WorkPool
config that carries an explicit itemSchema (typed items over the wire).
WorkPoolConfigWithItemSchema<function (type parameter) T in type WorkPoolConfigWithItemSchema<T, E, R, A = void>T, function (type parameter) E in type WorkPoolConfigWithItemSchema<T, E, R, A = void>E, function (type parameter) R in type WorkPoolConfigWithItemSchema<T, E, R, A = void>R, function (type parameter) A in type WorkPoolConfigWithItemSchema<T, E, R, A = void>A = void> = interface WorkPoolConfigBase<T>Shared queue configuration fields (see
WorkPoolConfig
).
WorkPoolConfigBase<function (type parameter) T in type WorkPoolConfigWithItemSchema<T, E, R, A = void>T> & {
readonly itemSchema: Schema.Codec<T, unknown, never, never>(property) itemSchema: {
Encoded: E;
DecodingServices: RD;
EncodingServices: RE;
Rebuild: Codec<T, E, RD, RE>;
Type: T;
Iso: Iso;
ast: Ast;
annotate: (annotations: Schema.Annotations.Bottom<T, any>) => Schema.Codec<T, unknown, never, never>;
annotateKey: (annotations: Schema.Annotations.Key<T>) => Schema.Codec<T, unknown, never, never>;
check: (checks_0: Check<T>, ...checks: Array<Check<T>>) => Schema.Codec<T, unknown, never, never>;
rebuild: (ast: AST) => Schema.Codec<T, unknown, never, never>;
make: (input: unknown, options?: MakeOptions) => T;
makeOption: (input: unknown, options?: MakeOptions) => Option_.Option<T>;
makeEffect: (input: unknown, options?: MakeOptions) => Effect.Effect<T, SchemaError, never>;
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; <…;
}
import SchemaSchema.interface Codec<out T, out E = T, out RD = never, out RE = never>Namespace of type-level helpers for
Codec
.
A schema that tracks the decoded type T, the encoded type E, and the
Effect services required during decoding (RD) and encoding (RE).
Details
Use Codec<T, E, RD, RE> when you need to preserve full type information
about a schema — both what it decodes to and what it serializes from/to.
Most concrete schemas produced by this module implement Codec.
For APIs that only need one direction, prefer the narrower views:
Decoder
<T, RD> — decode-only
Encoder
<E, RE> — encode-only
Schema
<T> — type-only (no encoded representation)
Example (Accepting a codec that decodes to number from string)
import { Schema } from "effect"
declare function serialize<T>(codec: Schema.Codec<T, string>): string
serialize(Schema.NumberFromString) // ok — decodes number, encoded as string
Codec<function (type parameter) T in type WorkPoolConfigWithItemSchema<T, E, R, A = void>T, unknown, never, never>;
/**
* Daemon each item. The **success channel `A`** is driven by the tag's `success` wire schema
* (default `void`): with a `success` schema the worker must return `Effect<A, E, R>` and that
* value rides `Completed.success` / `store.completed`; without one it stays `Effect<void, E, R>`.
*/
readonly effect: (
item: T,
ctx: EffectContext<T, QueueEnqueueErrors, R>
) => Effect.Effect<A, E, R>
Daemon each item. The success channel A is driven by the tag's success wire schema
(default void): with a success schema the worker must return Effect<A, E, R> and that
value rides Completed.success / store.completed; without one it stays Effect<void, E, R>.
effect: (item: Titem: function (type parameter) T in type WorkPoolConfigWithItemSchema<T, E, R, A = void>T, ctx: EffectContext<T, QueueEnqueueErrors, R>(parameter) ctx: {
add: QueueEnqueue<T, EEnqueue, R>;
prioritize: QueueEnqueue<T, EEnqueue, R>;
defer: QueueEnqueue<T, EEnqueue, R>;
attempts: number;
enqueuedAt: number;
priority: Priority;
}
interface EffectContext<T, EEnqueue = never, R = never>Context passed to the effect callback during item processing.
Provides guarded enqueue operations for spawning derived/follow-up work.
Attempting to enqueue the same item (by reference or by key) logs a warning
and silently drops the item to prevent infinite processing loops.
For intentional re-processing, fail the worker effect (auto re-enqueue applies up to
WorkPoolConfigBase.attempts
) or re-inject via queue.enqueue off the events stream.
EffectContext<function (type parameter) T in type WorkPoolConfigWithItemSchema<T, E, R, A = void>T, type QueueEnqueueErrors =
| QueueItemValidationError
| QueueBatchValidationError
Enqueue validation errors when
WorkPoolConfigWithItemSchema.itemSchema
is set.
QueueEnqueueErrors, function (type parameter) R in type WorkPoolConfigWithItemSchema<T, E, R, A = void>R>) => 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<function (type parameter) A in type WorkPoolConfigWithItemSchema<T, E, R, A = void>A, function (type parameter) E in type WorkPoolConfigWithItemSchema<T, E, R, A = void>E, function (type parameter) R in type WorkPoolConfigWithItemSchema<T, E, R, A = void>R>;
/** Optional inline per-error disposition. See {@link QueueOnFailure}. */
readonly onFailure?: QueueOnFailure<T, E, R>Optional inline per-error disposition. See
QueueOnFailure
.
onFailure?: interface QueueOnFailure<T, E, R>Optional inline failure hook — the one legitimate control callback (a stream is after-the-fact
and can't decide disposition). Runs in the worker R on each failed item, returning a
QueueFailureDisposition
. Observation still belongs on the events stream; this is
control.
QueueOnFailure<function (type parameter) T in type WorkPoolConfigWithItemSchema<T, E, R, A = void>T, function (type parameter) E in type WorkPoolConfigWithItemSchema<T, E, R, A = void>E, function (type parameter) R in type WorkPoolConfigWithItemSchema<T, E, R, A = void>R>;
/** Optional self-refill from a source on start / drain. See {@link QueueRefill}. */
readonly refill?: | QueueRefill<T, E, QueueEnqueueErrors, R, A>
| undefined
Optional self-refill from a source on start / drain. See
QueueRefill
.
refill?: interface QueueRefill<T, E, EEnqueue, R, A = void>Self-refill — load work into the queue from a source (DB, …) on start and/or whenever it
drains empty (a self-feeding queue). load receives the queue handle and runs in the worker R,
best-effort (failures are logged, not fatal). This is the toolkit equivalent of the legacy
onStart / onDrained lifecycle hooks — a defining queue behavior (a pull source), distinct
from after-the-fact observation on the events stream.
QueueRefill<function (type parameter) T in type WorkPoolConfigWithItemSchema<T, E, R, A = void>T, function (type parameter) E in type WorkPoolConfigWithItemSchema<T, E, R, A = void>E, type QueueEnqueueErrors =
| QueueItemValidationError
| QueueBatchValidationError
Enqueue validation errors when
WorkPoolConfigWithItemSchema.itemSchema
is set.
QueueEnqueueErrors, function (type parameter) R in type WorkPoolConfigWithItemSchema<T, E, R, A = void>R, function (type parameter) A in type WorkPoolConfigWithItemSchema<T, E, R, A = void>A>;
/**
* Internal store recorder — the engine records each published event through its narrow, semantic
* writes ({@link QueueStoreWriter}) at the source (`publishEvent`) so no burst is dropped by a late
* `Stream.fromPubSub` subscription. Wired by `WorkPool.layer` from the baked-in store. @internal
*/
readonly store?: {
readonly event: {
readonly append: (row: {
readonly _tag: "Start";
readonly key: string;
} | {
readonly _tag: "Enqueued";
readonly entries: readonly {
readonly item: T;
readonly entryId: string;
readonly priority: "high" | "normal" | "low";
readonly attempts: number;
readonly timestamps: {
readonly enqueuedAt: DateTime.Utc;
readonly startedAt?: DateTime.Utc | undefined;
readonly completedAt?: DateTime.Utc | undefined;
readonly interruptedAt?: DateTime.Utc | undefined;
};
readonly key?: string | undefined;
readonly batchId?: string | undefined;
readonly releaseId?: string | undefined;
readonly sourceHyperlinkId?: string | undefined;
readonly attributes?: {
...;
} | undefined;
}[];
readonly priority: "high" | ... 1 more ... | "low";
readonly batchId?: string | undefined;
} | ... 13 more ... | readonly ({
readonly _tag: "Start";
readonly key: string;
} | ... 13 more ... | {
...;
})[]) => Effect.Effect<...>;
readonly read: (payload?: StoreReadPayload<...> | undefined) => Effect.Effect<...>;
};
... 8 more ...;
readonly "hyperlink-ts/Store/StoreEffects": {
...;
};
} | undefined
Internal store recorder — the engine records each published event through its narrow, semantic
writes (
QueueStoreWriter
) at the source (publishEvent) so no burst is dropped by a late
Stream.fromPubSub subscription. Wired by WorkPool.layer from the baked-in store.
store?: type QueueStoreWriter<T, _E = unknown, _A = void> = {
readonly event: {
readonly append: (row: {
readonly _tag: "Start";
readonly key: string;
} | {
readonly _tag: "Enqueued";
readonly entries: readonly {
readonly item: T;
readonly entryId: string;
readonly priority: "high" | "normal" | "low";
readonly attempts: number;
readonly timestamps: {
readonly enqueuedAt: DateTime.Utc;
readonly startedAt?: DateTime.Utc | undefined;
readonly completedAt?: DateTime.Utc | undefined;
readonly interruptedAt?: DateTime.Utc | undefined;
};
readonly key?: string | undefined;
readonly batchId?: string | undefined;
readonly releaseId?: string | undefined;
readonly sourceHyperlinkId?: string | undefined;
readonly attributes?: {
...;
} | undefined;
}[];
readonly priority: "high" | ... 1 more ... | "low";
readonly batchId?: string | undefined;
} | ... 13 more ... | readonly ({
readonly _tag: "Start";
readonly key: string;
} | ... 13 more ... | {
...;
})[]) => Effect.Effect<...>;
readonly read: (payload?: StoreReadPayload<...> | undefined) => Effect.Effect<...>;
};
... 8 more ...;
readonly "hyperlink-ts/Store/StoreEffects": {
...;
};
}
The engine-facing store recorder — Storage-free semantic writes the engine calls at the
publishEvent sites. Built by toolkit layers from
materializeEngineQueueStoreForTag
/
materializeEngineQueueStoreForItem
(Store.effects +
Store.catchWriteErrors
+
Store.provideContext
). Each narrow
write funnels to the shared event.append; record is the base append alias for queue-level
facts without a narrow write.
QueueStoreWriter<function (type parameter) T in type WorkPoolConfigWithItemSchema<T, E, R, A = void>T, function (type parameter) E in type WorkPoolConfigWithItemSchema<T, E, R, A = void>E, function (type parameter) A in type WorkPoolConfigWithItemSchema<T, E, R, A = void>A>;
};