Release 0.3.0: Query/Mutation race fixes + offline support (#81)
Publish / publish (push) Successful in 1m56s
Lint / lint (push) Successful in 24s

## Summary

Merges `next` → `master` (12 commits, 22 files) for the effect-view **0.3.0** release: offline support (`networkMode`) for `Query`/`Mutation`, an operation-based rework that fixes `Query`/`Mutation` race conditions, and Stream-based request APIs. Full entry: `packages/effect-view/CHANGELOG.md`.

### Offline support (`networkMode`)
- `Query.networkMode` (new `Operation.NetworkMode`). `"online"` (default): a request made offline waits for a connection before running — its state has `paused: true` meanwhile, with `result.waiting` still `true`. `"offlineFirst"` runs right away and, if it fails while offline, pauses until the connection returns, then runs again. `"always"` ignores the connection.
- `refreshOnReconnect` (defaults `true`, except in `"always"` mode): refreshes the current key when the connection returns, subject to `staleTime`.
- `refreshOnFocus` / `refreshOnReconnect` accept `"always"` to refresh even a fresh result (`Query.RefreshOn`).
- Focus/reconnect refreshes no longer restart a request that is already running, and paused requests resume by themselves.
- New `AppConnectivity` and `AppFocus` services, each providing `layerBrowser` (`navigator.onLine` + `online`/`offline` events; page visibility), `layerDummy` (e.g. for tests) and `make(view)` (from any `View<boolean>`, e.g. on React Native).
- `Mutation.networkMode` (defaults to the client's new `defaultMutationNetworkMode`, `"online"`): mutations called offline are paused until the connection returns.
- `Query.withScheduledRefresh` now skips refreshes while the app is not focused; a skipped refresh still counts toward the schedule. Pass `refreshInBackground: true` to keep the previous behavior.
- `QueryClient` gains `defaultNetworkMode`, `defaultRefreshOnReconnect` and `defaultMutationNetworkMode`.

### Query/Mutation race fixes (new `Operation` module)
Each run of a Query or Mutation is now an `Operation`, and only the current one updates `state` / `latestFinalState`. Fixes, most visible on React Native:
- a request's final state could be lost, leaving `state` stuck on `waiting: true` and the result out of the cache;
- a superseded request could overwrite the state of a newer one, or be recorded in `latestFinalState` while still waiting;
- concurrent `fetch`/`refresh` calls superseding a running request could leave an extra request running that was never interrupted.

Also: serving a fresh result from the cache no longer resets that cache entry's staleness timer.

### Breaking changes
- `fetchView` / `refreshView` / `mutateView` → `fetchStream` / `refreshStream` / `mutateStream`: same request, but returns a `Stream` of that request's states. It emits the current state first, so subscribing late still yields the final state, then ends once the request settles or is interrupted. Consume with `Stream.use`.
- A `fetch`/`refresh` superseded by a newer request is now interrupted instead of returning its unfinished state typed as final; its state goes back to the previous result.
- `fiber` removed from `Query` / `Mutation`; use the new `operation` View (`key`, `state`, `stream`, `fiber`, `interrupt`, `await`).
- Overlapping mutations: `mutation.state` / `latestFinalState` follow only the most recently started call — earlier calls still run to completion but no longer update them. Use `mutateStream` to follow each call separately.
- Renames: `FinalQueryState` → `QueryFinalState`, `FinalMutationState` → `MutationFinalState`, `LatestMutationState` → `MutationLatestState`; removed `QueryStateLens`, `makeQueryStateLens`, `MutationStateLens` and `makeMutationStateLens`.
- `QueryClient.layer` is no longer a function: it's now a plain `Layer` with the default options, and `QueryClient.layerWithOptions` takes options. Both now require the `AppConnectivity` and `AppFocus` services; in a browser use `QueryClient.layerBrowser` / `layerBrowserWithOptions` (which provide both), elsewhere provide the services (e.g. the `layerDummy`s).
- `Mutation.make` and `MutationForm.make` now require a `QueryClient`.
- `refreshOnWindowFocus` → `refreshOnFocus`, `defaultRefreshOnWindowFocus` → `defaultRefreshOnFocus` (focus now comes from the `AppFocus` service).

### Docs & tests
- New `CHANGELOG.md`; docs (`query.md`, `mutation.md`, `getting-started.md`), ai-docs and the example app updated.
- New `Query.network.test.ts` (network modes); `Query.test.ts` and `Mutation.test.ts` expanded.

---------

Co-authored-by: Julien Valverdé <julien.valverde@mailo.com>
Reviewed-on: #81
This commit was merged in pull request #81.
This commit is contained in:
ThilawynandJulien Valverdé committed 2026-10-11 22:52:47 +02:00
1 parent 4d90ea2c0e
commit 663f536fa3
22 files changed
+2056 -543

No files matched your search

+1 -4
View File
@@ -82,13 +82,12 @@
},
"packages/effect-view": {
"name": "effect-view",
"version": "0.2.0",
"version": "0.3.0",
"dependencies": {
"@standard-schema/spec": "^1.1.0",
"effect-lens": "2.0.3",
},
"devDependencies": {
"@effect/platform-browser": "^4.0.0",
"@testing-library/react": "^16.3.0",
"effect": "^4.0.0",
"jsdom": "^26.1.0",
@@ -3130,8 +3129,6 @@
"dot-prop/is-obj": ["is-obj@2.0.0", "", {}, "sha512-drqDG3cbczxxEJRoOXcOjtdp1J/lyp1mNn0xaznRs8+muBhgQcrnbspox5X5fOw0HnMnbfDzvnEMEtqDEJEo8w=="],
"effect-view/@effect/platform-browser": ["@effect/platform-browser@4.0.0", "", { "peerDependencies": { "effect": "^4.0.0" } }, "sha512-OHo7EPpF3RVPiI+mjoU9yd05LdCdKPjzDXqs5wRgF9CsdVvwN76B0zgt54DxPUcvMYri1tVJOtLCVnK0tRkNQQ=="],
"effect-view/effect": ["effect@4.0.0", "", {}, "sha512-ooc1TG5t+FfzgYnFz2ff6BBKyZ7EwBRVXC7c4RhQUAD6/TZ2gTXXMeb4WX7a19ozQo4J73/QW+S00YAIresoMQ=="],
"effect-view/effect-lens": ["effect-lens@2.0.3", "", { "peerDependencies": { "effect": "^4.0.0" } }, "sha512-FLILgEXUf1aKxykcVR+Rc3qmWYPSeb4zLc2NdNEW+RNgnsICTan8rhhXW374e0dwEQAsEkMmcHxeEaQxBqy0jA=="],
+2 -2
View File
@@ -72,11 +72,11 @@ import { Layer } from "effect"
import { FetchHttpClient } from "effect/http"
import { ReactRuntime } from "effect-view"
const AppLive = Layer.empty.pipe(
const layer = Layer.empty.pipe(
Layer.provideMerge(FetchHttpClient.layer),
)
export const runtime = ReactRuntime.make(AppLive)
export const runtime = ReactRuntime.make(layer)
```
Keep the runtime stable. Creating it during a React render would create new
+46 -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
})
```
@@ -228,6 +231,27 @@ usually enough to prevent accidental overlap. For “latest request wins”
behavior, model that policy explicitly or use Query when the operation is
actually a reactive read.
## Network mode
Mutations take a `networkMode` option, just like queries. When it is not set,
the `QueryClient`'s `defaultMutationNetworkMode` is used, which is `"online"`
unless you change it. It is separate from `defaultNetworkMode`, which only
applies to queries. The connection status comes from the `AppConnectivity` service,
as described in [Query](./query.md#network-mode-and-offline-support).
| `networkMode` | Behavior |
| --- | --- |
| `"online"` | A call made offline waits for a connection before running, and `state.paused` is `true` meanwhile. |
| `"offlineFirst"` | A call runs right away. If it fails while offline, it is `paused` until the connection returns, then runs again. |
| `"always"` | Calls run regardless of the connection. |
```ts
const mutation = yield* Mutation.make({
f: (input: InviteInput) => sendInvite(input),
networkMode: "always",
})
```
## Update queries after a mutation
Mutations do not invalidate Query caches automatically. Compose invalidation
@@ -240,7 +264,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 +277,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 +290,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.
+153 -30
View File
@@ -19,40 +19,53 @@ 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` |
| `fetchStatus: 'paused'` | `state.paused` |
| `refetch` | `query.refresh` or `query.refreshStream` |
| `invalidateQueries` | `query.invalidateCache` or `invalidateCacheEntry` |
| `networkMode` | `networkMode` (`"online"`, `"offlineFirst"` or `"always"`) |
| `onlineManager` | The `AppConnectivity` service |
| `focusManager` | The `AppFocus` service |
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
Queries share successful results through a `QueryClient`. Add its layer to the
application runtime once, alongside the services used by your query effects:
application runtime once, alongside the services used by your query effects. In
a browser, use `QueryClient.layerBrowser`, or `QueryClient.layerBrowserWithOptions`
to change the defaults:
```tsx title="src/runtime.ts"
import { Layer } from "effect"
import { QueryClient, ReactRuntime } from "effect-view"
const AppLive = Layer.empty.pipe(
Layer.provideMerge(QueryClient.layer({
const layer = Layer.empty.pipe(
Layer.provideMerge(QueryClient.layerBrowserWithOptions({
defaultStaleTime: "30 seconds",
defaultRefreshOnWindowFocus: true,
defaultRefreshOnFocus: true,
cacheGcTime: "5 minutes",
})),
)
export const runtime = ReactRuntime.make(AppLive)
export const runtime = ReactRuntime.make(layer)
```
The client owns the cache and its cleanup lifecycle. Individual queries can
override `staleTime` and `refreshOnWindowFocus`; otherwise they inherit the
client defaults. Window-focus refresh also requires the optional
`@effect/platform-browser` package, as described under
[Staleness and cache lifetime](#staleness-and-cache-lifetime).
override `staleTime`, `networkMode`, `refreshOnFocus` and
`refreshOnReconnect`; otherwise they inherit the client defaults
(`defaultStaleTime`, `defaultNetworkMode`, `defaultRefreshOnFocus` and
`defaultRefreshOnReconnect`).
The client needs to know whether the app is focused and online, which it gets
from the `AppFocus` and `AppConnectivity` services. The browser layers provide
both from the window's events. `QueryClient.layer` and
`QueryClient.layerWithOptions` leave them for you to provide, for example on
other platforms such as [React Native](#react-native).
## Create a reactive query
@@ -216,7 +229,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 +247,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
@@ -277,6 +292,18 @@ const query = yield* Query.make(options).pipe(
The refresh fiber stops with the surrounding scope. The Effect returns the
original query. Cache and `staleTime` rules still apply.
Like TanStack Query's `refetchInterval`, scheduled refreshes are skipped while
the app is not focused, as reported by the `AppFocus` service. A skipped
refresh still counts toward the schedule. Pass `refreshInBackground: true` to
keep refreshing in the background:
```ts
const query = yield* Query.make(options).pipe(
Query.thenRun,
Query.withScheduledRefresh(Schedule.spaced("5 minutes"), { refreshInBackground: true }),
)
```
## Staleness and cache lifetime
`staleTime` controls how long a successful result can satisfy a fetch without
@@ -287,29 +314,125 @@ while the query refreshes it.
accessed are eventually removed after their stale period plus the configured
garbage-collection time.
By default, the client enables refresh on browser window focus. Set it globally
or override it for one query:
By default, the client enables refresh when the app regains focus. Set it
globally or override it for one query:
```tsx
const query = yield* Query.make({
key,
f: loadPost,
staleTime: "10 seconds",
refreshOnWindowFocus: false,
refreshOnFocus: false,
}).pipe(Query.thenRun)
```
Window-focus refresh depends on the optional `@effect/platform-browser`
integration. Install it in browser applications that use this behavior:
Regaining focus resolves the current key again, subject to the same freshness
check. Set `refreshOnFocus: "always"` to refresh even when the result is still
fresh. Focus comes from the `AppFocus` service: with
`QueryClient.layerBrowser` the app is focused while the page is visible,
following the document's `visibilitychange` events, like TanStack Query.
Other platforms, such as React Native, provide it themselves, as shown in
[React Native](#react-native).
```bash npm2yarn
npm install @effect/platform-browser
## Network mode and offline support
A query's `networkMode` decides how it behaves without a network connection:
| `networkMode` | Behavior |
| --- | --- |
| `"online"` (default) | A request waits for a connection before running `f`. Meanwhile its state is `paused`, with `result.waiting` still `true`. |
| `"offlineFirst"` | A request runs right away, e.g. to be served by a service worker or HTTP cache. If it fails while offline, it is `paused` until the connection returns, then runs again. |
| `"always"` | Requests ignore the connection. Suited to queries that do not use the network. |
Use `state.paused` to tell "loading" apart from "waiting for a connection":
```tsx
const [state] = yield* View.useAll([query.state])
if (state.paused) return <p>Offline. Waiting for a connection…</p>
```
When the package is available, focusing the window resolves the current key
again, subject to the same freshness check. Without it,
`refreshOnWindowFocus` has no effect; the rest of the Query API continues to
work normally. The integration is also ignored in non-browser environments.
With `refreshOnReconnect` (enabled by default, except in `"always"` mode),
regaining the connection resolves the current key again, subject to the usual
freshness check. Set it to `"always"` to refresh even when the result is still
fresh. Like TanStack Query, focus and reconnect refreshes never restart a
request that is already in flight: a running request keeps running, and one
paused for lack of a connection resumes on its own. A paused request can still
be superseded or interrupted like any other.
The connection status comes from the `AppConnectivity` service. With
`QueryClient.layerBrowser` it follows `navigator.onLine` and the window's
`online`/`offline` events.
### Provide AppConnectivity and AppFocus
Outside a browser, use `QueryClient.layer` and provide both services yourself.
Each module offers:
| Layer or function | Behavior |
| --- | --- |
| `layerBrowser` | Follows the browser: `navigator.onLine` with the `online`/`offline` events, and the page's visibility. |
| `layerDummy` | Always online, or always focused. Useful for tests and server rendering. |
| `make(view)` | Builds the service from a `View<boolean>`, to provide with `Layer.succeed`. |
```tsx
import { Layer } from "effect"
import { AppConnectivity, AppFocus, QueryClient } from "effect-view"
const queryClientLayer = QueryClient.layer.pipe(
Layer.provide(Layer.mergeAll(AppConnectivity.layerDummy, AppFocus.layerDummy)),
)
```
### React Native
React Native has neither window focus nor `online`/`offline` events. Build
both services from the platform APIs with `make`, and provide them to
`QueryClient.layer`:
```tsx title="src/runtime.ts"
import NetInfo from "@react-native-community/netinfo"
import { Effect, Layer, Queue, Stream } from "effect"
import { AppConnectivity, AppFocus, QueryClient, ReactRuntime, View } from "effect-view"
import { AppState } from "react-native"
const appConnectivityLayer = Layer.succeed(
AppConnectivity.AppConnectivity,
AppConnectivity.make(View.make({
get: Effect.map(Effect.promise(() => NetInfo.fetch()), state => !!state.isConnected),
// NetInfo calls the listener with the current state right away
changes: Stream.callback<boolean>(queue => Effect.acquireRelease(
Effect.sync(() => NetInfo.addEventListener(state => Queue.offerUnsafe(queue, !!state.isConnected))),
unsubscribe => Effect.sync(unsubscribe),
)),
})),
)
const appFocusLayer = Layer.succeed(
AppFocus.AppFocus,
AppFocus.make(View.make({
get: Effect.sync(() => AppState.currentState === "active"),
changes: Stream.concat(
Stream.sync(() => AppState.currentState === "active"),
Stream.callback<boolean>(queue => Effect.acquireRelease(
Effect.sync(() => AppState.addEventListener("change", status => Queue.offerUnsafe(queue, status === "active"))),
subscription => Effect.sync(() => subscription.remove()),
)),
),
})),
)
const layer = Layer.empty.pipe(
Layer.provideMerge(QueryClient.layer),
Layer.provideMerge(appConnectivityLayer),
Layer.provideMerge(appFocusLayer),
)
export const runtime = ReactRuntime.make(layer)
```
A service's `changes` stream starts with the current value, and the listeners
are removed when the layer is released.
## The Effect touch
+109
View File
@@ -0,0 +1,109 @@
# effect-view
## 0.3.0
### Minor Changes
- [`3875bc4`](https://git.valverde.cloud/Thilawyn/effect-view/commit/3875bc4f691dfa670137142e3bd62c226f9f5886) - Fix race conditions in `Query` and `Mutation`.
These were most visible on React Native, where scheduling made them easy to hit:
- A request's final state could be lost, leaving `state` stuck on `waiting: true` and the result out of the cache.
- A superseded request could overwrite the state of a newer one, or be recorded in `latestFinalState` while still waiting.
- Concurrent `fetch`/`refresh` calls superseding a running request could leave an extra request running that was never interrupted.
Each run of a Query or Mutation is now an `Operation` (new `Operation` module), and only the current one updates `state` and `latestFinalState`.
- [`3875bc4`](https://git.valverde.cloud/Thilawyn/effect-view/commit/3875bc4f691dfa670137142e3bd62c226f9f5886) - **Breaking:** replace `fetchView`, `refreshView` and `mutateView` with `fetchStream`, `refreshStream` and `mutateStream`.
They start the request the same way, but return a `Stream` of that request's states instead of a `View`. The stream emits the current state first, so subscribing late still yields the final state, then ends once the request settles or is interrupted. Consume it in a component with `Stream.use`.
```ts
// Before
runSync(query.refreshView)
runSync(mutation.mutateView(input))
// After
runSync(query.refreshStream)
runSync(mutation.mutateStream(input))
```
- [`3875bc4`](https://git.valverde.cloud/Thilawyn/effect-view/commit/3875bc4f691dfa670137142e3bd62c226f9f5886) - **Breaking:** a `fetch` or `refresh` superseded by a newer request is now interrupted, instead of returning its unfinished state typed as final. Its state goes back to the previous result, no longer waiting.
- [`3875bc4`](https://git.valverde.cloud/Thilawyn/effect-view/commit/3875bc4f691dfa670137142e3bd62c226f9f5886) - **Breaking:** remove `fiber` from `Query` and `Mutation`. Use the new `operation` View instead, which holds the current `Operation` (its `key`, `state`, `stream`, `fiber`, `interrupt` and `await`).
```ts
// Before
const fiber = yield* View.get(query.fiber)
// After
const operation = yield* View.get(query.operation)
if (Option.isSome(operation)) yield* operation.value.interrupt
```
- [`3875bc4`](https://git.valverde.cloud/Thilawyn/effect-view/commit/3875bc4f691dfa670137142e3bd62c226f9f5886) - **Breaking:** rename `FinalQueryState` to `QueryFinalState`, `FinalMutationState` to `MutationFinalState`, and `LatestMutationState` to `MutationLatestState`. Remove `QueryStateLens`, `makeQueryStateLens`, `MutationStateLens` and `makeMutationStateLens`.
- [`3875bc4`](https://git.valverde.cloud/Thilawyn/effect-view/commit/3875bc4f691dfa670137142e3bd62c226f9f5886) - **Breaking:** when mutations overlap, `mutation.state` and `latestFinalState` now follow only the most recently started call. Earlier calls still run to completion but no longer update them. Use `mutateStream` to follow each call separately.
- [`acb05fc`](https://git.valverde.cloud/Thilawyn/effect-view/commit/acb05fce44c6e67a9924ce878f20e39a4f59af71) - **Breaking:** `QueryClient.layer` is now a `Layer` using the default options. Use `QueryClient.layerWithOptions` to pass options.
```ts
// Before
QueryClient.layer()
QueryClient.layer({ defaultStaleTime: "30 seconds" })
// After
QueryClient.layer
QueryClient.layerWithOptions({ defaultStaleTime: "30 seconds" })
```
- [`2ed417a`](https://git.valverde.cloud/Thilawyn/effect-view/commit/2ed417a49f283c120af4034d1a496d3fe304dd05) - Add offline support to `Query`, like TanStack Query's `networkMode`.
- The new `networkMode` option (`Operation.NetworkMode`) is `"online"` by default: a request made offline waits for a connection before running, and its state has `paused: true` meanwhile, with `result.waiting` still `true`. With `"offlineFirst"`, a request runs right away, and if it fails while offline, it is paused until the connection returns, then runs again. With `"always"`, requests ignore the connection.
- The new `refreshOnReconnect` option refreshes the current key when the connection returns, subject to `staleTime`. It defaults to `true`, except in `"always"` mode.
- `refreshOnFocus` and `refreshOnReconnect` accept `"always"`, to refresh even a fresh result (`Query.RefreshOn`).
- `QueryClient` takes the matching defaults: `defaultNetworkMode` and `defaultRefreshOnReconnect`.
- Focus and reconnect refreshes no longer restart a request that is already running, and a paused request resumes by itself.
The connection status comes from the new `AppConnectivity` service, and focus from the new `AppFocus` service. Each module provides `layerBrowser` (`navigator.onLine` with the `online`/`offline` events, and the page's visibility, like TanStack Query), `layerDummy` (always online or focused, e.g. for tests) and `make(view)` to build the service from any `View<boolean>`, e.g. on React Native.
- [`79d9061`](https://git.valverde.cloud/Thilawyn/effect-view/commit/79d906190754193b1b86bd07cd176d402180509f) - **Breaking:** `Mutation` takes a `networkMode` option too, defaulting to the client's new `defaultMutationNetworkMode` (`"online"` by default). In the default `"online"` mode, mutations called offline are now paused until the connection returns. `Mutation.make` (and `MutationForm.make`) now require a `QueryClient`.
- [`2ed417a`](https://git.valverde.cloud/Thilawyn/effect-view/commit/2ed417a49f283c120af4034d1a496d3fe304dd05) - **Breaking:** `QueryClient.layer` and `QueryClient.layerWithOptions` now require the `AppConnectivity` and `AppFocus` services. In a browser, use `QueryClient.layerBrowser` and `QueryClient.layerBrowserWithOptions`, which provide both.
```ts
// Before
QueryClient.layer
QueryClient.layerWithOptions({ defaultStaleTime: "30 seconds" })
// After, in a browser
QueryClient.layerBrowser
QueryClient.layerBrowserWithOptions({ defaultStaleTime: "30 seconds" })
// After, elsewhere
QueryClient.layer.pipe(
Layer.provide(Layer.mergeAll(AppConnectivity.layerDummy, AppFocus.layerDummy)),
)
```
- [`2ed417a`](https://git.valverde.cloud/Thilawyn/effect-view/commit/2ed417a49f283c120af4034d1a496d3fe304dd05) - **Breaking:** rename `refreshOnWindowFocus` to `refreshOnFocus`, and `defaultRefreshOnWindowFocus` to `defaultRefreshOnFocus`. Focus is now read from the `AppFocus` service instead of the window directly, so it can be provided on any platform.
```ts
// Before
Query.make({ key, f, refreshOnWindowFocus: false })
QueryClient.layerWithOptions({ defaultRefreshOnWindowFocus: false })
// After
Query.make({ key, f, refreshOnFocus: false })
QueryClient.layerBrowserWithOptions({ defaultRefreshOnFocus: false })
```
- **Breaking:** like TanStack Query's `refetchInterval`, `Query.withScheduledRefresh` now skips refreshes while the app is not focused, as reported by the `AppFocus` service. A skipped refresh still counts toward the schedule. Pass `refreshInBackground: true` to keep the previous behavior.
```ts
Query.withScheduledRefresh(Schedule.spaced("5 minutes"), { refreshInBackground: true })
```
### Patch Changes
- [`3875bc4`](https://git.valverde.cloud/Thilawyn/effect-view/commit/3875bc4f691dfa670137142e3bd62c226f9f5886) - Serving a fresh result from the cache no longer resets that cache entry's staleness timer.
+14 -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,16 @@ 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.
## Network mode
Like queries, `Mutation.make({ f, networkMode? })` takes a `networkMode` (falls back to the `QueryClient`'s `defaultMutationNetworkMode`, `"online"` by default — separate from the queries' `defaultNetworkMode`). In `"online"` mode a call made offline waits for the `AppConnectivity` service to report a connection before running `f`; meanwhile `state.paused` is `true` (with `result.waiting` still `true`). `"offlineFirst"` runs `f` right away, and if it fails while offline, pauses until the connection returns and runs it again. `"always"` ignores the connection.
## Updating queries after a mutation
@@ -84,6 +88,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
}
```
+61 -14
View File
@@ -9,28 +9,31 @@ 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` |
| `fetchStatus: 'paused'` | `state.paused` |
| `refetch` | `query.refresh` / `query.refreshStream` |
| `invalidateQueries` | `query.invalidateCache` / `invalidateCacheEntry` |
| `networkMode` | `networkMode` (`"online"` / `"offlineFirst"` / `"always"`) |
| `onlineManager` / `focusManager` | the `AppConnectivity` / `AppFocus` services |
## QueryClient: the shared cache
Every `Query` reads and writes through a `QueryClient`, the Effect service that owns the cache and its garbage collection. Add it once to the application runtime:
Every `Query` reads and writes through a `QueryClient`, the Effect service that owns the cache and its garbage collection. Add it once to the application runtime (browser):
```tsx
import { Layer } from "effect"
import { QueryClient, ReactRuntime } from "effect-view"
const AppLive = Layer.empty.pipe(
Layer.provideMerge(QueryClient.layer({
const appLayer = Layer.empty.pipe(
Layer.provideMerge(QueryClient.layerBrowserWithOptions({
defaultStaleTime: "30 seconds", // default: "0 minutes"
defaultRefreshOnWindowFocus: true, // default: true
defaultRefreshOnFocus: true, // default: true
cacheGcTime: "5 minutes", // default: "5 minutes"
})),
)
export const runtime = ReactRuntime.make(AppLive)
export const runtime = ReactRuntime.make(appLayer)
```
`QueryClient.layer(options?)` builds the service and forks its background garbage-collection loop into the layer's scope. Individual queries may override `staleTime`/`refreshOnWindowFocus`; unset options fall back to these client defaults. `cacheGcTime` controls how long a stale, unaccessed cache entry is kept before eviction. You interact with the client only indirectly through `Query` instances — no need to call `QueryClientService` methods directly.
`QueryClient.layer` (defaults) / `QueryClient.layerWithOptions(options)` build the service and fork its background garbage-collection loop into the layer's scope. Individual queries may override `staleTime`/`networkMode`/`refreshOnFocus`/`refreshOnReconnect`; unset options fall back to these client defaults (`defaultStaleTime`, `defaultNetworkMode`, `defaultRefreshOnFocus`, `defaultRefreshOnReconnect`); mutations use `defaultMutationNetworkMode` instead of `defaultNetworkMode`. `cacheGcTime` controls how long a stale, unaccessed cache entry is kept before eviction. They require the `AppConnectivity` and `AppFocus` services (see "Network mode and offline support"); `QueryClient.layerBrowser` / `QueryClient.layerBrowserWithOptions(options)` provide the browser implementations of both. You interact with the client only indirectly through `Query` instances — no need to call `QueryClientService` methods directly.
## Create and run a query
@@ -59,7 +62,7 @@ const [postId, query] = yield* Component.useOnMount(() =>
)
```
- `Query.make({ key, f, staleTime?, refreshOnWindowFocus?, keyEquivalence? })` constructs the query; `Query.thenRun` starts watching `key` in the current scope.
- `Query.make({ key, f, staleTime?, networkMode?, refreshOnFocus?, refreshOnReconnect?, keyEquivalence? })` constructs the query; `Query.thenRun` starts watching `key` in the current scope.
- Create each query once and keep it stable — the usual home is `Component.useOnMount` (component-owned) or an Effect service (shared across components). Change the key, don't recreate the query, when input changes.
- Keys use Effect equality by default (`keyEquivalence` overrides it). A tuple key plays the same role as `['post', id]` in TanStack Query.
- `f` keeps its full `Effect<A, E, R>` type: required services, schema decoding, retries, tracing all compose normally. The context is captured at creation time.
@@ -88,13 +91,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"
@@ -105,14 +108,58 @@ const query = yield* Query.make(options).pipe(
)
```
`Query.withScheduledRefresh(schedule)` forks a refresh fiber tied to the surrounding scope; cache/`staleTime` rules still apply.
`Query.withScheduledRefresh(schedule, options?)` forks a refresh fiber tied to the surrounding scope; cache/`staleTime` rules still apply. As with TanStack's `refetchInterval`, refreshes due while the app is not focused (`AppFocus`) are skipped (still counting as a schedule recurrence) unless `{ refreshInBackground: true }`.
## Staleness and lifetime
- `staleTime` (per query, falls back to `QueryClient` default): how long a successful result satisfies a fetch without re-running `f`. A stale entry stays available as previous data while it refreshes.
- `cacheGcTime` (on `QueryClient`): entries unused for `staleTime + cacheGcTime` are evicted.
- `refreshOnWindowFocus` (per query, falls back to `QueryClient` default): re-resolves the current key on window focus. Requires the optional `@effect/platform-browser` package; without it, the option is silently ignored (rest of the API works normally). Also a no-op outside browser environments.
- `refreshOnFocus` (per query, falls back to `QueryClient` default): re-resolves the current key when the app regains focus, as reported by the `AppFocus` service (page visibility, via the document's `visibilitychange` events, with `AppFocus.layerBrowser`). `true` refreshes only a stale result, `"always"` regardless of `staleTime` (type `Query.RefreshOn`).
## Network mode and offline support
- `networkMode: "online"` (default): a request waits for a connection before running `f`; meanwhile `state.paused` is `true` (with `result.waiting` still `true`). Paused requests can be superseded/interrupted like any other, which clears `paused`.
- `networkMode: "offlineFirst"`: runs `f` right away (e.g. served by a service worker / HTTP cache); if it fails while offline, the request is `paused` until the connection returns, then `f` runs again. A failure while online settles normally.
- `networkMode: "always"`: ignores the connection (for queries that don't use the network).
- `refreshOnReconnect` (defaults to the client default, `false` in `"always"` mode): re-resolves the current key when the connection returns, subject to freshness (`"always"`: regardless of `staleTime`). Focus/reconnect refreshes never restart an in-flight request (a paused one resumes by itself), as in TanStack Query.
- The connection comes from the `AppConnectivity` service (`navigator.onLine` + window `online`/`offline` events with `AppConnectivity.layerBrowser`). `AppConnectivity` and `AppFocus` are `Context.Service`s and must be provided to `QueryClient.layer`/`layerWithOptions` (the `layerBrowser*` variants do it for you).
- Each module has `layerBrowser` (`navigator.onLine` + `online`/`offline` events; page visibility), `layerDummy` (always online / always focused; tests, SSR) and `make(view: View<boolean>)` to build the service from any `View`, provided with `Layer.succeed`. A `View`'s `changes` stream starts with the current value.
- React Native: build them from platform APIs:
```ts
const appConnectivityLayer = Layer.succeed(
AppConnectivity.AppConnectivity,
AppConnectivity.make(View.make({
get: Effect.map(Effect.promise(() => NetInfo.fetch()), state => !!state.isConnected),
// NetInfo calls the listener with the current state right away
changes: Stream.callback<boolean>(queue => Effect.acquireRelease(
Effect.sync(() => NetInfo.addEventListener(state => Queue.offerUnsafe(queue, !!state.isConnected))),
unsubscribe => Effect.sync(unsubscribe),
)),
})),
)
const appFocusLayer = Layer.succeed(
AppFocus.AppFocus,
AppFocus.make(View.make({
get: Effect.sync(() => AppState.currentState === "active"),
changes: Stream.concat(
Stream.sync(() => AppState.currentState === "active"),
Stream.callback<boolean>(queue => Effect.acquireRelease(
Effect.sync(() => AppState.addEventListener("change", status => Queue.offerUnsafe(queue, status === "active"))),
subscription => Effect.sync(() => subscription.remove()),
)),
),
})),
)
const appLayer = Layer.empty.pipe(
Layer.provideMerge(QueryClient.layer),
Layer.provideMerge(appConnectivityLayer),
Layer.provideMerge(appFocusLayer),
)
```
## 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`).
+2 -2
View File
@@ -8,9 +8,9 @@ Owns a managed Effect runtime and exposes it to a React subtree through context.
import { Layer } from "effect"
import { ReactRuntime } from "effect-view"
const AppLive = Layer.empty // add application layers here
const layer = Layer.empty // add application layers here
export const runtime = ReactRuntime.make(AppLive)
export const runtime = ReactRuntime.make(layer)
```
```tsx
+14 -2
View File
@@ -1,11 +1,12 @@
{
"name": "effect-view",
"description": "Write React function components with Effect",
"version": "0.2.0",
"version": "0.3.0",
"type": "module",
"files": [
"./README.md",
"./AGENTS.md",
"./CHANGELOG.md",
"./src/**/*.ts",
"!./src/**/*.test.ts",
"./dist/**/*.js",
@@ -23,6 +24,14 @@
"types": "./dist/index.d.ts",
"default": "./dist/index.js"
},
"./AppConnectivity": {
"types": "./dist/AppConnectivity.d.ts",
"default": "./dist/AppConnectivity.js"
},
"./AppFocus": {
"types": "./dist/AppFocus.d.ts",
"default": "./dist/AppFocus.js"
},
"./Async": {
"types": "./dist/Async.d.ts",
"default": "./dist/Async.js"
@@ -55,6 +64,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"
@@ -103,7 +116,6 @@
"clean:modules": "rm -rf node_modules"
},
"devDependencies": {
"@effect/platform-browser": "^4.0.0",
"@testing-library/react": "^16.3.0",
"effect": "^4.0.0",
"jsdom": "^26.1.0",
@@ -0,0 +1,63 @@
import { Context, Effect, Layer, Pipeable, Predicate, Queue, Stream } from "effect"
import * as View from "./View.js"
export const AppConnectivityServiceTypeId: unique symbol = Symbol.for("@effect-view/AppConnectivity/AppConnectivityService")
export type AppConnectivityServiceTypeId = typeof AppConnectivityServiceTypeId
/** Whether the app has a network connection, as used by Query's `networkMode` and `refreshOnReconnect`. */
export interface AppConnectivityService extends Pipeable.Pipeable {
readonly [AppConnectivityServiceTypeId]: AppConnectivityServiceTypeId
readonly online: View.View<boolean>
}
export const isAppConnectivityService = (u: unknown): u is AppConnectivityService => Predicate.hasProperty(u, AppConnectivityServiceTypeId)
export class AppConnectivityServiceImpl extends Pipeable.Class implements AppConnectivityService {
readonly [AppConnectivityServiceTypeId]: AppConnectivityServiceTypeId = AppConnectivityServiceTypeId
constructor(
readonly online: View.View<boolean>,
) {
super()
}
}
export const make = (online: View.View<boolean>): AppConnectivityService => new AppConnectivityServiceImpl(online)
/** Whether the app has a network connection, as used by Query's `networkMode` and `refreshOnReconnect`. */
export class AppConnectivity extends Context.Service<AppConnectivity, AppConnectivityService>()(
"@effect-view/AppConnectivity/AppConnectivity"
) {}
export const layerDummy: Layer.Layer<AppConnectivity> = Layer.succeed(
AppConnectivity,
make(View.make({
get: Effect.succeed(true),
changes: Stream.succeed(true),
})),
)
export const layerBrowser: Layer.Layer<AppConnectivity> = Layer.succeed(
AppConnectivity,
make(View.make({
get: Effect.sync(() => globalThis.navigator.onLine),
changes: Stream.concat(
Stream.sync(() => globalThis.navigator.onLine),
Stream.callback<boolean>(queue => Effect.acquireRelease(
Effect.sync(() => {
const onlineListener = () => Queue.offerUnsafe(queue, true)
globalThis.window.addEventListener("online", onlineListener)
const offlineListener = () => Queue.offerUnsafe(queue, false)
globalThis.window.addEventListener("offline", offlineListener)
return [onlineListener, offlineListener] as const
}),
([onlineListener, offlineListener]) => Effect.sync(() => {
globalThis.window.removeEventListener("online", onlineListener)
globalThis.window.removeEventListener("offline", offlineListener)
}),
)),
),
})),
)
+60
View File
@@ -0,0 +1,60 @@
import { Context, Effect, Layer, Pipeable, Predicate, Queue, Stream } from "effect"
import * as View from "./View.js"
export const AppFocusServiceTypeId: unique symbol = Symbol.for("@effect-view/AppFocus/AppFocusService")
export type AppFocusServiceTypeId = typeof AppFocusServiceTypeId
/** Whether the app is focused, as used by Query's `refreshOnFocus`. */
export interface AppFocusService extends Pipeable.Pipeable {
readonly [AppFocusServiceTypeId]: AppFocusServiceTypeId
readonly focused: View.View<boolean>
}
export const isAppFocusService = (u: unknown): u is AppFocusService => Predicate.hasProperty(u, AppFocusServiceTypeId)
export class AppFocusServiceImpl extends Pipeable.Class implements AppFocusService {
readonly [AppFocusServiceTypeId]: AppFocusServiceTypeId = AppFocusServiceTypeId
constructor(
readonly focused: View.View<boolean>,
) {
super()
}
}
export const make = (focused: View.View<boolean>): AppFocusService => new AppFocusServiceImpl(focused)
/** Whether the app is focused, as used by Query's `refreshOnFocus`. */
export class AppFocus extends Context.Service<AppFocus, AppFocusService>()(
"@effect-view/AppFocus/AppFocus"
) {}
export const layerDummy: Layer.Layer<AppFocus> = Layer.succeed(
AppFocus,
make(View.make({
get: Effect.succeed(true),
changes: Stream.succeed(true),
})),
)
const isDocumentVisible = () => globalThis.document.visibilityState === "visible"
/** Focused while the page is visible, following the document's `visibilitychange` events. */
export const layerBrowser: Layer.Layer<AppFocus> = Layer.succeed(
AppFocus,
make(View.make({
get: Effect.sync(isDocumentVisible),
changes: Stream.concat(
Stream.sync(isDocumentVisible),
Stream.callback<boolean>(queue => Effect.acquireRelease(
Effect.sync(() => {
const listener = () => Queue.offerUnsafe(queue, isDocumentVisible())
globalThis.document.addEventListener("visibilitychange", listener)
return listener
}),
listener => Effect.sync(() => globalThis.document.removeEventListener("visibilitychange", listener)),
)),
),
})),
)
+164 -18
View File
@@ -1,12 +1,37 @@
import { Cause, Deferred, Effect, Option, type Scope } from "effect"
import { Cause, Deferred, Effect, Fiber, Layer, Option, type Scope, Stream, SubscriptionRef } from "effect"
import { AsyncResult } from "effect/reactivity"
import { describe, expect, it } from "vitest"
import * as AppConnectivity from "./AppConnectivity.js"
import * as AppFocus from "./AppFocus.js"
import * as Lens from "./Lens.js"
import * as Mutation from "./Mutation.js"
import * as QueryClient from "./QueryClient.js"
import * as View from "./View.js"
const runMutationTest = <A, E>(effect: Effect.Effect<A, E, Scope.Scope>) =>
Effect.runPromise(Effect.scoped(effect))
const runMutationTest = <A, E>(effect: Effect.Effect<A, E, QueryClient.QueryClient | Scope.Scope>) =>
Effect.runPromise(Effect.scoped(effect.pipe(
Effect.provide(QueryClient.layer.pipe(
Layer.provide(Layer.mergeAll(AppConnectivity.layerDummy, AppFocus.layerDummy)),
)),
)))
/** Runs `f` with an AppConnectivity whose online status is controlled through the returned ref. */
const withAppConnectivity = <A, E, R>(
initial: boolean,
f: (online: SubscriptionRef.SubscriptionRef<boolean>) => Effect.Effect<A, E, R>,
) => Effect.flatMap(
SubscriptionRef.make(initial),
online => Effect.provide(f(online), Layer.succeed(AppConnectivity.AppConnectivity, AppConnectivity.make(Lens.fromSubscriptionRef(online)))),
)
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 +57,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 +125,153 @@ 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"))
})
})
describe("Mutation network mode", () => {
it("in online mode, a mutation made offline is paused until the connection returns", async () => {
const result = await runMutationTest(withAppConnectivity(false, online => Effect.gen(function*() {
let calls = 0
const mutation = yield* Mutation.make({
f: (key: number) => Effect.sync(() => {
calls += 1
return `value:${key}`
}),
})
const stream = yield* mutation.mutateStream(1)
const states = yield* Effect.forkScoped(Stream.runCollect(stream))
yield* Effect.yieldNow
const whileOffline = { calls, state: yield* View.get(mutation.state) }
yield* SubscriptionRef.set(online, true)
return {
whileOffline,
states: yield* Fiber.join(states).pipe(Effect.timeout("1 second")),
calls,
}
})))
expect(result.whileOffline.calls).toBe(0)
expect(result.whileOffline.state.paused).toBe(true)
expect(result.whileOffline.state.result.waiting).toBe(true)
expect(result.calls).toBe(1)
const last = result.states.at(-1)
if (!last)
throw new Error("Expected at least one state")
expect(last.paused).toBe(false)
expect(expectSuccessValue(last)).toBe("value:1")
})
it("in always mode, mutations ignore the connection", async () => {
const result = await runMutationTest(withAppConnectivity(false, () => Effect.gen(function*() {
const mutation = yield* Mutation.make({
f: (key: number) => Effect.succeed(`value:${key}`),
networkMode: "always",
})
return yield* mutation.mutate(1).pipe(Effect.timeout("1 second"))
})))
expect(expectSuccessValue(result)).toBe("value:1")
expect(result.paused).toBe(false)
})
it("uses the client's defaultMutationNetworkMode, not defaultNetworkMode", async () => {
const result = await Effect.runPromise(Effect.scoped(Effect.gen(function*() {
const mutation = yield* Mutation.make({
f: (key: number) => Effect.succeed(`value:${key}`),
})
return yield* mutation.mutate(1).pipe(Effect.timeout("1 second"))
}).pipe(
Effect.provide(QueryClient.layerWithOptions({
defaultNetworkMode: "online",
defaultMutationNetworkMode: "always",
}).pipe(
Layer.provide(Layer.mergeAll(
Layer.succeed(AppConnectivity.AppConnectivity, AppConnectivity.make(View.make({
get: Effect.succeed(false),
changes: Stream.make(false),
}))),
AppFocus.layerDummy,
)),
)),
)))
expect(expectSuccessValue(result)).toBe("value:1")
expect(result.paused).toBe(false)
})
})
+81 -155
View File
@@ -1,7 +1,9 @@
import { Cause, type Context, Effect, Exit, type Fiber, Option, Pipeable, Predicate, PubSub, Ref, type Scope, Semaphore, Stream, SubscriptionRef } from "effect"
import { 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 * as QueryClient from "./QueryClient.js"
import type * as View from "./View.js"
export const MutationTypeId: unique symbol = Symbol.for("@effect-view/Mutation/Mutation")
@@ -11,31 +13,39 @@ export interface Mutation<in out K, out A, out E = never, in out R = never>
extends Pipeable.Pipeable {
readonly [MutationTypeId]: MutationTypeId
readonly context: Context.Context<Scope.Scope | R>
readonly context: Context.Context<Mutation.Context<R>>
readonly f: (key: K) => Effect.Effect<A, E, R>
readonly networkMode: Operation.NetworkMode
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 namespace Mutation {
export type Context<R> = Scope.Scope | QueryClient.QueryClient | QueryClient.QueryClientService.Context | R
}
export interface MutationLatestState<out K, out A, out E = never> {
readonly key: Option.Option<K>
readonly result: AsyncResult.AsyncResult<A, E>
readonly paused: boolean
}
export interface MutationState<out K, out A, out E = never> {
readonly key: Option.Some<K>
readonly result: AsyncResult.AsyncResult<A, E>
readonly paused: boolean
}
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>
readonly paused: false
}
export const isMutation = (u: unknown): u is Mutation<unknown, unknown, unknown, unknown> => Predicate.hasProperty(u, MutationTypeId)
@@ -46,135 +56,85 @@ extends Pipeable.Class implements Mutation<K, A, E, R> {
readonly [MutationTypeId]: MutationTypeId = MutationTypeId
constructor(
readonly context: Context.Context<Scope.Scope | R>,
readonly context: Context.Context<Mutation.Context<R>>,
readonly f: (key: K) => Effect.Effect<A, E, R>,
readonly networkMode: Operation.NetworkMode,
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
Mutation.Context<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,
networkMode: this.networkMode,
})
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,
}
}
}
export declare namespace make {
export interface Options<K = never, A = void, E = never, R = never> {
readonly f: (key: K) => Effect.Effect<A, E, R>
readonly networkMode?: Operation.NetworkMode
}
}
@@ -183,59 +143,25 @@ export const make = Effect.fnUntraced(function* <K = never, A = void, E = never,
): Effect.fn.Return<
Mutation<K, A, E, R>,
never,
Scope.Scope | R
Scope.Scope | QueryClient.QueryClient | R
> {
const client = yield* QueryClient.QueryClient
return new MutationImpl(
yield* Effect.context<Scope.Scope | R>(),
Context.mergeAll(
client.context,
yield* Effect.context<Scope.Scope | QueryClient.QueryClient | R>(),
),
options.f,
options.networkMode ?? client.defaultMutationNetworkMode,
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(),
paused: false,
})),
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)),
)
+5 -4
View File
@@ -4,6 +4,7 @@ import { AsyncResult } from "effect/reactivity"
import * as Form from "./Form.js"
import * as Lens from "./Lens.js"
import * as Mutation from "./Mutation.js"
import type * as QueryClient from "./QueryClient.js"
import * as View from "./View.js"
@@ -24,7 +25,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 +142,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 +156,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
>>,
@@ -204,7 +205,7 @@ export const make = Effect.fnUntraced(function* <A, I = A, RD = never, RE = neve
): Effect.fn.Return<
MutationForm<A, I, RD, RE, MA, ME, MR>,
never,
Scope.Scope | RD | RE | MR
Scope.Scope | QueryClient.QueryClient | RD | RE | MR
> {
return new MutationFormImpl(
options.schema,
+331
View File
@@ -0,0 +1,331 @@
import { Cause, type Context, Deferred, Effect, Exit, Fiber, Function, Option, Pipeable, Predicate, Stream, SubscriptionRef } from "effect"
import { AsyncResult } from "effect/reactivity"
import * as AppConnectivity from "./AppConnectivity.js"
import * as Lens from "./Lens.js"
import type * as Query from "./Query.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, in out R = never>
extends Pipeable.Pipeable {
readonly [OperationTypeId]: OperationTypeId
readonly context: Context.Context<Operation.Context<R>>
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`. Depending on the
* network mode, the Operation may be paused while `online` is false (see `NetworkMode`).
* Returns the running fiber, or `None` if it has already completed.
*/
readonly start: Effect.Effect<Option.Option<Fiber.Fiber<A, E>>>
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>>
}
export namespace Operation {
export type Context<R> = Query.Query.Context<R>
}
/** 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>
/** Whether the Operation is waiting for a network connection before running its effect. */
readonly paused: boolean
}
export interface OperationFinalState<out K, out A, out E = never> {
readonly key: K
readonly result: AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>
readonly paused: false
}
/**
* - `"online"`: an Operation waits for a network connection (see `AppConnectivity`) before running, and is `paused`
* meanwhile.
* - `"offlineFirst"`: an Operation runs once regardless of the network connection, e.g. for effects that may be served
* from a cache. If it fails while offline, it waits for a network connection, `paused` meanwhile, and runs again.
* - `"always"`: Operations ignore the network connection, e.g. for effects that do not use the network.
*/
export type NetworkMode = "online" | "offlineFirst" | "always"
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 context: Context.Context<Operation.Context<R>>,
readonly key: K,
readonly effect: Effect.Effect<A, E, R>,
readonly listener: OperationListener<K, A, E, R>,
readonly networkMode: NetworkMode,
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>>> {
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.runWithNetworkMode,
),
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)
}).pipe(
Effect.provide(this.context)
)
}
// biome-ignore lint/suspicious/useGetterReturn: the switch is exhaustive, which TypeScript checks
get runWithNetworkMode(): Effect.Effect<A, E, AppConnectivity.AppConnectivity | R> {
switch (this.networkMode) {
case "online":
return Effect.andThen(this.awaitOnline, this.effect)
case "offlineFirst":
return this.runRetryingOnReconnect
case "always":
return this.effect
}
}
/** Runs `effect`, and if it fails while offline, waits for `online`, paused meanwhile, and runs it again. */
get runRetryingOnReconnect(): Effect.Effect<A, E, AppConnectivity.AppConnectivity | R> {
return Effect.catchCause(
this.effect,
cause => Effect.gen({ self: this }, function*() {
const { online } = yield* AppConnectivity.AppConnectivity
if (Cause.hasInterruptsOnly(cause) || (yield* View.get(online)))
return yield* Effect.failCause(cause)
yield* this.awaitOnline
return yield* this.runRetryingOnReconnect
}),
)
}
/** Waits for `online`, paused meanwhile. */
get awaitOnline(): Effect.Effect<void, never, AppConnectivity.AppConnectivity> {
return Effect.gen({ self: this }, function*() {
const { online } = yield* AppConnectivity.AppConnectivity
if (yield* View.get(online))
return
yield* this.update(state => ({ ...state, paused: true }))
yield* Stream.runHead(Stream.filter(View.changes(online), Function.identity))
yield* this.update(state => ({ ...state, paused: false }))
})
}
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>
readonly networkMode: NetworkMode
}
}
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>, never, Operation.Context<R>> {
return new OperationImpl(
yield* Effect.context<Operation.Context<R>>(),
options.key,
options.effect,
options.listener,
options.networkMode,
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,
paused: false,
})),
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>, never, Operation.Context<R>> {
const result = yield* Deferred.make<OperationFinalState<K, A, E>>()
yield* Deferred.succeed(result, options.state)
return new OperationImpl(
yield* Effect.context<Operation.Context<R>>(),
options.key,
options.effect,
{
onChange: () => Effect.void,
onSettle: () => Effect.void,
onInterrupt: () => Effect.void,
},
"always",
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,
}),
}),
paused: false,
})
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),
paused: false,
}),
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,
}),
}),
paused: false,
}),
})
@@ -0,0 +1,440 @@
import { Deferred, Effect, Fiber, Layer, Option, Schedule, type Scope, Stream, SubscriptionRef } from "effect"
import { AsyncResult } from "effect/reactivity"
import { describe, expect, it } from "vitest"
import * as AppConnectivity from "./AppConnectivity.js"
import * as AppFocus from "./AppFocus.js"
import * as Lens from "./Lens.js"
import * as Query from "./Query.js"
import * as QueryClient from "./QueryClient.js"
import * as View from "./View.js"
const runQueryTest = <A, E>(effect: Effect.Effect<A, E, QueryClient.QueryClient | Scope.Scope>) =>
Effect.runPromise(Effect.scoped(effect.pipe(
Effect.provide(QueryClient.layer.pipe(
Layer.provide(Layer.mergeAll(AppConnectivity.layerDummy, AppFocus.layerDummy)),
)),
)))
/** Runs `f` with an AppConnectivity whose online status is controlled through the returned ref. */
const withAppConnectivity = <A, E, R>(
initial: boolean,
f: (online: SubscriptionRef.SubscriptionRef<boolean>) => Effect.Effect<A, E, R>,
) => Effect.flatMap(
SubscriptionRef.make(initial),
online => Effect.provide(f(online), Layer.succeed(AppConnectivity.AppConnectivity, AppConnectivity.make(Lens.fromSubscriptionRef(online)))),
)
/** Runs `f` with an AppFocus whose focus is controlled through the returned ref. */
const withAppFocus = <A, E, R>(
initial: boolean,
f: (focused: SubscriptionRef.SubscriptionRef<boolean>) => Effect.Effect<A, E, R>,
) => Effect.flatMap(
SubscriptionRef.make(initial),
focused => Effect.provide(f(focused), Layer.succeed(AppFocus.AppFocus, AppFocus.make(Lens.fromSubscriptionRef(focused)))),
)
const staticKey = <K>(key: K): View.View<K> => View.make({
get: Effect.succeed(key),
changes: Stream.make(key),
})
describe("Query network mode", () => {
it("in online mode, a request made offline is paused until the connection returns", async () => {
const result = await runQueryTest(withAppConnectivity(false, online => Effect.gen(function*() {
let calls = 0
const query = yield* Query.make({
key: staticKey(1),
f: (id: number) => Effect.sync(() => {
calls += 1
return `value:${id}`
}),
})
const stream = yield* query.fetchStream(1)
const states = yield* Effect.forkScoped(Stream.runCollect(stream))
yield* Effect.yieldNow
const whileOffline = { calls, state: yield* View.get(query.state) }
yield* SubscriptionRef.set(online, true)
return {
whileOffline,
states: yield* Fiber.join(states).pipe(Effect.timeout("1 second")),
calls,
}
})))
expect(result.whileOffline.calls).toBe(0)
expect(result.whileOffline.state.paused).toBe(true)
expect(result.whileOffline.state.result.waiting).toBe(true)
expect(result.calls).toBe(1)
// The stream starts from the current state, which is already paused
expect(result.states.map(s => [s.result._tag, s.result.waiting, s.paused])).toEqual([
["Initial", true, true],
["Initial", true, false],
["Success", false, false],
])
})
it("in always mode, requests ignore the connection", async () => {
const result = await runQueryTest(withAppConnectivity(false, () => Effect.gen(function*() {
const query = yield* Query.make({
key: staticKey(1),
f: (id: number) => Effect.succeed(`value:${id}`),
networkMode: "always",
})
return yield* query.fetch(1).pipe(Effect.timeout("1 second"))
})))
expect(AsyncResult.isSuccess(result.result)).toBe(true)
expect(result.paused).toBe(false)
})
it("interrupting a paused request leaves it neither paused nor waiting", async () => {
const result = await runQueryTest(withAppConnectivity(false, () => Effect.gen(function*() {
const query = yield* Query.make({
key: staticKey(1),
f: (id: number) => Effect.succeed(`value:${id}`),
})
const stream = yield* query.fetchStream(1)
yield* Effect.yieldNow
const operation = yield* View.get(query.operation)
if (Option.isSome(operation))
yield* operation.value.interrupt
return {
states: yield* Stream.runCollect(stream).pipe(Effect.timeout("1 second")),
state: yield* View.get(query.state),
}
})))
expect(result.state.paused).toBe(false)
expect(result.state.result.waiting).toBe(false)
expect(result.states.at(-1)?.paused).toBe(false)
})
it("refreshes a stale query when the connection returns", async () => {
const result = await runQueryTest(withAppConnectivity(true, online => Effect.gen(function*() {
let calls = 0
const query = yield* Query.make({
key: staticKey(1),
f: (id: number) => Effect.sync(() => {
calls += 1
return `value:${id}:${calls}`
}),
staleTime: "0 millis",
}).pipe(Query.thenRun)
yield* Effect.sleep("10 millis")
const afterStart = calls
yield* SubscriptionRef.set(online, false)
yield* Effect.sleep("10 millis")
const whileOffline = calls
yield* SubscriptionRef.set(online, true)
yield* Effect.sleep("10 millis")
return { afterStart, whileOffline, afterReconnect: calls, state: yield* View.get(query.state) }
})))
expect(result).toMatchObject({ afterStart: 1, whileOffline: 1, afterReconnect: 2 })
expect(AsyncResult.isSuccess(result.state.result) && result.state.result.value).toBe("value:1:2")
})
it("reconnecting resumes a paused request instead of restarting it", async () => {
const result = await runQueryTest(withAppConnectivity(false, online => Effect.gen(function*() {
let started = 0
let completed = 0
const query = yield* Query.make({
key: staticKey(1),
f: (id: number) => Effect.gen(function*() {
started += 1
yield* Effect.sleep("5 millis")
completed += 1
return `value:${id}`
}),
}).pipe(Query.thenRun)
yield* Effect.sleep("10 millis")
const pausedWhileOffline = (yield* View.get(query.state)).paused
yield* SubscriptionRef.set(online, true)
yield* Effect.sleep("30 millis")
return { pausedWhileOffline, started, completed, state: yield* View.get(query.state) }
})))
expect(result.pausedWhileOffline).toBe(true)
expect(result.started).toBe(1)
expect(result.completed).toBe(1)
expect(AsyncResult.isSuccess(result.state.result)).toBe(true)
})
it("a focus or reconnect refresh does not restart a request that is already in flight", async () => {
const result = await runQueryTest(withAppConnectivity(true, online => withAppFocus(true, focused => Effect.gen(function*() {
let started = 0
const query = yield* Query.make({
key: staticKey(1),
f: () => Effect.gen(function*() {
started += 1
yield* Effect.sleep("40 millis")
return started
}),
staleTime: "0 millis",
}).pipe(Query.thenRun)
yield* Effect.sleep("10 millis")
yield* SubscriptionRef.set(focused, false)
yield* SubscriptionRef.set(focused, true)
yield* SubscriptionRef.set(online, false)
yield* SubscriptionRef.set(online, true)
yield* Effect.sleep("100 millis")
return { started, state: yield* View.get(query.state) }
}))))
expect(result.started).toBe(1)
expect(AsyncResult.isSuccess(result.state.result)).toBe(true)
})
it("does not refresh on reconnect when refreshOnReconnect is false", async () => {
const calls = await runQueryTest(withAppConnectivity(true, online => Effect.gen(function*() {
let calls = 0
yield* Query.make({
key: staticKey(1),
f: () => Effect.sync(() => { calls += 1 }),
staleTime: "0 millis",
refreshOnReconnect: false,
}).pipe(Query.thenRun)
yield* Effect.sleep("10 millis")
yield* SubscriptionRef.set(online, false)
yield* SubscriptionRef.set(online, true)
yield* Effect.sleep("10 millis")
return calls
})))
expect(calls).toBe(1)
})
it("refreshes a stale query when the app becomes focused, but not at startup", async () => {
const result = await runQueryTest(withAppFocus(true, focused => Effect.gen(function*() {
let calls = 0
const gate = yield* Deferred.make<void>()
yield* Query.make({
key: staticKey(1),
f: () => Effect.andThen(Effect.sync(() => { calls += 1 }), Deferred.succeed(gate, undefined)),
staleTime: "0 millis",
}).pipe(Query.thenRun)
yield* Deferred.await(gate)
yield* Effect.sleep("10 millis")
const afterStart = calls
yield* SubscriptionRef.set(focused, false)
yield* SubscriptionRef.set(focused, true)
yield* Effect.sleep("10 millis")
return { afterStart, afterFocus: calls }
})))
expect(result).toEqual({ afterStart: 1, afterFocus: 2 })
})
it("in offlineFirst mode, a request made offline runs right away", async () => {
const result = await runQueryTest(withAppConnectivity(false, () => Effect.gen(function*() {
const query = yield* Query.make({
key: staticKey(1),
f: (id: number) => Effect.succeed(`value:${id}`),
networkMode: "offlineFirst",
})
return yield* query.fetch(1).pipe(Effect.timeout("1 second"))
})))
expect(AsyncResult.isSuccess(result.result)).toBe(true)
expect(result.paused).toBe(false)
})
it("in offlineFirst mode, a request that fails offline is paused, then runs again when the connection returns", async () => {
const result = await runQueryTest(withAppConnectivity(false, online => Effect.gen(function*() {
let calls = 0
const query = yield* Query.make({
key: staticKey(1),
f: (id: number) => Effect.suspend(() => {
calls += 1
return calls === 1 ? Effect.fail("offline") : Effect.succeed(`value:${id}`)
}),
networkMode: "offlineFirst",
})
const stream = yield* query.fetchStream(1)
const states = yield* Effect.forkScoped(Stream.runCollect(stream))
yield* Effect.yieldNow
const whileOffline = { calls, state: yield* View.get(query.state) }
yield* SubscriptionRef.set(online, true)
return {
whileOffline,
states: yield* Fiber.join(states).pipe(Effect.timeout("1 second")),
calls,
}
})))
expect(result.whileOffline.calls).toBe(1)
expect(result.whileOffline.state.paused).toBe(true)
expect(result.whileOffline.state.result.waiting).toBe(true)
expect(result.calls).toBe(2)
const last = result.states.at(-1)
expect(last && AsyncResult.isSuccess(last.result) && last.result.value).toBe("value:1")
expect(last?.paused).toBe(false)
})
it("in offlineFirst mode, a request that fails online is not run again", async () => {
const result = await runQueryTest(withAppConnectivity(true, () => Effect.gen(function*() {
let calls = 0
const query = yield* Query.make({
key: staticKey(1),
f: (_id: number) => Effect.suspend(() => {
calls += 1
return Effect.fail("error")
}),
networkMode: "offlineFirst",
})
return { final: yield* query.fetch(1).pipe(Effect.timeout("1 second")), calls }
})))
expect(AsyncResult.isFailure(result.final.result)).toBe(true)
expect(result.calls).toBe(1)
})
it("refreshes a fresh query on focus only when refreshOnFocus is \"always\"", async () => {
const result = await runQueryTest(withAppFocus(true, focused => Effect.gen(function*() {
const calls = { true: 0, always: 0 }
const makeQuery = (refreshOnFocus: true | "always") => Query.make({
key: staticKey(refreshOnFocus),
f: () => Effect.sync(() => { calls[refreshOnFocus === true ? "true" : "always"] += 1 }),
staleTime: "1 hour",
refreshOnFocus,
}).pipe(Query.thenRun)
yield* makeQuery(true)
yield* makeQuery("always")
yield* Effect.sleep("10 millis")
yield* SubscriptionRef.set(focused, false)
yield* SubscriptionRef.set(focused, true)
yield* Effect.sleep("10 millis")
return calls
})))
expect(result).toEqual({ true: 1, always: 2 })
})
it("refreshes a fresh query on reconnect when refreshOnReconnect is \"always\"", async () => {
const calls = await runQueryTest(withAppConnectivity(true, online => Effect.gen(function*() {
let calls = 0
yield* Query.make({
key: staticKey(1),
f: () => Effect.sync(() => { calls += 1 }),
staleTime: "1 hour",
refreshOnReconnect: "always",
}).pipe(Query.thenRun)
yield* Effect.sleep("10 millis")
yield* SubscriptionRef.set(online, false)
yield* SubscriptionRef.set(online, true)
yield* Effect.sleep("10 millis")
return calls
})))
expect(calls).toBe(2)
})
})
describe("browser layers", () => {
it("AppConnectivity.layerBrowser follows the online and offline events", async () => {
const values = await Effect.runPromise(Effect.gen(function*() {
const { online } = yield* AppConnectivity.AppConnectivity
const values = yield* Effect.forkScoped(Stream.runCollect(Stream.take(View.changes(online), 3)))
yield* Effect.sleep("10 millis")
window.dispatchEvent(new Event("offline"))
window.dispatchEvent(new Event("online"))
return yield* Fiber.join(values).pipe(Effect.timeout("1 second"))
}).pipe(
Effect.provide(AppConnectivity.layerBrowser),
Effect.scoped,
))
expect(values).toEqual([navigator.onLine, false, true])
})
it("AppFocus.layerBrowser follows the document's visibility", async () => {
let visibilityState: DocumentVisibilityState = "visible"
const setVisibility = (state: DocumentVisibilityState) => {
visibilityState = state
document.dispatchEvent(new Event("visibilitychange"))
}
Object.defineProperty(document, "visibilityState", { configurable: true, get: () => visibilityState })
try {
const values = await Effect.runPromise(Effect.gen(function*() {
const { focused } = yield* AppFocus.AppFocus
const values = yield* Effect.forkScoped(Stream.runCollect(Stream.take(View.changes(focused), 3)))
yield* Effect.sleep("10 millis")
setVisibility("hidden")
setVisibility("visible")
return yield* Fiber.join(values).pipe(Effect.timeout("1 second"))
}).pipe(
Effect.provide(AppFocus.layerBrowser),
Effect.scoped,
))
expect(values).toEqual([true, false, true])
}
finally {
// Restores jsdom's own getter, from the prototype
delete (document as { visibilityState?: unknown }).visibilityState
}
})
})
describe("withScheduledRefresh", () => {
const countRefreshes = (focused: boolean, options?: Query.withScheduledRefresh.Options) =>
runQueryTest(withAppFocus(focused, () => Effect.gen(function*() {
let calls = 0
yield* Query.make({
key: staticKey(1),
f: () => Effect.sync(() => { calls += 1 }),
staleTime: "0 millis",
}).pipe(
Query.thenRun,
Query.withScheduledRefresh(Schedule.spaced("10 millis").pipe(Schedule.upTo({ times: 2 })), options),
)
yield* Effect.sleep("100 millis")
return calls
})))
it("refreshes while the app is focused", async () => {
expect(await countRefreshes(true)).toBe(3)
})
it("skips refreshes while the app is not focused", async () => {
expect(await countRefreshes(false)).toBe(1)
})
it("keeps refreshing while the app is not focused with refreshInBackground", async () => {
expect(await countRefreshes(false, { refreshInBackground: true })).toBe(3)
})
})
+203 -3
View File
@@ -1,7 +1,9 @@
import { Effect, Schedule, type Scope, Stream } from "effect"
import { Deferred, Effect, Exit, Fiber, Layer, Option, Schedule, type Scope, Stream } from "effect"
import { AsyncResult } from "effect/reactivity"
import { TestClock } from "effect/testing"
import { describe, expect, it } from "vitest"
import * as AppConnectivity from "./AppConnectivity.js"
import * as AppFocus from "./AppFocus.js"
import * as Query from "./Query.js"
import * as QueryClient from "./QueryClient.js"
import * as View from "./View.js"
@@ -9,7 +11,9 @@ import * as View from "./View.js"
const runQueryTest = <A, E>(effect: Effect.Effect<A, E, QueryClient.QueryClient | Scope.Scope>) =>
Effect.runPromise(Effect.scoped(effect.pipe(
Effect.provide(QueryClient.layer()),
Effect.provide(QueryClient.layer.pipe(
Layer.provide(Layer.mergeAll(AppConnectivity.layerDummy, AppFocus.layerDummy)),
)),
)))
const staticKey = <K>(key: K): View.View<K> => View.make({
@@ -18,7 +22,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 +212,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 })
})
})
+225 -265
View File
@@ -1,6 +1,9 @@
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, 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 AppConnectivity from "./AppConnectivity.js"
import * as AppFocus from "./AppFocus.js"
import * as Lens from "./Lens.js"
import * as Operation from "./Operation.js"
import * as QueryClient from "./QueryClient.js"
import * as View from "./View.js"
@@ -12,38 +15,55 @@ export interface Query<in out K, out A, out E = never, in out R = never>
extends Pipeable.Pipeable {
readonly [QueryTypeId]: QueryTypeId
readonly context: Context.Context<Scope.Scope | QueryClient.QueryClient | R>
readonly context: Context.Context<Query.Context<R>>
readonly key: View.View<K>
readonly keyEquivalence: Equivalence.Equivalence<K>
readonly f: (key: K) => Effect.Effect<A, E, R>
readonly staleTime: Duration.Duration
readonly refreshOnWindowFocus: boolean
readonly networkMode: Operation.NetworkMode
readonly refreshOnFocus: RefreshOn
readonly refreshOnReconnect: RefreshOn
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>
}
export namespace Query {
export type Context<R> = Scope.Scope | QueryClient.QueryClient | QueryClient.QueryClientService.Context | R
}
export interface QueryState<out K, out A, out E = never> {
readonly key: K
readonly result: AsyncResult.AsyncResult<A, E>
/** Whether the request is waiting for a network connection before fetching. See `networkMode`. */
readonly paused: boolean
}
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>
readonly paused: false
}
/**
* When a Query refreshes its current key on an event, such as the app regaining focus or connection:
* - `true`: if the cached result is stale.
* - `"always"`: regardless of `staleTime`.
* - `false`: never.
*/
export type RefreshOn = boolean | "always"
export const isQuery = (u: unknown): u is Query<unknown, unknown, unknown, unknown> => Predicate.hasProperty(u, QueryTypeId)
@@ -52,19 +72,22 @@ extends Pipeable.Class implements Query<K, A, E, R> {
readonly [QueryTypeId]: QueryTypeId = QueryTypeId
constructor(
readonly context: Context.Context<Scope.Scope | QueryClient.QueryClient | R>,
readonly context: Context.Context<Query.Context<R>>,
readonly key: View.View<K>,
readonly keyEquivalence: Equivalence.Equivalence<K>,
readonly f: (key: K) => Effect.Effect<A, E, R>,
readonly staleTime: Duration.Duration,
readonly refreshOnWindowFocus: boolean,
readonly networkMode: Operation.NetworkMode,
readonly refreshOnFocus: RefreshOn,
readonly refreshOnReconnect: RefreshOn,
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,33 +96,22 @@ 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,
)
: Effect.void
),
Effect.catchDefect(() => Effect.void),
),
this.refreshOnFocus
? Effect.flatMap(AppFocus.AppFocus, ({ focused }) => Stream.runForEach(
becomesTrue(focused),
() => this.refreshUnlessWaiting(this.refreshOnFocus === "always"),
))
: Effect.void,
this.refreshOnReconnect
? Effect.flatMap(AppConnectivity.AppConnectivity, ({ online }) => Stream.runForEach(
becomesTrue(online),
() => this.refreshUnlessWaiting(this.refreshOnReconnect === "always"),
))
: Effect.void,
], { concurrency: "unbounded" }).pipe(
Effect.ignore,
this.runSemaphore.withPermits(1),
@@ -108,219 +120,156 @@ 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),
paused: 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),
paused: 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>,
): Effect.Effect<
View.View<QueryState<K, A, E>>,
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),
}))
/**
* Refreshes the current key, unless a request is already in flight. Used when the app regains focus or connection:
* a request that is running keeps running, and one paused for lack of a connection resumes by itself. With
* `ignoreStaleTime`, a fresh cached result is refreshed too.
*/
refreshUnlessWaiting(ignoreStaleTime: boolean): Effect.Effect<void, Cause.NoSuchElementError, Query.Context<R>> {
return Effect.flatMap(
Lens.get(this.state),
state => state.result.waiting
? Effect.void
: Effect.asVoid(this.start(this.previousForLatestKey, ignoreStaleTime)),
)
}
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),
paused: 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 unless `ignoreStaleTime` is true.
*/
start(
previous: QueryState<K, A, E>,
previous: Effect.Effect<QueryState<K, A, E>>,
ignoreStaleTime = false,
): Effect.Effect<
View.View<QueryState<K, A, E>>,
never,
Scope.Scope | R
Operation.Operation<K, A, E, QueryClient.QueryClient | R>,
Cause.NoSuchElementError,
Query.Context<R>
> {
return Effect.gen({ self: this }, function*() {
const state = yield* makeQueryStateLens(previous)
yield* this.interrupt
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,
}),
}),
}
)),
const { key, result } = yield* previous
const entry = yield* this.getCacheEntry(key)
const isFresh = !ignoreStaleTime
&& Option.isSome(entry)
&& !(yield* QueryClient.isQueryClientCacheEntryStale(entry.value))
Effect.onExit(this.f(previous.key), exit => Effect.gen({ self: this }, function*() {
const fiberId = yield* Effect.fiberId
const fiber = yield* Lens.get(this.fiber)
const effect = Effect.tap(this.f(key), v => this.setCacheEntry(key, AsyncResult.success(v)))
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
})
if (Option.isSome(entry) && isFresh) {
const state: QueryFinalState<K, A, E> = {
key,
result: entry.value.result as AsyncResult.Success<A, E>,
paused: false,
}
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,
networkMode: this.networkMode,
})
yield* Lens.set(this.operation, Option.some(operation))
yield* operation.start
return operation
}
}).pipe(
this.startSemaphore.withPermits(1),
)
}
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>>
/** 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,
)
}
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 {
@@ -377,7 +326,10 @@ export declare namespace make {
readonly f: (key: K) => Effect.Effect<A, E, R>
readonly staleTime?: Duration.Input
readonly refreshOnWindowFocus?: boolean
readonly networkMode?: Operation.NetworkMode
readonly refreshOnFocus?: RefreshOn
/** Defaults to the client's `defaultRefreshOnReconnect`, and to `false` in `"always"` network mode. */
readonly refreshOnReconnect?: RefreshOn
}
}
@@ -389,27 +341,42 @@ export const make = Effect.fnUntraced(function* <K, A, E = never, R = never>(
Scope.Scope | QueryClient.QueryClient | R
> {
const client = yield* QueryClient.QueryClient
const networkMode = options.networkMode ?? client.defaultNetworkMode
return new QueryImpl(
yield* Effect.context<Scope.Scope | QueryClient.QueryClient | R>(),
Context.mergeAll(
client.context,
yield* Effect.context<Scope.Scope | QueryClient.QueryClient | R>(),
),
options.key,
options.keyEquivalence ?? Equal.asEquivalence(),
options.f,
options.staleTime ? yield* Effect.fromOption(Duration.fromInput(options.staleTime)) : client.defaultStaleTime,
options.refreshOnWindowFocus ?? client.defaultRefreshOnWindowFocus,
networkMode,
options.refreshOnFocus ?? client.defaultRefreshOnFocus,
options.refreshOnReconnect ?? (networkMode === "always" ? false : client.defaultRefreshOnReconnect),
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),
paused: 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),
)
})
/** Emits each time `view` becomes true, after its current value. */
const becomesTrue = (view: View.View<boolean>): Stream.Stream<true> => View.changes(view).pipe(
Stream.changes,
Stream.drop(1),
Stream.filter((value): value is true => value),
)
export const thenRun = <K, A, E = never, R = never, E2 = never, R2 = never>(
self: Effect.Effect<Query<K, A, E, R>, E2, R2>,
): Effect.Effect<Query<K, A, E, R>, E2, Scope.Scope | R2> => Effect.tap(
@@ -417,8 +384,18 @@ export const thenRun = <K, A, E = never, R = never, E2 = never, R2 = never>(
query => Effect.forkScoped(query.run),
)
export declare namespace withScheduledRefresh {
export interface Options {
/**
* Whether to keep refreshing while the app is not focused (see `AppFocus`). Defaults to `false`: a refresh due
* while the app is not focused is skipped, but still counts as a recurrence of the schedule.
*/
readonly refreshInBackground?: boolean
}
}
/**
* Refreshes the Query on a schedule and returns it.
* Refreshes the Query on a schedule and returns it. By default, refreshes are skipped while the app is not focused.
*
* @example Refresh every five minutes, starting after five minutes
* ```ts
@@ -437,62 +414,45 @@ export const thenRun = <K, A, E = never, R = never, E2 = never, R2 = never>(
* ),
* )
* ```
*
* @example Keep refreshing while the app is in the background
* ```ts
* yield* Query.make(options).pipe(
* Query.thenRun,
* Query.withScheduledRefresh(Schedule.spaced("5 minutes"), { refreshInBackground: true }),
* )
* ```
*/
export const withScheduledRefresh: {
<Output, Error, Env>(
schedule: Schedule.Schedule<Output, unknown, Error, Env>,
options?: withScheduledRefresh.Options,
): <K, A, E, R, E2, R2>(
self: Effect.Effect<Query<K, A, E, R>, E2, R2>,
) => Effect.Effect<Query<K, A, E, R>, E2, Scope.Scope | Env | R2>
<K, A, E, R, E2, R2, Output, Error, Env>(
self: Effect.Effect<Query<K, A, E, R>, E2, R2>,
schedule: Schedule.Schedule<Output, unknown, Error, Env>,
options?: withScheduledRefresh.Options,
): Effect.Effect<Query<K, A, E, R>, E2, Scope.Scope | Env | R2>
} = Function.dual(2, <K, A, E, R, E2, R2, Output, Error, Env>(
} = Function.dual(args => Effect.isEffect(args[0]), <K, A, E, R, E2, R2, Output, Error, Env>(
self: Effect.Effect<Query<K, A, E, R>, E2, R2>,
schedule: Schedule.Schedule<Output, unknown, Error, Env>,
options?: withScheduledRefresh.Options,
) => Effect.tap(
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),
),
query => Effect.forkScoped(Effect.schedule(
options?.refreshInBackground
? query.refresh
: Effect.flatMap(
Effect.provide(
Effect.flatMap(AppFocus.AppFocus, ({ focused }) => View.get(focused)),
query.context,
),
}),
)
}
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)),
)
focused => focused
? Effect.asVoid(query.refresh)
: Effect.void,
),
schedule,
)),
))
+76 -9
View File
@@ -1,6 +1,10 @@
import { type Cause, Context, DateTime, Duration, Effect, Equal, Equivalence, Hash, HashMap, Layer, type Option, Pipeable, Predicate, Schedule, type Scope, Semaphore, SubscriptionRef } from "effect"
import type { AsyncResult } from "effect/reactivity"
import * as AppConnectivity from "./AppConnectivity.js"
import * as AppFocus from "./AppFocus.js"
import * as Lens from "./Lens.js"
import type * as Operation from "./Operation.js"
import type * as Query from "./Query.js"
import type * as View from "./View.js"
@@ -10,12 +14,18 @@ export type QueryClientServiceTypeId = typeof QueryClientServiceTypeId
export interface QueryClientService extends Pipeable.Pipeable {
readonly [QueryClientServiceTypeId]: QueryClientServiceTypeId
readonly context: Context.Context<QueryClientService.Context>
readonly cache: View.View<HashMap.HashMap<QueryClientCacheKey, QueryClientCacheEntry>>
readonly cacheGcTime: Duration.Duration
readonly defaultStaleTime: Duration.Duration
readonly defaultRefreshOnWindowFocus: boolean
readonly defaultNetworkMode: Operation.NetworkMode
readonly defaultRefreshOnFocus: Query.RefreshOn
readonly defaultRefreshOnReconnect: Query.RefreshOn
readonly defaultMutationNetworkMode: Operation.NetworkMode
readonly run: Effect.Effect<void, Cause.NoSuchElementError>
getCacheEntry(key: QueryClientCacheKey): Effect.Effect<Option.Option<QueryClientCacheEntry>>
setCacheEntry(
key: QueryClientCacheKey,
@@ -26,9 +36,9 @@ export interface QueryClientService extends Pipeable.Pipeable {
invalidateCacheEntry(key: QueryClientCacheKey): Effect.Effect<void>
}
export class QueryClient extends Context.Service<QueryClient, QueryClientService>()(
"@effect-view/QueryClient/QueryClient"
) {}
export namespace QueryClientService {
export type Context = AppConnectivity.AppConnectivity | AppFocus.AppFocus
}
export class QueryClientServiceImpl
extends Pipeable.Class
@@ -36,10 +46,16 @@ implements QueryClientService {
readonly [QueryClientServiceTypeId]: QueryClientServiceTypeId = QueryClientServiceTypeId
constructor(
readonly context: Context.Context<QueryClientService.Context>,
readonly cache: Lens.Lens<HashMap.HashMap<QueryClientCacheKey, QueryClientCacheEntry>>,
readonly cacheGcTime: Duration.Duration,
readonly defaultStaleTime: Duration.Duration,
readonly defaultRefreshOnWindowFocus: boolean,
readonly defaultNetworkMode: Operation.NetworkMode,
readonly defaultRefreshOnFocus: Query.RefreshOn,
readonly defaultRefreshOnReconnect: Query.RefreshOn,
readonly defaultMutationNetworkMode: Operation.NetworkMode,
readonly runSemaphore: Semaphore.Semaphore,
) {
super()
@@ -99,18 +115,36 @@ export declare namespace make {
export interface Options {
readonly cacheGcTime?: Duration.Input
readonly defaultStaleTime?: Duration.Input
readonly defaultRefreshOnWindowFocus?: boolean
/** The default network mode of queries. */
readonly defaultNetworkMode?: Operation.NetworkMode
readonly defaultRefreshOnFocus?: Query.RefreshOn
readonly defaultRefreshOnReconnect?: Query.RefreshOn
/** The default network mode of mutations. */
readonly defaultMutationNetworkMode?: Operation.NetworkMode
}
}
export const make = Effect.fnUntraced(function* (
options: make.Options = {}
): Effect.fn.Return<QueryClientService, Cause.NoSuchElementError, never> {
): Effect.fn.Return<
QueryClientService,
Cause.NoSuchElementError,
QueryClientService.Context
> {
return new QueryClientServiceImpl(
Context.empty().pipe(
Context.add(AppConnectivity.AppConnectivity, yield* AppConnectivity.AppConnectivity),
Context.add(AppFocus.AppFocus, yield* AppFocus.AppFocus),
),
Lens.fromSubscriptionRef(yield* SubscriptionRef.make(HashMap.empty<QueryClientCacheKey, QueryClientCacheEntry>())),
yield* Effect.fromOption(Duration.fromInput(options.cacheGcTime ?? "5 minutes")),
yield* Effect.fromOption(Duration.fromInput(options.defaultStaleTime ?? "0 minutes")),
options.defaultRefreshOnWindowFocus ?? true,
options.defaultNetworkMode ?? "online",
options.defaultRefreshOnFocus ?? true,
options.defaultRefreshOnReconnect ?? true,
options.defaultMutationNetworkMode ?? "online",
yield* Semaphore.make(1),
)
})
@@ -122,7 +156,40 @@ export const thenRun = <E = never, R = never>(
client => Effect.forkScoped(client.run),
)
export const layer = (options?: make.Options) => Layer.effect(QueryClient, thenRun(make(options)))
export class QueryClient extends Context.Service<QueryClient, QueryClientService>()(
"@effect-view/QueryClient/QueryClient"
) {}
export const layer: Layer.Layer<
QueryClient,
Cause.NoSuchElementError,
QueryClientService.Context
> = Layer.effect(QueryClient, thenRun(make()))
export const layerWithOptions = (options: make.Options): Layer.Layer<
QueryClient,
Cause.NoSuchElementError,
QueryClientService.Context
> => Layer.effect(QueryClient, thenRun(make(options)))
export const layerBrowser: Layer.Layer<
QueryClient,
Cause.NoSuchElementError,
never
> = layer.pipe(
Layer.provide(AppConnectivity.layerBrowser),
Layer.provide(AppFocus.layerBrowser),
)
export const layerBrowserWithOptions = (options: make.Options): Layer.Layer<
QueryClient,
Cause.NoSuchElementError,
never
> => layerWithOptions(options).pipe(
Layer.provide(AppConnectivity.layerBrowser),
Layer.provide(AppFocus.layerBrowser),
)
export const QueryClientCacheKeyTypeId: unique symbol = Symbol.for("@effect-view/QueryClient/QueryClientCacheKey")
+3
View File
@@ -1,3 +1,5 @@
export * as AppConnectivity from "./AppConnectivity.js"
export * as AppFocus from "./AppFocus.js"
export * as Async from "./Async.js"
export * as Component from "./Component.js"
export * as Form from "./Form.js"
@@ -6,6 +8,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"
+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>
+1 -1
View File
@@ -5,7 +5,7 @@ import { QueryClient, ReactRuntime } from "effect-view"
export const layer = Layer.empty.pipe(
Layer.provideMerge(QueryClient.layer()),
Layer.provideMerge(QueryClient.layerBrowser),
Layer.provideMerge(DateTime.layerCurrentZoneLocal),
Layer.provideMerge(Clipboard.layer),
Layer.provideMerge(Geolocation.layer),