Thilawyn/effect-view
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor Query/Mutation
Lint / lint (push) Successful in 1m3s
Details

Browse Source
This commit is contained in:
Julien Valverdé committed 2026-10-09 02:14:20 +02:00
1 parent 4d90ea2c0e
commit 3875bc4f69
13 files changed
+751 -470
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/docs/docs/mutation.md
+25 -22
View File
@@ -18,10 +18,10 @@ The familiar concepts are present, but the operation itself is an Effect:
| Mutation result | `mutation.state`, a `View<AsyncResult<A, E>>` |
| `isPending` | `result.waiting` |
| `mutateAsync` | `mutation.mutate(key)` |
| Start without awaiting | `mutation.mutateView(key)` |
| Start without awaiting | `mutation.mutateStream(key)` |
Unlike a Query, a Mutation has no cache, reactive key, or automatic execution.
It runs only when `mutate` or `mutateView` is called. This makes it suitable for
It runs only when `mutate` or `mutateStream` is called. This makes it suitable for
commands, while Query remains the model for cached server state.
## Create a mutation
@@ -62,7 +62,7 @@ const InviteButtonView = Component.make("InviteButton")(
<button
disabled={result.waiting}
onClick={() =>
runSync(mutation.mutateView({
runSync(mutation.mutateStream({
email: props.email,
role: "member",
}))
@@ -132,7 +132,7 @@ during retries or repeated submissions.
A failure contains an Effect `Cause<E>`, not only `E`. Typed failures, defects,
and interruption information remain available for logging or presentation.
## Choose mutate or mutateView
## Choose mutate or mutateStream
Both methods start the same mutation effect. They differ in what the caller
waits for:
@@ -140,7 +140,7 @@ waits for:
| Method | Return value | Best suited to |
| --- | --- | --- |
| `mutate(key)` | The final `Success` or `Failure` | Effect workflows that need the outcome. |
| `mutateView(key)` | A live `View<AsyncResult<A, E>>` | UI callbacks that should return after starting the work. |
| `mutateStream(key)` | A `Stream` of this call's successive states, ending once it settles | UI callbacks that should return after starting the work. |
Use `mutate` with an asynchronous component runner when later logic depends on
the final result:
@@ -164,20 +164,22 @@ The mutation Effect does not fail with `E`. It captures the operation's `Exit`
and returns a final `AsyncResult.Success` or `AsyncResult.Failure`. Inspect or
match that value when control flow depends on the outcome.
Use `mutateView` when the UI only needs to start the operation and react to its
Use `mutateStream` when the UI only needs to start the operation and react to its
state:
```tsx
const runSync = yield* Component.useRunSync()
const startSave = () => {
const state = runSync(mutation.mutateView(input))
// `state` is a View for this specific call.
const states = runSync(mutation.mutateStream(input))
// `states` streams this specific call's states, and ends once it settles.
}
```
`mutateView` starts the scoped work and returns immediately with a per-call
View. The shared `mutation.state` is also updated as that call progresses.
`mutateStream` starts the scoped work and returns immediately with a per-call
`Stream`: it emits the call's current state, then each change, and ends once
the call settles or is interrupted. The shared `mutation.state` is also updated
as that call progresses. Consume the stream in a component with `Stream.use`.
## Track the latest call
@@ -188,7 +190,7 @@ In addition to `state`, a mutation exposes reactive metadata:
| `state` | The latest mutation state published to the shared View. |
| `latestKey` | The most recently supplied input as an `Option<K>`. |
| `latestFinalResult` | The latest completed success or failure as an `Option`. |
| `fiber` | The most recently started mutation fiber as an `Option`. |
| `operation` | The most recently started call as an `Option<Operation>`: its key, state, fiber, and `interrupt`. |
Subscribe to any of them with `View.useAll`:
@@ -207,19 +209,20 @@ any other value accepted by the mutation function.
## Concurrent mutations
Starting a mutation does not automatically interrupt an earlier mutation.
Calls may overlap, and each call has its own state View. This is useful for
Calls may overlap, and each call has its own state stream. This is useful for
independent operations such as uploading several files.
When calls overlap, `mutation.state` reflects updates published by all calls;
the last update to arrive wins. Use the View returned by `mutateView` when each
concurrent operation needs its own progress indicator:
When calls overlap, `mutation.state` follows only the most recently started
call: earlier calls still run to completion, but no longer update it. Use the
stream returned by `mutateStream` when each concurrent operation needs its own
progress indicator:
```tsx
const upload = (file: File) =>
Effect.gen(function* () {
const uploadState = yield* mutation.mutateView(file)
// Store or pass `uploadState` to the row rendering this file.
return uploadState
const uploadStates = yield* mutation.mutateStream(file)
// Pass `uploadStates` to the row rendering this file, e.g. via `Stream.use`.
return uploadStates
})
```
@@ -240,7 +243,7 @@ const saveAndRefresh = (input: UpdatePostInput) =>
if (AsyncResult.isSuccess(result)) {
yield* posts.invalidateCacheEntry(["post", result.value.id] as const)
yield* posts.refreshView
yield* posts.refresh
}
return result
@@ -253,7 +256,7 @@ also include tracing, transactions, retries, notifications, or parallel cache
updates without introducing a separate callback API.
Remember that Query invalidation removes cached data but does not refetch by
itself. Follow it with `refreshView` when the current screen should update
itself. Follow it with `refresh` or `refreshStream` when the current screen should update
immediately.
## The Effect touch
@@ -266,8 +269,8 @@ execution model:
- Failures are represented as `Cause<E>`, including defects and interruption.
- Fibers are scoped, so component unmounting cleans up in-flight operations.
- `mutate` composes directly inside larger Effect workflows.
- `mutateView` exposes call-specific progress as a View for React or Effect
consumers.
- `mutateStream` exposes call-specific progress as a `Stream` for React or
Effect consumers.
- Retry schedules, timeouts, tracing, logging, metrics, schema validation, and
concurrency controls can be applied with normal Effect operators.
packages/docs/docs/query.md
+13 -10
View File
@@ -19,14 +19,15 @@ If you already know TanStack Query, the main concepts translate directly:
| `queryFn` | `f: (key: K) => Effect<A, E, R>` |
| `useQuery` result | `query.state`, a `View<QueryState<K, A, E>>` |
| `isFetching` | `result.waiting` |
| `refetch` | `query.refresh` or `query.refreshView` |
| `refetch` | `query.refresh` or `query.refreshStream` |
| `invalidateQueries` | `query.invalidateCache` or `invalidateCacheEntry` |
The goal is familiar query behavior without leaving Effect's model. Query
functions keep their typed success, error, and service channels. They can use
services from the runtime, be composed with schema decoding and retry policies,
and are interrupted automatically when their scope ends or a new key supersedes
the current request.
the current request. A `fetch` or `refresh` that a newer request supersedes is
interrupted as well, rather than returning a result.
## Provide a QueryClient
@@ -216,7 +217,7 @@ const runSync = yield* Component.useRunSync()
return (
<div>
<button onClick={() => runSync(query.refreshView)}>
<button onClick={() => runSync(query.refreshStream)}>
Refresh current post
</button>
<button onClick={() => runSync(query.invalidateCacheEntry(["post", id] as const))}>
@@ -234,18 +235,20 @@ The methods come in two styles:
| Method | Behavior |
| --- | --- |
| `fetch(key)` | Fetch a specific key and wait for its final state. |
| `fetchView(key)` | Start fetching and immediately return a live state `View`. |
| `fetchStream(key)` | Start fetching and immediately return a `Stream` of the request's states, ending once it settles. |
| `refresh` | Resolve the current key again and wait for its final state. |
| `refreshView` | Resolve the current key again and immediately return a live state `View`. |
| `refreshStream` | Resolve the current key again and immediately return a `Stream` of the request's states, ending once it settles. |
| `invalidateCacheEntry(key)` | Remove the cached success for one key. |
| `invalidateCache` | Remove every cached success associated with this query function. |
The `*View` variants are convenient in synchronous UI callbacks: they return
after the scoped request has started, while the returned View and `query.state`
continue to publish progress. The non-View variants are useful in Effect
workflows that need to wait for the final success or failure state.
The `*Stream` variants are convenient in synchronous UI callbacks: they return
after the scoped request has started. The returned stream emits the request's
current state, then each change, and ends once the request settles or is
superseded; `query.state` keeps publishing progress either way. The non-Stream
variants are useful in Effect workflows that need to wait for the final success
or failure state.
Invalidating does not itself refetch. Follow it with `refreshView`, change the
Invalidating does not itself refetch. Follow it with `refresh` or `refreshStream`, change the
key, or allow a later fetch to repopulate the cache.
### Refresh on an interval
packages/effect-view/ai-docs/Mutation.md
+10 -10
View File
@@ -9,7 +9,7 @@ effect-view's counterpart to TanStack Query mutations: user-triggered asynchrono
| mutation result | `mutation.state`, a `View<{ key: Option<K>; result: AsyncResult<A, E> }>` |
| `isPending` | `state.result.waiting` |
| `mutateAsync` | `mutation.mutate(key)` |
| start without awaiting | `mutation.mutateView(key)` |
| start without awaiting | `mutation.mutateStream(key)` |
## Create
@@ -25,7 +25,7 @@ const mutation = yield* Component.useOnMount(() =>
## AsyncResult state
`mutation.state` is a `View` of `{ key: Option<K>, result: AsyncResult<A, E> }`. `result` starts `Initial` (`waiting: false`); calling `mutate`/`mutateView` sets `waiting: true`, then publishes `Success` or `Failure`. Match on `state.result`, not `state` itself:
`mutation.state` is a `View` of `{ key: Option<K>, result: AsyncResult<A, E> }`. `result` starts `Initial` (`waiting: false`); calling `mutate`/`mutateStream` sets `waiting: true`, then publishes `Success` or `Failure`. Match on `state.result`, not `state` itself:
```tsx
import { AsyncResult } from "effect/reactivity"
@@ -41,12 +41,12 @@ AsyncResult.match(state.result, {
`waiting` is independent of the result tag: after one success, starting another call keeps the value visible while `waiting: true`; if that call fails, the failure can retain `previousSuccess`. Failures carry a full `Cause<E>`.
## mutate vs mutateView
## mutate vs mutateStream
| Method | Returns | Use for |
|---|---|---|
| `mutate(key)` | the final `FinalMutationState` (`{ key: Option.Some<K>, result: Success \| Failure }`) | an Effect workflow that needs the outcome |
| `mutateView(key)` | a live per-call `View<{ key: Option.Some<K>, result: AsyncResult<A, E> }>` | a UI callback that just starts the work |
| `mutate(key)` | the final `MutationFinalState` (`{ key: Option.Some<K>, result: Success \| Failure }`) | an Effect workflow that needs the outcome |
| `mutateStream(key)` | a per-call `Stream<{ key: Option.Some<K>, result: AsyncResult<A, E> }>` of the call's states, starting with the current one and ending once it settles | a UI callback that just starts the work |
```tsx
const runPromise = yield* Component.useRunPromise()
@@ -58,7 +58,7 @@ void runPromise(Effect.gen(function* () {
```tsx
const runSync = yield* Component.useRunSync()
const state = runSync(mutation.mutateView(input)) // a View for this specific call
const states = runSync(mutation.mutateStream(input)) // a Stream for this specific call; consume with `Stream.use`
```
The mutation Effect never fails with `E` itself — it captures the operation's `Exit` and always resolves to a final state wrapping an `AsyncResult.Success`/`Failure`.
@@ -69,12 +69,12 @@ The mutation Effect never fails with `E` itself — it captures the operation's
|---|---|
| `state` | latest mutation state, shared `View` |
| `latestKey` | most recent input, `Option<K>` |
| `latestFinalState` | latest completed final state, `Option<FinalMutationState<K, A, E>>` |
| `fiber` | most recently started mutation fiber, `Option` |
| `latestFinalState` | latest completed final state, `Option<MutationFinalState<K, A, E>>` |
| `operation` | most recently started call, `Option<Operation>` (key, state, `fiber`, `interrupt`) |
## Concurrency
Starting a mutation does not interrupt an earlier one — calls can overlap, each with its own `mutateView` state; `mutation.state` reflects whichever update arrived last. For a single submit button, disabling while `result.waiting` is usually enough. Use per-call `mutateView` Views (e.g. per uploaded file) when concurrent operations each need their own progress indicator.
Starting a mutation does not interrupt an earlier one — calls can overlap, each with its own `mutateStream` stream; `mutation.state` (and `latestFinalState`) follow only the most recently started call; earlier calls still run to completion but no longer update them. For a single submit button, disabling while `result.waiting` is usually enough. Use per-call `mutateStream` streams (e.g. per uploaded file) when concurrent operations each need their own progress indicator.
## Updating queries after a mutation
@@ -84,6 +84,6 @@ Mutations never auto-invalidate `Query` caches — compose it explicitly:
const final = yield* updatePost.mutate(input)
if (AsyncResult.isSuccess(final.result)) {
yield* posts.invalidateCacheEntry(["post", final.result.value.id] as const)
yield* posts.refreshView // invalidation alone does not refetch
yield* posts.refresh // invalidation alone does not refetch
}
```
packages/effect-view/ai-docs/Query.md
+5 -5
View File
@@ -9,7 +9,7 @@ effect-view's take on TanStack Query: reactive query keys, cached results, stale
| `queryFn` | `f: (key: K) => Effect<A, E, R>` |
| `useQuery` result | `query.state`, a `View<QueryState<K, A, E>>` |
| `isFetching` | `result.waiting` |
| `refetch` | `query.refresh` / `query.refreshView` |
| `refetch` | `query.refresh` / `query.refreshStream` |
| `invalidateQueries` | `query.invalidateCache` / `invalidateCacheEntry` |
## QueryClient: the shared cache
@@ -88,13 +88,13 @@ AsyncResult.match(state.result, {
| Method | Behavior |
|---|---|
| `fetch(key)` | fetch a specific key, wait for its final state |
| `fetchView(key)` | start fetching a key, return immediately as a live state `View` |
| `fetchStream(key)` | start fetching a key, return immediately with a `Stream` of that request's states (ends once it settles or is superseded) |
| `refresh` | resolve the current key again, wait for its final state |
| `refreshView` | resolve the current key again, return immediately as a live `View` |
| `refreshStream` | resolve the current key again, return immediately with a `Stream` of that request's states (ends once it settles or is superseded) |
| `invalidateCacheEntry(key)` | remove the cached success for one key |
| `invalidateCache` | remove every cached success for this query |
The `*View` variants suit synchronous UI callbacks (`runSync(query.refreshView)`); the non-`View` variants suit Effect workflows waiting on the outcome. **Invalidating does not refetch by itself** — follow with `refreshView`, a key change, or a later natural fetch.
The `*Stream` variants suit synchronous UI callbacks (`runSync(query.refreshStream)`); the non-`Stream` variants suit Effect workflows waiting on the outcome. Each stream emits its request's current state first, so subscribing late still yields the final state. **Invalidating does not refetch by itself** — follow with `refresh`/`refreshStream`, a key change, or a later natural fetch.
```ts
import { Schedule } from "effect"
@@ -115,4 +115,4 @@ const query = yield* Query.make(options).pipe(
## The Effect touch
Request fibers belong to the creation scope and are interrupted on unmount, key replacement, or scope closure. Results are `View`s usable outside React too. Mutations do not auto-invalidate queries — compose it explicitly (see `Mutation.md`).
Request fibers belong to the creation scope and are interrupted on unmount, key replacement, or scope closure. A `fetch`/`refresh` superseded by a newer request is interrupted (it does not return a result), and the interrupted request's state stops waiting. Results are `View`s usable outside React too. Mutations do not auto-invalidate queries — compose it explicitly (see `Mutation.md`).
packages/effect-view/package.json
+4
View File
@@ -55,6 +55,10 @@
"types": "./dist/MutationForm.d.ts",
"default": "./dist/MutationForm.js"
},
"./Operation": {
"types": "./dist/Operation.d.ts",
"default": "./dist/Operation.js"
},
"./PubSub": {
"types": "./dist/PubSub.d.ts",
"default": "./dist/PubSub.js"
packages/effect-view/src/Mutation.test.ts
+65 -16
View File
@@ -1,4 +1,4 @@
import { Cause, Deferred, Effect, Option, type Scope } from "effect"
import { Cause, Deferred, Effect, Fiber, Option, type Scope, Stream } from "effect"
import { AsyncResult } from "effect/reactivity"
import { describe, expect, it } from "vitest"
import * as Mutation from "./Mutation.js"
@@ -8,6 +8,14 @@ import * as View from "./View.js"
const runMutationTest = <A, E>(effect: Effect.Effect<A, E, Scope.Scope>) =>
Effect.runPromise(Effect.scoped(effect))
const currentFiber = <K, A, E, R>(mutation: Mutation.Mutation<K, A, E, R>) => Effect.flatMap(
View.get(mutation.operation),
Option.match({
onSome: operation => View.get(operation.fiber),
onNone: () => Effect.succeedNone,
}),
)
const expectSuccessValue = <A, E>(state: { readonly result: AsyncResult.AsyncResult<A, E> }): A => {
expect(AsyncResult.isSuccess(state.result)).toBe(true)
@@ -32,7 +40,7 @@ describe("Mutation", () => {
latestKey: yield* View.get(mutation.latestKey),
state: yield* View.get(mutation.state),
latestFinalState: yield* View.get(mutation.latestFinalState),
fiber: yield* View.get(mutation.fiber),
fiber: yield* currentFiber(mutation),
}
}))
@@ -100,32 +108,73 @@ describe("Mutation", () => {
expect(expectSuccessValue(result.second)).toBe("b")
})
it("mutateView returns a waiting state without waiting for completion", async () => {
it("mutateStream returns without waiting for completion, and streams the states until settled", async () => {
const result = await runMutationTest(Effect.gen(function*() {
const deferred = yield* Deferred.make<string>()
const mutation = yield* Mutation.make({
f: (_key: string) => Deferred.await(deferred),
})
const state = yield* mutation.mutateView("save")
const stream = yield* mutation.mutateStream("save")
const states = yield* Effect.forkScoped(Stream.runCollect(stream))
yield* Effect.yieldNow
const pending = yield* View.get(state)
const hasRunningFiber = Option.isSome(yield* View.get(mutation.fiber))
const hasRunningFiber = Option.isSome(yield* currentFiber(mutation))
yield* Deferred.succeed(deferred, "saved")
yield* Effect.yieldNow
const final = yield* View.get(mutation.latestFinalState).pipe(Effect.flatMap(Effect.fromOption))
return { pending, hasRunningFiber, final }
return {
states: yield* Fiber.join(states).pipe(Effect.timeout("1 second")),
hasRunningFiber,
}
}))
expect(result.pending.key.value).toBe("save")
expect(AsyncResult.isInitial(result.pending.result)).toBe(true)
expect(result.pending.result.waiting).toBe(true)
expect(result.hasRunningFiber).toBe(true)
expect(result.final.key.value).toBe("save")
expect(expectSuccessValue(result.final)).toBe("saved")
expect(result.states).toHaveLength(2)
const [pending, final] = result.states
expect(pending.key.value).toBe("save")
expect(AsyncResult.isInitial(pending.result)).toBe(true)
expect(pending.result.waiting).toBe(true)
expect(final.key.value).toBe("save")
expect(expectSuccessValue(final)).toBe("saved")
})
it("exposes the most recently started operation", async () => {
const result = await runMutationTest(Effect.gen(function*() {
const mutation = yield* Mutation.make({
f: (key: string) => Effect.succeed(key),
})
yield* mutation.mutate("a")
yield* mutation.mutate("b")
return Option.map(yield* View.get(mutation.operation), operation => operation.key)
}))
expect(result).toEqual(Option.some(Option.some("b")))
})
it("an older mutation finishing late does not overwrite the newer state", async () => {
const result = await runMutationTest(Effect.gen(function*() {
const slow = yield* Deferred.make<string>()
const mutation = yield* Mutation.make({
f: (key: string) => key === "a" ? Deferred.await(slow) : Effect.succeed(key),
})
yield* Effect.asVoid(mutation.mutateStream("a"))
yield* Effect.yieldNow
yield* mutation.mutate("b")
yield* Deferred.succeed(slow, "a")
yield* Effect.sleep("10 millis")
return {
state: yield* View.get(mutation.state),
latestFinalState: yield* View.get(mutation.latestFinalState),
}
}))
expect(result.state.key).toEqual(Option.some("b"))
expect(expectSuccessValue(result.state)).toBe("b")
expect(Option.map(result.latestFinalState, s => s.key.value)).toEqual(Option.some("b"))
})
})
packages/effect-view/src/Mutation.ts
+54 -150
View File
@@ -1,7 +1,8 @@
import { Cause, type Context, Effect, Exit, type Fiber, Option, Pipeable, Predicate, PubSub, Ref, type Scope, Semaphore, Stream, SubscriptionRef } from "effect"
import { type Context, Effect, Option, Pipeable, Predicate, type Scope, type Stream, SubscriptionRef } from "effect"
import { AsyncResult } from "effect/reactivity"
import * as Lens from "./Lens.js"
import * as View from "./View.js"
import * as Operation from "./Operation.js"
import type * as View from "./View.js"
export const MutationTypeId: unique symbol = Symbol.for("@effect-view/Mutation/Mutation")
@@ -15,15 +16,15 @@ extends Pipeable.Pipeable {
readonly f: (key: K) => Effect.Effect<A, E, R>
readonly latestKey: View.View<Option.Option<K>>
readonly fiber: View.View<Option.Option<Fiber.Fiber<A, E>>>
readonly state: View.View<LatestMutationState<K, A, E>>
readonly latestFinalState: View.View<Option.Option<FinalMutationState<K, A, E>>>
readonly operation: View.View<Option.Option<Operation.Operation<Option.Some<K>, A, E, R>>>
readonly state: View.View<MutationLatestState<K, A, E>>
readonly latestFinalState: View.View<Option.Option<MutationFinalState<K, A, E>>>
mutate(key: K): Effect.Effect<FinalMutationState<K, A, E>>
mutateView(key: K): Effect.Effect<View.View<MutationState<K, A, E>>>
mutate(key: K): Effect.Effect<MutationFinalState<K, A, E>>
mutateStream(key: K): Effect.Effect<Stream.Stream<MutationState<K, A, E>>>
}
export interface LatestMutationState<out K, out A, out E = never> {
export interface MutationLatestState<out K, out A, out E = never> {
readonly key: Option.Option<K>
readonly result: AsyncResult.AsyncResult<A, E>
}
@@ -33,7 +34,7 @@ export interface MutationState<out K, out A, out E = never> {
readonly result: AsyncResult.AsyncResult<A, E>
}
export interface FinalMutationState<out K, out A, out E = never> {
export interface MutationFinalState<out K, out A, out E = never> {
readonly key: Option.Some<K>
readonly result: AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>
}
@@ -50,125 +51,69 @@ extends Pipeable.Class implements Mutation<K, A, E, R> {
readonly f: (key: K) => Effect.Effect<A, E, R>,
readonly latestKey: Lens.Lens<Option.Option<K>>,
readonly fiber: Lens.Lens<Option.Option<Fiber.Fiber<A, E>>>,
readonly state: Lens.Lens<LatestMutationState<K, A, E>>,
readonly latestFinalState: Lens.Lens<Option.Option<FinalMutationState<K, A, E>>>,
readonly operation: Lens.Lens<Option.Option<Operation.Operation<Option.Some<K>, A, E, R>>>,
readonly state: Lens.Lens<MutationLatestState<K, A, E>>,
readonly latestFinalState: Lens.Lens<Option.Option<MutationFinalState<K, A, E>>>,
) {
super()
}
mutate(key: K): Effect.Effect<FinalMutationState<K, A, E>> {
return Lens.set(this.latestKey, Option.some(key)).pipe(
Effect.andThen(this.start(key)),
Effect.flatMap(state => this.watch(state)),
mutate(key: K): Effect.Effect<MutationFinalState<K, A, E>> {
return this.start(key).pipe(
Effect.flatMap(operation => operation.await),
Effect.provide(this.context),
)
}
mutateView(key: K): Effect.Effect<View.View<MutationState<K, A, E>>> {
return Lens.set(this.latestKey, Option.some(key)).pipe(
Effect.andThen(this.start(key)),
Effect.tap(state => Effect.forkScoped(this.watch(state))),
mutateStream(key: K): Effect.Effect<Stream.Stream<MutationState<K, A, E>>> {
return this.start(key).pipe(
Effect.map(operation => operation.stream),
Effect.provide(this.context),
)
}
/** Starts a new Operation, which supersedes the current one without interrupting it. */
start(key: K): Effect.Effect<
View.View<MutationState<K, A, E>>,
Operation.Operation<Option.Some<K>, A, E, R>,
never,
Scope.Scope | R
> {
return Effect.gen({ self: this }, function*() {
const currentKey = Option.some(key) as Option.Some<K>
const latestFinalState = yield* Lens.get(this.latestFinalState)
const operation = yield* Operation.make({
key: Option.some(key) as Option.Some<K>,
effect: this.f(key),
previous: Option.isSome(latestFinalState)
? latestFinalState.value.result
: AsyncResult.initial<A, E>(),
listener: this.listener,
})
const previous: MutationState<K, A, E> = Option.getOrElse(yield* Lens.get(this.latestFinalState), () => ({
key: currentKey,
result: AsyncResult.initial(),
}))
const state = yield* makeMutationStateLens(previous)
const fiber = yield* Effect.forkScoped(Effect.andThen(
Lens.update<MutationState<K, A, E>, never, never, never, never>(
state,
previous => AsyncResult.match(previous.result, {
onInitial: () => ({
key: currentKey,
result: AsyncResult.initial(true),
}),
onSuccess: result => ({
key: currentKey,
result: AsyncResult.success(result.value, {
waiting: true,
}),
}),
onFailure: result => ({
key: currentKey,
result: AsyncResult.failure(result.cause, {
waiting: true,
previousSuccess: result.previousSuccess,
}),
}),
}
)),
Effect.onExit(this.f(key), exit => Effect.gen({ self: this }, function*() {
const fiberId = yield* Effect.fiberId
const fiber = yield* Lens.get(this.fiber)
if (Option.isSome(fiber) && fiberId === fiber.value.id)
yield* Lens.set(this.fiber, Option.none())
const finalState = (yield* Lens.updateAndGet<MutationState<K, A, E>, never, never, never, never>(
state,
previous => Exit.match(exit, {
onSuccess: v => ({
key: currentKey,
result: AsyncResult.success(v),
}),
onFailure: c => Cause.hasInterruptsOnly(c)
? previous
: AsyncResult.match(previous.result, {
onInitial: () => ({
key: currentKey,
result: AsyncResult.failure(c),
}),
onSuccess: v => ({
key: currentKey,
result: AsyncResult.failure(c, {
previousSuccess: Option.some(v),
}),
}),
onFailure: v => ({
key: currentKey,
result: AsyncResult.failure(c, {
previousSuccess: v.previousSuccess,
}),
}),
}),
}),
)) as FinalMutationState<K, A, E>
yield* Lens.set(this.latestFinalState, Option.some(finalState))
yield* PubSub.shutdown(state.pubsub)
}))
))
yield* Lens.set(this.fiber, Option.some(fiber))
return state
yield* Lens.set(this.latestKey, Option.some(key))
yield* Lens.set(this.operation, Option.some(operation))
yield* operation.start
return operation
})
}
watch(
state: View.View<MutationState<K, A, E>>
): Effect.Effect<FinalMutationState<K, A, E>> {
return View.get(state).pipe(
Effect.andThen(initial => Stream.runFoldEffect(
View.changes(state),
() => initial,
(_, result) => Effect.as(Lens.set(this.state, result), result),
) as Effect.Effect<FinalMutationState<K, A, E>>),
Effect.tap(result => Lens.set(this.latestFinalState, Option.some(result))),
/** Runs `effect` only if `operation` is the current Operation. */
ifCurrent(operation: Operation.Operation<Option.Some<K>, A, E, R>, effect: Effect.Effect<void>): Effect.Effect<void> {
return Effect.flatMap(
Lens.get(this.operation),
current => Option.isSome(current) && current.value === operation
? effect
: Effect.void,
)
}
/** Records the state of the current Operation. Superseded Operations are ignored. */
get listener(): Operation.OperationListener<Option.Some<K>, A, E, R> {
return {
onChange: (operation, state) => this.ifCurrent(operation, Lens.set(this.state, state)),
onSettle: (operation, state) => this.ifCurrent(operation, Lens.set(this.latestFinalState, Option.some(state))),
onInterrupt: () => Effect.void,
}
}
}
@@ -190,52 +135,11 @@ export const make = Effect.fnUntraced(function* <K = never, A = void, E = never,
options.f,
Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<K>())),
Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<Fiber.Fiber<A, E>>())),
Lens.fromSubscriptionRef(yield* SubscriptionRef.make<LatestMutationState<K, A, E>>({
Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<Operation.Operation<Option.Some<K>, A, E, R>>())),
Lens.fromSubscriptionRef(yield* SubscriptionRef.make<MutationLatestState<K, A, E>>({
key: Option.none(),
result: AsyncResult.initial(),
})),
Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<FinalMutationState<K, A, E>>())),
Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<MutationFinalState<K, A, E>>())),
)
})
export class MutationStateLens<in out K, in out A, in out E = never>
extends Lens.LensImpl<MutationState<K, A, E>, never, never, never, never> {
constructor(
readonly ref: Ref.Ref<MutationState<K, A, E>>,
readonly pubsub: PubSub.PubSub<MutationState<K, A, E>>,
readonly semaphore: Semaphore.Semaphore,
) {
super()
}
get resolve(): Effect.Effect<Lens.LensImpl.Resolved<MutationState<K, A, E>>, never, never> {
return Effect.map(
Ref.get(this.ref),
value => ({
value,
commit: next => Effect.flatMap(
next,
value => Effect.andThen(
Ref.set(this.ref, value),
PubSub.publish(this.pubsub, value),
),
),
}),
)
}
get changes() { return Stream.fromPubSub(this.pubsub) }
get lock() { return Effect.succeed(this.semaphore.withPermit) }
}
export const makeMutationStateLens = <K, A, E = never>(
initial: MutationState<K, A, E>,
) => Effect.all([
Ref.make(initial),
PubSub.unbounded<MutationState<K, A, E>>({ replay: 1 }),
Semaphore.make(1),
]).pipe(
Effect.tap(([, pubsub]) => PubSub.publish(pubsub, initial)),
Effect.map(([ref, pubsub, semaphore]) => new MutationStateLens(ref, pubsub, semaphore)),
)
packages/effect-view/src/MutationForm.ts
+3 -3
View File
@@ -24,7 +24,7 @@ extends Form.Form<readonly [], A, I, never, never> {
readonly run: Effect.Effect<void>
readonly submit: Effect.Effect<
Option.Option<Mutation.FinalMutationState<
Option.Option<Mutation.MutationFinalState<
readonly [value: A, form: MutationForm<A, I, RD, RE, unknown, unknown, unknown>],
MA, ME
>>,
@@ -141,7 +141,7 @@ extends Pipeable.Class implements MutationForm<A, I, RD, RE, MA, ME, MR> {
}
get submit(): Effect.Effect<
Option.Option<Mutation.FinalMutationState<
Option.Option<Mutation.MutationFinalState<
readonly [value: A, form: MutationForm<A, I, RD, RE, unknown, unknown, unknown>],
MA, ME
>>,
@@ -155,7 +155,7 @@ extends Pipeable.Class implements MutationForm<A, I, RD, RE, MA, ME, MR> {
}
submitValue(value: A): Effect.Effect<
Option.Option<Mutation.FinalMutationState<
Option.Option<Mutation.MutationFinalState<
readonly [value: A, form: MutationForm<A, I, RD, RE, unknown, unknown, unknown>],
MA, ME
>>,
packages/effect-view/src/Operation.ts
+258
View File
@@ -0,0 +1,258 @@
import { Cause, Deferred, Effect, Exit, Fiber, Option, Pipeable, Predicate, type Scope, Stream, SubscriptionRef } from "effect"
import { AsyncResult } from "effect/reactivity"
import * as Lens from "./Lens.js"
import * as View from "./View.js"
export const OperationTypeId: unique symbol = Symbol.for("@effect-view/Operation/Operation")
export type OperationTypeId = typeof OperationTypeId
/**
* A single run of a Query's or Mutation's effect for one key: owns its fiber and its state, from the previous result
* through waiting to the final result.
*
* An Operation that is interrupted has no final state: its state stops waiting, and awaiting it is interrupted.
*/
export interface Operation<out K, out A, out E = never, out R = never>
extends Pipeable.Pipeable {
readonly [OperationTypeId]: OperationTypeId
readonly key: K
readonly effect: Effect.Effect<A, E, R>
readonly fiber: View.View<Option.Option<Fiber.Fiber<A, E>>>
readonly state: View.View<OperationState<K, A, E>>
/**
* The successive states of the Operation, starting from the current one and ending once the Operation has settled or
* aborted.
*/
readonly stream: Stream.Stream<OperationState<K, A, E>>
/**
* Forks `effect` in the current scope, moving the state to waiting, then to the result of `effect`. Returns the
* running fiber, or `None` if it has already completed.
*/
readonly start: Effect.Effect<Option.Option<Fiber.Fiber<A, E>>, never, Scope.Scope | R>
readonly interrupt: Effect.Effect<void>
/** Waits for the Operation to settle, and returns its final state. Interrupted if the Operation is interrupted. */
readonly await: Effect.Effect<OperationFinalState<K, A, E>>
}
/** How an Operation reports to its owner (a Query or a Mutation). */
export interface OperationListener<in out K, in out A, in out E = never, in out R = never> {
onChange(
operation: Operation<K, A, E, R>,
state: OperationState<K, A, E>,
): Effect.Effect<void>
/** Called once, when the Operation settles with a final state. */
onSettle(
operation: Operation<K, A, E, R>,
state: OperationFinalState<K, A, E>,
): Effect.Effect<void>
/** Called once, when the Operation is interrupted before reaching a final state. */
onInterrupt(
operation: Operation<K, A, E, R>,
state: OperationState<K, A, E>,
): Effect.Effect<void>
}
export interface OperationState<out K, out A, out E = never> {
readonly key: K
readonly result: AsyncResult.AsyncResult<A, E>
}
export interface OperationFinalState<out K, out A, out E = never> {
readonly key: K
readonly result: AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>
}
export const isOperation = (u: unknown): u is Operation<unknown, unknown, unknown, unknown> => Predicate.hasProperty(u, OperationTypeId)
export class OperationImpl<in out K, in out A, in out E = never, in out R = never>
extends Pipeable.Class implements Operation<K, A, E, R> {
readonly [OperationTypeId]: OperationTypeId = OperationTypeId
constructor(
readonly key: K,
readonly effect: Effect.Effect<A, E, R>,
readonly listener: OperationListener<K, A, E, R>,
readonly fiber: Lens.Lens<Option.Option<Fiber.Fiber<A, E>>>,
readonly state: Lens.Lens<OperationState<K, A, E>>,
readonly result: Deferred.Deferred<OperationFinalState<K, A, E>>,
) {
super()
}
get start(): Effect.Effect<Option.Option<Fiber.Fiber<A, E>>, never, Scope.Scope | R> {
return Effect.gen({ self: this }, function*() {
// Started immediately so that `onExit` is always registered, even if the fiber is interrupted right away:
// otherwise the Operation would never settle or be reported as interrupted
const fiber = yield* Effect.forkScoped(
Effect.onExit(
Effect.andThen(
this.update(state => withWaiting(state, true)),
this.effect,
),
exit => Effect.flatMap(
Lens.get(this.state),
state => Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)
? this.abort(withWaiting(state, false))
: this.settle(done(state, exit)),
),
),
{ startImmediately: true },
)
// The fiber may already have completed and settled the Operation
if (!fiber.pollUnsafe())
yield* Lens.set(this.fiber, Option.some(fiber))
return yield* Lens.get(this.fiber)
})
}
get stream(): Stream.Stream<OperationState<K, A, E>> {
// The state is waiting from the moment the Operation starts, and stops waiting once it has settled or aborted
return Stream.takeUntil(View.changes(this.state), state => !state.result.waiting)
}
get interrupt(): Effect.Effect<void> {
return Effect.flatMap(Lens.get(this.fiber), Option.match({
onSome: Fiber.interrupt,
onNone: () => Effect.void,
}))
}
get await(): Effect.Effect<OperationFinalState<K, A, E>> {
return Deferred.await(this.result)
}
update(
f: (state: OperationState<K, A, E>) => OperationState<K, A, E>,
): Effect.Effect<OperationState<K, A, E>> {
return Effect.tap(
Lens.updateAndGet(this.state, f),
state => this.listener.onChange(this, state),
)
}
settle(state: OperationFinalState<K, A, E>): Effect.Effect<void> {
return Effect.gen({ self: this }, function*() {
yield* Lens.set(this.fiber, Option.none())
yield* this.update(() => state)
yield* this.listener.onSettle(this, state)
yield* Deferred.succeed(this.result, state)
})
}
abort(state: OperationState<K, A, E>): Effect.Effect<void> {
return Effect.gen({ self: this }, function*() {
yield* Lens.set(this.fiber, Option.none())
yield* this.update(() => state)
yield* this.listener.onInterrupt(this, state)
yield* Deferred.interrupt(this.result)
})
}
}
export declare namespace make {
export interface Options<K, A, E = never, R = never> {
readonly key: K
readonly effect: Effect.Effect<A, E, R>
readonly previous: AsyncResult.AsyncResult<A, E>
readonly listener: OperationListener<K, A, E, R>
}
}
export const make = Effect.fnUntraced(function* <K, A, E = never, R = never>(
options: make.Options<K, A, E, R>
): Effect.fn.Return<Operation<K, A, E, R>> {
return new OperationImpl(
options.key,
options.effect,
options.listener,
Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<Fiber.Fiber<A, E>>())),
Lens.fromSubscriptionRef(yield* SubscriptionRef.make<OperationState<K, A, E>>({
key: options.key,
result: options.previous,
})),
yield* Deferred.make<OperationFinalState<K, A, E>>(),
)
})
export declare namespace makeSettled {
export interface Options<K, A, E = never, R = never> {
readonly key: K
readonly effect: Effect.Effect<A, E, R>
readonly state: OperationFinalState<K, A, E>
}
}
/**
* Makes an Operation that has already settled with `state`, without running its effect, e.g. to serve a result from a
* cache. Being already settled, it never notifies a listener.
*/
export const makeSettled = Effect.fnUntraced(function* <K, A, E = never, R = never>(
options: makeSettled.Options<K, A, E, R>
): Effect.fn.Return<Operation<K, A, E, R>> {
const result = yield* Deferred.make<OperationFinalState<K, A, E>>()
yield* Deferred.succeed(result, options.state)
return new OperationImpl(
options.key,
options.effect,
{
onChange: () => Effect.void,
onSettle: () => Effect.void,
onInterrupt: () => Effect.void,
},
Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<Fiber.Fiber<A, E>>())),
Lens.fromSubscriptionRef(yield* SubscriptionRef.make<OperationState<K, A, E>>(options.state)),
result,
)
})
const withWaiting = <K, A, E>(
state: OperationState<K, A, E>,
waiting: boolean,
): OperationState<K, A, E> => ({
key: state.key,
result: AsyncResult.match(state.result, {
onInitial: () => AsyncResult.initial(waiting),
onSuccess: result => AsyncResult.success(result.value, {
waiting,
}),
onFailure: result => AsyncResult.failure(result.cause, {
waiting,
previousSuccess: result.previousSuccess,
}),
}),
})
const done = <K, A, E>(
state: OperationState<K, A, E>,
exit: Exit.Exit<A, E>,
): OperationFinalState<K, A, E> => Exit.match(exit, {
onSuccess: v => ({
key: state.key,
result: AsyncResult.success(v),
}),
onFailure: c => ({
key: state.key,
result: AsyncResult.match(state.result, {
onInitial: () => AsyncResult.failure(c),
onSuccess: v => AsyncResult.failure(c, {
previousSuccess: Option.some(v),
}),
onFailure: v => AsyncResult.failure(c, {
previousSuccess: v.previousSuccess,
}),
}),
}),
})
packages/effect-view/src/Query.test.ts
+198 -2
View File
@@ -1,4 +1,4 @@
import { Effect, Schedule, type Scope, Stream } from "effect"
import { Deferred, Effect, Exit, Fiber, Option, Schedule, type Scope, Stream } from "effect"
import { AsyncResult } from "effect/reactivity"
import { TestClock } from "effect/testing"
import { describe, expect, it } from "vitest"
@@ -18,7 +18,7 @@ const staticKey = <K>(key: K): View.View<K> => View.make({
})
const expectSuccessValue = <A, E>(
state: Query.FinalQueryState<unknown, A, E>,
state: Query.QueryFinalState<unknown, A, E>,
): A => {
expect(AsyncResult.isSuccess(state.result)).toBe(true)
@@ -208,4 +208,200 @@ describe("Query", () => {
expect(result.state.key).toEqual([1])
expect(expectSuccessValue(result.latestFinalState)).toBe("value:1:1")
})
it("a superseded fetch does not overwrite the newer state", async () => {
const result = await runQueryTest(Effect.gen(function*() {
const never = yield* Deferred.make<string>()
const query = yield* Query.make({
key: staticKey(1),
f: (id: number) => id === 1 ? Deferred.await(never) : Effect.succeed(`value:${id}`),
})
yield* Effect.asVoid(query.fetchStream(1))
yield* Effect.yieldNow
const second = yield* query.fetch(2)
yield* Effect.sleep("10 millis")
return {
second,
state: yield* View.get(query.state),
latestFinalState: yield* View.get(query.latestFinalState),
}
}))
expect(expectSuccessValue(result.second)).toBe("value:2")
expect(result.state.key).toBe(2)
expect(result.state.result.waiting).toBe(false)
expect(Option.map(result.latestFinalState, s => s.key)).toEqual(Option.some(2))
})
it("an interrupted fetch is not recorded as a final state", async () => {
const result = await runQueryTest(Effect.gen(function*() {
const never = yield* Deferred.make<string>()
const query = yield* Query.make({
key: staticKey(1),
f: (_id: number) => Deferred.await(never),
})
yield* Effect.asVoid(query.fetchStream(1))
yield* Effect.yieldNow
yield* Effect.asVoid(query.fetchStream(2))
yield* Effect.sleep("10 millis")
return {
state: yield* View.get(query.state),
latestFinalState: yield* View.get(query.latestFinalState),
}
}))
expect(result.state.key).toBe(2)
expect(result.latestFinalState).toEqual(Option.none())
})
it("a superseded fetch is interrupted, and its state stops waiting", async () => {
const result = await runQueryTest(Effect.gen(function*() {
const never = yield* Deferred.make<string>()
const query = yield* Query.make({
key: staticKey(1),
f: (id: number) => id === 1 ? Deferred.await(never) : Effect.succeed(`value:${id}`),
})
const firstFetch = yield* Effect.forkScoped(query.fetch(1))
yield* Effect.yieldNow
const first = yield* query.fetchStream(1)
yield* query.fetch(2)
return {
firstFetch: yield* Fiber.await(firstFetch),
firstStates: yield* Stream.runCollect(first).pipe(Effect.timeout("1 second")),
}
}))
expect(Exit.hasInterrupts(result.firstFetch)).toBe(true)
expect(result.firstStates.at(-1)?.result.waiting).toBe(false)
})
it("fetchStream ends once the fetch settles, and gives the final state to late subscribers", async () => {
const result = await runQueryTest(Effect.gen(function*() {
const gate = yield* Deferred.make<void>()
const query = yield* Query.make({
key: staticKey(1),
f: (id: number) => Effect.as(Deferred.await(gate), `value:${id}`),
})
const stream = yield* query.fetchStream(1)
const early = yield* Effect.forkScoped(Stream.runCollect(stream))
yield* Effect.yieldNow
yield* Deferred.succeed(gate, undefined)
const earlyStates = yield* Fiber.join(early).pipe(Effect.timeout("1 second"))
const lateStates = yield* Stream.runCollect(stream).pipe(Effect.timeout("1 second"))
return { earlyStates, lateStates }
}))
expect(result.earlyStates.map(s => [s.result._tag, s.result.waiting])).toEqual([
["Initial", true],
["Success", false],
])
expect(result.lateStates.map(s => s.result._tag)).toEqual(["Success"])
})
it("fetchStream ends once the fetch is aborted", async () => {
const result = await runQueryTest(Effect.gen(function*() {
const never = yield* Deferred.make<string>()
const query = yield* Query.make({
key: staticKey(1),
f: (id: number) => id === 1 ? Deferred.await(never) : Effect.succeed(`value:${id}`),
})
const stream = yield* query.fetchStream(1)
const states = yield* Effect.forkScoped(Stream.runCollect(stream))
yield* Effect.yieldNow
yield* query.fetch(2)
return yield* Fiber.join(states).pipe(Effect.timeout("1 second"))
}))
expect(result.at(-1)?.result.waiting).toBe(false)
})
it("exposes the current operation, replaced by each new request", async () => {
const result = await runQueryTest(Effect.gen(function*() {
const query = yield* Query.make({
key: staticKey(1),
f: (id: number) => Effect.succeed(`value:${id}`),
})
const before = yield* View.get(query.operation)
yield* query.fetch(1)
const first = yield* View.get(query.operation)
yield* query.fetch(2)
const second = yield* View.get(query.operation)
return {
before: Option.isNone(before),
firstKey: Option.map(first, operation => operation.key),
secondKey: Option.map(second, operation => operation.key),
}
}))
expect(result).toEqual({
before: true,
firstKey: Option.some(1),
secondKey: Option.some(2),
})
})
it("concurrent fetches superseding a running fetch leave a single running fiber", async () => {
const result = await runQueryTest(Effect.gen(function*() {
const never = yield* Deferred.make<string>()
let running = 0
const query = yield* Query.make({
key: staticKey(1),
f: (_id: number) => Effect.suspend(() => {
running += 1
return Deferred.await(never)
}).pipe(
// Asynchronous cleanup (e.g. aborting a request), so that interrupting it suspends
Effect.ensuring(Effect.andThen(Effect.sleep("1 millis"), Effect.sync(() => { running -= 1 }))),
),
})
yield* Effect.asVoid(query.fetchStream(0))
yield* Effect.yieldNow
yield* Effect.all([query.fetchStream(1), query.fetchStream(2)], { concurrency: "unbounded", discard: true })
yield* Effect.sleep("10 millis")
return running
}))
expect(result).toBe(1)
})
it("concurrent fetches never leave an orphaned running fiber", async () => {
const result = await runQueryTest(Effect.gen(function*() {
const never = yield* Deferred.make<string>()
let running = 0
const query = yield* Query.make({
key: staticKey(1),
f: (_id: number) => Effect.suspend(() => {
running += 1
return Deferred.await(never)
}).pipe(
Effect.ensuring(Effect.sync(() => { running -= 1 })),
),
})
yield* Effect.all([query.fetchStream(1), query.fetchStream(2)], { concurrency: "unbounded" })
yield* Effect.sleep("10 millis")
const runningBeforeInterrupt = running
const operation = yield* View.get(query.operation)
if (Option.isSome(operation))
yield* operation.value.interrupt
yield* Effect.sleep("10 millis")
return { runningBeforeInterrupt, runningAfterInterrupt: running }
}))
expect(result).toEqual({ runningBeforeInterrupt: 1, runningAfterInterrupt: 0 })
})
})
packages/effect-view/src/Query.ts
+113 -250
View File
@@ -1,6 +1,7 @@
import { Cause, type Context, Duration, Effect, Equal, type Equivalence, Exit, Fiber, Function, Option, Pipeable, Predicate, PubSub, Ref, type Schedule, type Scope, Semaphore, Stream, SubscriptionRef } from "effect"
import { type Cause, type Context, Duration, Effect, Equal, type Equivalence, Function, Option, Pipeable, Predicate, type Schedule, type Scope, Semaphore, Stream, SubscriptionRef } from "effect"
import { AsyncResult } from "effect/reactivity"
import * as Lens from "./Lens.js"
import * as Operation from "./Operation.js"
import * as QueryClient from "./QueryClient.js"
import * as View from "./View.js"
@@ -20,15 +21,15 @@ extends Pipeable.Pipeable {
readonly staleTime: Duration.Duration
readonly refreshOnWindowFocus: boolean
readonly fiber: View.View<Option.Option<Fiber.Fiber<A, E>>>
readonly operation: View.View<Option.Option<Operation.Operation<K, A, E, QueryClient.QueryClient | R>>>
readonly state: View.View<QueryState<K, A, E>>
readonly latestFinalState: View.View<Option.Option<FinalQueryState<K, A, E>>>
readonly latestFinalState: View.View<Option.Option<QueryFinalState<K, A, E>>>
readonly run: Effect.Effect<void>
fetch(key: K): Effect.Effect<FinalQueryState<K, A, E>, Cause.NoSuchElementError>
fetchView(key: K): Effect.Effect<View.View<QueryState<K, A, E>>, Cause.NoSuchElementError>
readonly refresh: Effect.Effect<FinalQueryState<K, A, E>, Cause.NoSuchElementError>
readonly refreshView: Effect.Effect<View.View<QueryState<K, A, E>>, Cause.NoSuchElementError>
fetch(key: K): Effect.Effect<QueryFinalState<K, A, E>, Cause.NoSuchElementError>
fetchStream(key: K): Effect.Effect<Stream.Stream<QueryState<K, A, E>>, Cause.NoSuchElementError>
readonly refresh: Effect.Effect<QueryFinalState<K, A, E>, Cause.NoSuchElementError>
readonly refreshStream: Effect.Effect<Stream.Stream<QueryState<K, A, E>>, Cause.NoSuchElementError>
readonly invalidateCache: Effect.Effect<void>
invalidateCacheEntry(key: K): Effect.Effect<void>
@@ -39,7 +40,7 @@ export interface QueryState<out K, out A, out E = never> {
readonly result: AsyncResult.AsyncResult<A, E>
}
export interface FinalQueryState<out K, out A, out E = never> {
export interface QueryFinalState<out K, out A, out E = never> {
readonly key: K
readonly result: AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>
}
@@ -60,11 +61,12 @@ extends Pipeable.Class implements Query<K, A, E, R> {
readonly staleTime: Duration.Duration,
readonly refreshOnWindowFocus: boolean,
readonly fiber: Lens.Lens<Option.Option<Fiber.Fiber<A, E>>>,
readonly operation: Lens.Lens<Option.Option<Operation.Operation<K, A, E, QueryClient.QueryClient | R>>>,
readonly state: Lens.Lens<QueryState<K, A, E>>,
readonly latestFinalState: Lens.Lens<Option.Option<FinalQueryState<K, A, E>>>,
readonly latestFinalState: Lens.Lens<Option.Option<QueryFinalState<K, A, E>>>,
readonly runSemaphore: Semaphore.Semaphore,
readonly startSemaphore: Semaphore.Semaphore,
) {
super()
}
@@ -73,28 +75,14 @@ extends Pipeable.Class implements Query<K, A, E, R> {
return Effect.all([
Stream.runForEach(
this.key.changes,
key => Effect.gen({ self: this }, function*() {
yield* this.interrupt
const latestFinalState = yield* Lens.get(this.latestFinalState)
const state = yield* this.startCached(
Option.isSome(latestFinalState) && this.keyEquivalence(key, latestFinalState.value.key)
? latestFinalState.value
: {
key,
result: AsyncResult.initial(false),
}
)
yield* Effect.forkScoped(this.watch(state))
}),
key => this.start(this.previousFor(key)),
),
Effect.promise(() => import("@effect/platform-browser")).pipe(
Effect.flatMap(({ BrowserStream }) => this.refreshOnWindowFocus
? Stream.runForEach(
BrowserStream.fromEventListenerWindow("focus"),
() => this.refreshView,
() => this.refreshStream,
)
: Effect.void
),
@@ -108,219 +96,134 @@ extends Pipeable.Class implements Query<K, A, E, R> {
}
get interrupt(): Effect.Effect<void> {
return Effect.flatMap(Lens.get(this.fiber), Option.match({
onSome: Fiber.interrupt,
return Effect.flatMap(Lens.get(this.operation), Option.match({
onSome: operation => operation.interrupt,
onNone: () => Effect.void,
}))
}
fetch(key: K): Effect.Effect<FinalQueryState<K, A, E>, Cause.NoSuchElementError> {
return Effect.gen({ self: this }, function*() {
yield* this.interrupt
const state = yield* this.startCached({
key,
result: AsyncResult.initial(false),
})
return yield* this.watch(state)
}).pipe(
fetch(key: K): Effect.Effect<QueryFinalState<K, A, E>, Cause.NoSuchElementError> {
return this.start(Effect.succeed({
key,
result: AsyncResult.initial(false),
})).pipe(
Effect.flatMap(operation => operation.await),
Effect.provide(this.context),
)
}
fetchView(key: K): Effect.Effect<
View.View<QueryState<K, A, E>>,
fetchStream(key: K): Effect.Effect<
Stream.Stream<QueryState<K, A, E>>,
Cause.NoSuchElementError
> {
return Effect.gen({ self: this }, function*() {
yield* this.interrupt
const state = yield* this.startCached({
key,
result: AsyncResult.initial(false),
})
yield* Effect.forkScoped(this.watch(state))
return state
}).pipe(
return this.start(Effect.succeed({
key,
result: AsyncResult.initial(false),
})).pipe(
Effect.map(operation => operation.stream),
Effect.provide(this.context),
)
}
get refresh(): Effect.Effect<FinalQueryState<K, A, E>, Cause.NoSuchElementError> {
return Effect.gen({ self: this }, function*() {
yield* this.interrupt
const latestState = yield* Lens.get(this.state)
const latestFinalState = yield* Lens.get(this.latestFinalState)
const state = yield* this.startCached(
Option.isSome(latestFinalState) && this.keyEquivalence(latestState.key, latestFinalState.value.key)
? latestFinalState.value
: {
key: latestState.key,
result: AsyncResult.initial(false),
}
)
return yield* this.watch(state)
}).pipe(
get refresh(): Effect.Effect<QueryFinalState<K, A, E>, Cause.NoSuchElementError> {
return this.start(this.previousForLatestKey).pipe(
Effect.flatMap(operation => operation.await),
Effect.provide(this.context),
)
}
get refreshView(): Effect.Effect<
View.View<QueryState<K, A, E>>,
get refreshStream(): Effect.Effect<
Stream.Stream<QueryState<K, A, E>>,
Cause.NoSuchElementError
> {
return Effect.gen({ self: this }, function*() {
yield* this.interrupt
const latestState = yield* Lens.get(this.state)
const latestFinalState = yield* Lens.get(this.latestFinalState)
const state = yield* this.startCached(
Option.isSome(latestFinalState) && this.keyEquivalence(latestState.key, latestFinalState.value.key)
? latestFinalState.value
: {
key: latestState.key,
result: AsyncResult.initial(false),
}
)
yield* Effect.forkScoped(this.watch(state))
return state
}).pipe(
return this.start(this.previousForLatestKey).pipe(
Effect.map(operation => operation.stream),
Effect.provide(this.context),
)
}
startCached(
previous: QueryState<K, A, E>,
previousFor(key: K): Effect.Effect<QueryState<K, A, E>> {
return Effect.map(Lens.get(this.latestFinalState), latestFinalState =>
Option.isSome(latestFinalState) && this.keyEquivalence(key, latestFinalState.value.key)
? latestFinalState.value
: {
key,
result: AsyncResult.initial(false),
}
)
}
get previousForLatestKey(): Effect.Effect<QueryState<K, A, E>> {
return Effect.flatMap(Lens.get(this.state), state => this.previousFor(state.key))
}
/**
* Interrupts the current Operation and replaces it with a new one, started from `previous` or served from the
* cache.
*/
start(
previous: Effect.Effect<QueryState<K, A, E>>,
): Effect.Effect<
View.View<QueryState<K, A, E>>,
Operation.Operation<K, A, E, QueryClient.QueryClient | R>,
Cause.NoSuchElementError,
Scope.Scope | QueryClient.QueryClient | R
> {
return Effect.flatMap(this.getCacheEntry(previous.key), Option.match({
onSome: entry => Effect.flatMap(
QueryClient.isQueryClientCacheEntryStale(entry),
isStale => isStale
? this.start({
key: previous.key,
result: entry.result as AsyncResult.AsyncResult<A, E>,
})
: Effect.succeed(View.make({
get: Effect.succeed({
key: previous.key,
result: entry.result as AsyncResult.AsyncResult<A, E>,
}),
get changes() {
return Stream.make({
key: previous.key,
result: entry.result as AsyncResult.AsyncResult<A, E>,
})
},
})),
),
onNone: () => this.start(previous),
}))
return Effect.gen({ self: this }, function*() {
yield* this.interrupt
const { key, result } = yield* previous
const entry = yield* this.getCacheEntry(key)
const isFresh = Option.isSome(entry) && !(yield* QueryClient.isQueryClientCacheEntryStale(entry.value))
const effect = Effect.tap(this.f(key), v => this.setCacheEntry(key, AsyncResult.success(v)))
if (Option.isSome(entry) && isFresh) {
const state: QueryFinalState<K, A, E> = {
key,
result: entry.value.result as AsyncResult.Success<A, E>,
}
const operation = yield* Operation.makeSettled({ key, effect, state })
yield* Lens.set(this.operation, Option.some(operation))
yield* this.listener.onChange(operation, state)
yield* this.listener.onSettle(operation, state)
return operation
}
else {
const operation = yield* Operation.make({
key,
effect,
previous: Option.isSome(entry) ? entry.value.result as AsyncResult.AsyncResult<A, E> : result,
listener: this.listener,
})
yield* Lens.set(this.operation, Option.some(operation))
yield* operation.start
return operation
}
}).pipe(
this.startSemaphore.withPermits(1),
)
}
start(
previous: QueryState<K, A, E>,
): Effect.Effect<
View.View<QueryState<K, A, E>>,
never,
Scope.Scope | R
> {
return Effect.gen({ self: this }, function*() {
const state = yield* makeQueryStateLens(previous)
const fiber = yield* Effect.forkScoped(Effect.andThen(
Lens.update<QueryState<K, A, E>, never, never, never, never>(
state,
previous => AsyncResult.match(previous.result, {
onInitial: () => ({
key: previous.key,
result: AsyncResult.initial(true),
}),
onSuccess: result => ({
key: previous.key,
result: AsyncResult.success(result.value, {
waiting: true,
}),
}),
onFailure: result => ({
key: previous.key,
result: AsyncResult.failure(result.cause, {
waiting: true,
previousSuccess: result.previousSuccess,
}),
}),
}
)),
Effect.onExit(this.f(previous.key), exit => Effect.gen({ self: this }, function*() {
const fiberId = yield* Effect.fiberId
const fiber = yield* Lens.get(this.fiber)
if (Option.isSome(fiber) && fiberId === fiber.value.id)
yield* Lens.set(this.fiber, Option.none())
const finalState = (yield* Lens.updateAndGet<QueryState<K, A, E>, never, never, never, never>(
state,
previous => Exit.match(exit, {
onSuccess: v => ({
key: previous.key,
result: AsyncResult.success(v),
}),
onFailure: c => Cause.hasInterruptsOnly(c)
? previous
: AsyncResult.match(previous.result, {
onInitial: () => ({
key: previous.key,
result: AsyncResult.failure(c),
}),
onSuccess: v => ({
key: previous.key,
result: AsyncResult.failure(c, {
previousSuccess: Option.some(v),
}),
}),
onFailure: v => ({
key: previous.key,
result: AsyncResult.failure(c, {
previousSuccess: v.previousSuccess,
}),
}),
}),
}),
)) as FinalQueryState<K, A, E>
yield* Lens.set(this.latestFinalState, Option.some(finalState))
yield* PubSub.shutdown(state.pubsub)
}))
))
yield* Lens.set(this.fiber, Option.some(fiber))
return state
})
/** Runs `effect` only if `operation` is the current Operation. */
ifCurrent(
operation: Operation.Operation<K, A, E, QueryClient.QueryClient | R>,
effect: Effect.Effect<void>,
): Effect.Effect<void> {
return Effect.flatMap(
Lens.get(this.operation),
current => Option.isSome(current) && current.value === operation
? effect
: Effect.void,
)
}
watch(
view: View.View<QueryState<K, A, E>>
): Effect.Effect<FinalQueryState<K, A, E>, never, QueryClient.QueryClient> {
return Effect.gen({ self: this }, function*() {
const initial = yield* View.get(view)
const final = yield* Stream.runFoldEffect(
View.changes(view),
() => initial,
(_, state) => Effect.as(Lens.set(this.state, state), state),
) as Effect.Effect<FinalQueryState<K, A, E>>
yield* Lens.set(this.latestFinalState, Option.some(final))
if (AsyncResult.isSuccess(final.result))
yield* this.setCacheEntry(final.key, final.result)
return final
})
/** Records the state of the current Operation. Superseded Operations are ignored. */
get listener(): Operation.OperationListener<K, A, E, QueryClient.QueryClient | R> {
return {
onChange: (operation, state) => this.ifCurrent(operation, Lens.set(this.state, state)),
onSettle: (operation, state) => this.ifCurrent(operation, Lens.set(this.latestFinalState, Option.some(state))),
onInterrupt: () => Effect.void,
}
}
makeCacheKey(key: K): QueryClient.QueryClientCacheKey {
@@ -399,14 +302,15 @@ export const make = Effect.fnUntraced(function* <K, A, E = never, R = never>(
options.staleTime ? yield* Effect.fromOption(Duration.fromInput(options.staleTime)) : client.defaultStaleTime,
options.refreshOnWindowFocus ?? client.defaultRefreshOnWindowFocus,
Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<Fiber.Fiber<A, E>>())),
Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<Operation.Operation<K, A, E, QueryClient.QueryClient | R>>())),
Lens.fromSubscriptionRef(yield* SubscriptionRef.make<QueryState<K, A, E>>({
key: yield* View.get(options.key),
result: AsyncResult.initial(false),
})),
Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<FinalQueryState<K, A, E>>())),
Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<QueryFinalState<K, A, E>>())),
yield* Semaphore.make(1),
yield* Semaphore.make(1),
)
})
@@ -455,44 +359,3 @@ export const withScheduledRefresh: {
self,
query => Effect.forkScoped(Effect.schedule(query.refresh, schedule)),
))
export class QueryStateLens<in out K, in out A, in out E = never>
extends Lens.LensImpl<QueryState<K, A, E>, never, never, never, never> {
constructor(
readonly ref: Ref.Ref<QueryState<K, A, E>>,
readonly pubsub: PubSub.PubSub<QueryState<K, A, E>>,
readonly semaphore: Semaphore.Semaphore,
) {
super()
}
get resolve(): Effect.Effect<Lens.LensImpl.Resolved<QueryState<K, A, E>>, never, never> {
return Effect.map(
Ref.get(this.ref),
value => ({
value,
commit: next => Effect.flatMap(
next,
value => Effect.andThen(
Ref.set(this.ref, value),
PubSub.publish(this.pubsub, value),
),
),
}),
)
}
get changes() { return Stream.fromPubSub(this.pubsub) }
get lock() { return Effect.succeed(this.semaphore.withPermit) }
}
export const makeQueryStateLens = <K, A, E = never>(
initial: QueryState<K, A, E>,
) => Effect.all([
Ref.make(initial),
PubSub.unbounded<QueryState<K, A, E>>({ replay: 1 }),
Semaphore.make(1),
]).pipe(
Effect.tap(([, pubsub]) => PubSub.publish(pubsub, initial)),
Effect.map(([ref, pubsub, semaphore]) => new QueryStateLens(ref, pubsub, semaphore)),
)
packages/effect-view/src/index.ts
+1
View File
@@ -6,6 +6,7 @@ export * as LensForm from "./LensForm.js"
export * as Memoized from "./Memoized.js"
export * as Mutation from "./Mutation.js"
export * as MutationForm from "./MutationForm.js"
export * as Operation from "./Operation.js"
export * as PubSub from "./PubSub.js"
export * as Query from "./Query.js"
export * as QueryClient from "./QueryClient.js"
packages/example/src/routes/query.tsx
+2 -2
View File
@@ -79,7 +79,7 @@ const QueryRouteComponent = Component.make("QueryRouteView")(function*() {
<PostResultView result={queryState.result} />
<Flex direction="row" justify="center" align="center" gap="1">
<Button onClick={() => runSync(query.refreshView)}>
<Button onClick={() => runSync(query.refreshStream)}>
Refresh
</Button>
<Button onClick={() => runSync(query.invalidateCache)}>
@@ -89,7 +89,7 @@ const QueryRouteComponent = Component.make("QueryRouteView")(function*() {
<PostResultView result={mutationState.result} />
<Button onClick={() => runSync(mutation.mutateView([id]))}>
<Button onClick={() => runSync(mutation.mutateStream([id]))}>
Mutate
</Button>
</Flex>