Elliott/query live take two

.changeset/cuddly-rats-unite.md

@@ -0,0 +1,5 @@
+---
+'@sveltejs/kit': patch
+---
+
+breaking: on the server, make the promise returned from `refresh` represent adding the refresh to the map, not the time it takes to run the remote function

.changeset/wet-rings-repair.md

@@ -0,0 +1,5 @@
+---
+'@sveltejs/kit': minor
+---
+
+feat: experimental `query.live` function

documentation/docs/20-core-concepts/60-remote-functions.md

@@ -274,6 +274,45 @@ export const getWeather = query.batch(v.string(), async (cityIds) => {
 {/if}
 ```
 
+## query.live
+
+`query.live` is for accessing real-time data from the server. It behaves similarly to `query`, but the callback — typically an async [generator function](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Operators/function*) — returns an `AsyncIterable`:
+
+```js
+import { query } from '$app/server';
+
+export const getTime = query.live(async function* () {
+	while (true) {
+		yield new Date();
+		await new Promise((f) => setTimeout(f, 1000));
+	}
+});
+```
+
+During server-side rendering, `await getTime()` returns the first yielded value then closes the iterator. This initial value is serialized and reused during hydration.
+
+On the client, the query stays connected while it's actively used in a component. Multiple instances share a connection. When there are no active uses left, the stream disconnects and server-side iteration is stopped.
+
+Live queries expose a `connected` property and `reconnect()` method:
+
+```svelte
+<script>
+	import { getTime } from './time.remote.js';
+
+	const time = getTime();
+</script>
+
+<p>{await time}</p>
+<p>connected: {time.connected}</p>
+<button onclick={() => time.reconnect()}>Reconnect</button>
+```
+
+If the connection drops, `connected` becomes `false`. SvelteKit will attempt to reconnect passively, with exponential backoff, and actively if `navigator.onLine` goes from `false` to `true`.
+
+Unlike `query`, live queries do not have a `refresh()` method, as they are self-updating.
+
+As with `query` and `query.batch`, call `.run()` outside render when you need imperative access. For live queries, `run()` returns a `Promise<AsyncIterator<T>>`.
+
 ## form
 
 The `form` function makes it easy to write data to the server. It takes a callback that receives `data` constructed from the submitted [`FormData`](https://developer.mozilla.org/en-US/docs/Web/API/FormData)...
@@ -974,6 +1013,29 @@ export const updatePost = form(
 
 Because queries are keyed based on their arguments, `await getPost(post.id).set(result)` on the server knows to look up the matching `getPost(id)` on the client to update it. The same goes for `getPosts().refresh()` -- it knows to look up `getPosts()` with no argument on the client.
 
+### Reconnecting live queries in mutations
+
+Single-flight mutations can also reconnect `query.live` instances. In a `form`/`command` handler, call `.reconnect()` on the live query resource you want to reconnect:
+
+```js
+import * as v from 'valibot';
+import { form, query } from '$app/server';
+
+export const getNotifications = query.live(v.string(), async function* (userId) {
+	while (true) {
+		yield await db.notifications(userId);
+		await wait(1000);
+	}
+});
+
+export const markAllRead = form(v.object({ userId: v.string() }), async ({ userId }) => {
+	// mutation logic...
+	+++getNotifications(userId).reconnect();+++
+});
+```
+
+This schedules a reconnect for the matching active client instances and applies it as part of the mutation response (i.e. in the same flight as the form/command result).
+
 ### Client-requested refreshes
 
 Unfortunately, life isn't always as simple as the preceding example. The server always knows which query _functions_ to update, but it may not know which specific query _instances_ to update. For example, if `getPosts({ filter: 'author:santa' })` is rendered on the client, calling `getPosts().refresh()` in the server handler won't update it. You'd need to call `getPosts({ filter: 'author:santa' }).refresh()` instead — but how could you know which specific combinations of filters are currently rendered on the client, especially if your query argument is more complicated than an object with just one key?

packages/kit/src/exports/internal/remote-functions.js

@@ -1,7 +1,7 @@
 /** @import { RemoteInternals } from 'types' */
 
 /** @type {RemoteInternals['type'][]} */
-const types = ['command', 'form', 'prerender', 'query', 'query_batch'];
+const types = ['command', 'form', 'prerender', 'query', 'query_batch', 'query_live'];
 
 /**
  * @param {Record<string, any>} module

packages/kit/src/exports/public.d.ts

@@ -1452,7 +1452,9 @@ export interface Page<
  */
 export type ParamMatcher = (param: string) => boolean;
 
-export type RequestedResult<T> = Iterable<T> &
+export type RequestedResult<T> = QueryRequestedResult<T> | LiveQueryRequestedResult<T>;
+
+export type QueryRequestedResult<T> = Iterable<T> &
 	AsyncIterable<T> & {
 		/**
 		 * Call `refresh` on all queries selected by this `requested` invocation.
@@ -1468,6 +1470,22 @@ export type RequestedResult<T> = Iterable<T> &
 		refreshAll: () => Promise<void>;
 	};
 
+export type LiveQueryRequestedResult<T> = Iterable<T> &
+	AsyncIterable<T> & {
+		/**
+		 * Call `reconnect` on all live queries selected by this `requested` invocation.
+		 * This is identical to:
+		 * ```ts
+		 * import { requested } from '$app/server';
+		 *
+		 * for await (const arg of requested(query, ...)) {
+		 *   query(arg).reconnect();
+		 * }
+		 * ```
+		 */
+		reconnectAll: () => Promise<void>;
+	};
+
 export interface RequestEvent<
 	Params extends AppLayoutParams<'/'> = AppLayoutParams<'/'>,
 	RouteId extends AppRouteId | null = AppRouteId | null
@@ -2079,7 +2097,7 @@ export interface ValidationError {
 }
 
 /**
- * The return value of a remote `form` function. See [Remote functions](https://svelte.dev/docs/kit/remote-functions#form) for full documentation.
+ * The type of a remote `form` function. See [Remote functions](https://svelte.dev/docs/kit/remote-functions#form) for full documentation.
  */
 export type RemoteForm<Input extends RemoteFormInput | void, Output> = {
 	/** Attachment that sets up an event handler that intercepts the form submission on the client to prevent a full page reload */
@@ -2134,7 +2152,7 @@ export type RemoteForm<Input extends RemoteFormInput | void, Output> = {
 };
 
 /**
- * The return value of a remote `command` function. See [Remote functions](https://svelte.dev/docs/kit/remote-functions#command) for full documentation.
+ * The type of a remote `command` function. See [Remote functions](https://svelte.dev/docs/kit/remote-functions#command) for full documentation.
  */
 export type RemoteCommand<Input, Output> = {
 	(arg: undefined extends Input ? Input | void : Input): Promise<Output> & {
@@ -2146,7 +2164,9 @@ export type RemoteCommand<Input, Output> = {
 
 export type RemoteQueryUpdate =
 	| RemoteQuery<any>
+	| RemoteLiveQuery<any>
 	| RemoteQueryFunction<any, any>
+	| RemoteLiveQueryFunction<any, any>
 	| RemoteQueryOverride;
 
 export type RemoteResource<T> = Promise<T> & {
@@ -2210,20 +2230,42 @@ export type RemoteQuery<T> = RemoteResource<T> & {
 	withOverride(update: (current: T) => T): RemoteQueryOverride;
 };
 
+export type RemoteLiveQuery<T> = RemoteResource<T> & {
+	/**
+	 * Returns an async iterator with live updates.
+	 * Unlike awaiting the resource directly, this can only be used _outside_ render
+	 * (i.e. in load functions, event handlers and so on)
+	 */
+	run(): AsyncGenerator<T>;
+	/** `true` if the live stream is currently connected. */
+	readonly connected: boolean;
+	/** `true` once the current live stream iterator is done. */
+	readonly done: boolean;
+	/** Reconnects the live stream immediately. */
+	reconnect(): Promise<void>;
+};
+
 export type RemoteQueryOverride = () => void;
 
 /**
- * The return value of a remote `prerender` function. See [Remote functions](https://svelte.dev/docs/kit/remote-functions#prerender) for full documentation.
+ * The type of a remote `prerender` function. See [Remote functions](https://svelte.dev/docs/kit/remote-functions#prerender) for full documentation.
  */
 export type RemotePrerenderFunction<Input, Output> = (
 	arg: undefined extends Input ? Input | void : Input
 ) => RemoteResource<Output>;
 
 /**
- * The return value of a remote `query` function. See [Remote functions](https://svelte.dev/docs/kit/remote-functions#query) for full documentation.
+ * The type of a remote `query` function. See [Remote functions](https://svelte.dev/docs/kit/remote-functions#query) for full documentation.
  */
 export type RemoteQueryFunction<Input, Output> = (
 	arg: undefined extends Input ? Input | void : Input
 ) => RemoteQuery<Output>;
 
+/**
+ * The type of a remote `query.live` function. See [Remote functions](https://svelte.dev/docs/kit/remote-functions#query.live) for full documentation.
+ */
+export type RemoteLiveQueryFunction<Input, Output> = (
+	arg: undefined extends Input ? Input | void : Input
+) => RemoteLiveQuery<Output>;
+
 export * from './index.js';

packages/kit/src/runtime/app/server/remote/command.js

@@ -56,7 +56,7 @@ export function command(validate_or_fn, maybe_fn) {
 	const fn = maybe_fn ?? validate_or_fn;
 
 	/** @type {(arg?: any) => MaybePromise<Input>} */
-	const validate = create_validator(validate_or_fn, maybe_fn);
+	const validate = create_validator(() => __, validate_or_fn, maybe_fn);
 
 	/** @type {RemoteCommandInternals} */
 	const __ = { type: 'command', id: '', name: '' };
@@ -77,7 +77,8 @@ export function command(validate_or_fn, maybe_fn) {
 			);
 		}
 
-		state.remote.refreshes ??= {};
+		state.remote.refreshes ??= new Map();
+		state.remote.reconnects ??= new Map();
 
 		const promise = Promise.resolve(
 			run_remote_function(event, state, true, () => validate(arg), fn)

packages/kit/src/runtime/app/server/remote/form.js

@@ -136,7 +136,8 @@ export function form(validate_or_fn, maybe_fn) {
 						data = validated.value;
 					}
 
-					state.remote.refreshes ??= {};
+					state.remote.refreshes ??= new Map();
+					state.remote.reconnects ??= new Map();
 
 					const issue = create_issues();
 

packages/kit/src/runtime/app/server/remote/prerender.js

@@ -75,7 +75,7 @@ export function prerender(validate_or_fn, fn_or_options, maybe_options) {
 	const fn = maybe_fn ?? validate_or_fn;
 
 	/** @type {(arg?: any) => MaybePromise<Input>} */
-	const validate = create_validator(validate_or_fn, maybe_fn);
+	const validate = create_validator(() => __, validate_or_fn, maybe_fn);
 
 	/** @type {RemotePrerenderInternals} */
 	const __ = {