---
title: Observable queries
editUrl: https://github.com/Cratis/Arc.TypeScript/edit/main/Documentation/queries/observable-queries.md
description: Serve a current snapshot and live updates from one query route, choose a source, and control paging, authorization, and subscription lifetime.
---


A task board should update when someone adds a task, without the browser polling. An observable query serves the current snapshot on an ordinary GET and streams every change to subscribers, from the same route and the same pipeline.

## Declare an observable query

The [Tasks sample](https://github.com/Cratis/Arc.TypeScript/blob/main/Samples/Tasks/Features/Tasks/Listing/Listing.ts) declares its live list as an ordinary query that returns a source:

```typescript
@query()
static observeAllTasks(tasks: Tasks): BehaviorSubject<TaskItem[]> { return tasks.observeAll(); }
```

The sample's `Tasks` service keeps a `new BehaviorSubject<TaskItem[]>([])` from RxJS and calls `next(...)` whenever a task is registered.

Arc has to know that a query is observable before it runs: snapshots, server-sent events, WebSocket admission, introspection, and generated clients all depend on it. The sample installs [generated artifact metadata](/arc/backend/typescript/proxy-generation/generated-artifact-metadata/), which reads the declared return type, so bare `@query()` is enough. Without generated metadata, say so explicitly and list the parameters:

```typescript
@query({ observable: true }, service(Tasks))
static observeAllTasks(tasks: Tasks): BehaviorSubject<TaskItem[]> { return tasks.observeAll(); }
```

## Choose a source

| Source | Snapshot behavior |
| --- | --- |
| RxJS `BehaviorSubject<T>` | Has a current value: GET answers 200 immediately, and new subscribers receive it first |
| RxJS `Subject<T>` or `Observable<T>` | No current value: GET answers 202 with `isReady: false`; `waitForFirstResult=true` subscribes until the first value |
| RxJS `ReplaySubject<T>` | No readable current value: GET answers 202 even after an emission; a waiting GET receives the buffered value |
| `AsyncIterable<T>` | No current value until the first item |
| `CurrentValueSubject<T>` (deprecated) | Legacy current/pending source; use RxJS `BehaviorSubject` or `Subject` instead |

A `BehaviorSubject` exposes its current value, including `undefined`; `Subject` and `ReplaySubject` do not. The core accepts structural subscribables and async iterables without loading RxJS at runtime, so RxJS is an optional peer dependency for consumers using only async iterables.

## When there is nothing to show

A live lookup, such as "the task with this ID", may have nothing to show yet. Two different situations look alike from the outside:

| The source | A snapshot GET answers | A subscriber receives |
| --- | --- | --- |
| Has no current value yet (`Subject`, `Observable`, an empty async iterable) | 202 with `isReady: false` | Nothing until the first emission |
| Emits `undefined` or `null` (a `BehaviorSubject<TaskItem \| undefined>` with nothing stored) | 200, `isReady: true`, no `data` property | A ready result without `data` |

In both cases the subscription stays open. When the document appears and the source emits it, the subscriber receives it like any other update, and when it disappears again the source can emit `undefined`. Emit `undefined` for "we looked and nothing is there", and leave a source silent only when you genuinely do not know yet; a client can tell the two apart by `isReady`.

An error is neither. A source that errors ends the subscription with a failed result, and a source that completes before its first value returns an error to a waiting GET. Do not turn a storage failure into an `undefined` emission.

## Subscription lifetime

Each subscription runs your query method once, after authorization and validation, and then follows the source it returned:

- The subscription owns its own service scope for its whole life. Scoped services are not shared between subscribers, and they are disposed when the subscription ends.
- The method can read the caller with `currentContext()` from `@cratis/arc.core`, which returns the subscription's execution context, and its `signal` aborts when the subscription ends. Pass it to anything that must stop with the subscriber.
- The subscription ends when the client disconnects or unsubscribes, when the source completes or errors, or when an [emission guard](/arc/backend/typescript/queries/observable-query-emission-guards/) denies an emission. Arc then cancels the source and disposes the scope.

## Read the snapshot and subscribe from the terminal

With the Tasks sample running:

```bash
curl http://127.0.0.1:3000/api/tasks/listing/observe-all-tasks
curl -N -H 'Accept: text/event-stream' http://127.0.0.1:3000/api/tasks/listing/observe-all-tasks
```

The first answers 200 with the current tasks in `data`. The second keeps the connection open and prints a `data: <query result JSON>` frame now, and another whenever you register a task. [Using observable queries with curl](/arc/backend/typescript/queries/using-observable-queries-with-curl/) covers waiting for a first result and the error codes.

## Page and sort a live list

The paging and sorting parameters work on an observable query exactly as on a snapshot query, and they apply to **every** emission:

```bash
curl -N -H 'Accept: text/event-stream' \
  'http://127.0.0.1:3000/api/tasks/listing/observe-all-tasks?pageSize=1&page=1&sortBy=title&sortDirection=desc'
```

With two tasks registered, each frame holds the second task in descending title order, and `paging` reports `{"page":1,"size":1,"totalItems":2,"totalPages":2}`. When a third task arrives, the next frame is sorted and cut again, and `totalItems` follows the whole list. A generated client's `useWithPaging(pageSize)` hook sends the same parameters.

Arc pages the arrays your source emits, in memory. Model-bound observable queries cannot return a `queryPage`; generated metadata rejects that declaration at build, and generated clients do not support observable-page results. Drizzle model-bound queries use `observe()` for complete small arrays, as MongoDB does. For larger collections, narrow the source with query arguments. Low-level `defineObservableQuery` without generated metadata or proxies can emit a `QueryPage` (including Drizzle's `observePage`); Core validates and streams each page, but the generated client cannot consume this shape. The [paging rules](/arc/backend/typescript/queries/model-bound/paging/#request-parameters) for invalid sizes and sort fields are the same as for snapshots.

## Authorize a live query

An observable query takes the same `@roles`, `@authorize`, and `@allowAnonymous` declarations as any query, and Arc checks them, with validation, when a subscription opens. A denied snapshot GET answers 401 or 403 with `isAuthorized: false`, following the [status code rules](/arc/backend/typescript/authorizing-commands-and-queries/#status-codes); a denied hub subscription receives an `Unauthorized` frame. The query method never runs for a denied caller.

That check happens **once**. Arc keeps a copy of the caller's identity for the life of the subscription and does not re-run authorization on each emission, so a role revoked or a session that expires after the subscription opened does not close it. When access must be re-checked while the stream runs, add an [emission guard](/arc/backend/typescript/queries/observable-query-emission-guards/), which sees every result before delivery and can end the subscription.

A role decides who may subscribe, not which rows they see. For a query like "my tasks", filter inside the source by the caller's identity:

```typescript title="MyTasks.ts"
import { field } from '@cratis/fundamentals';
import { authorize, currentContext, query, readModel } from '@cratis/arc.core';
import { BehaviorSubject, map, type Observable } from 'rxjs';

const tasks = new BehaviorSubject<OwnedTask[]>([]);

@readModel()
@authorize()
export class OwnedTask {
    @field(String) id!: string;
    @field(String) title!: string;
    @field(String) owner!: string;

    @query({ observable: true })
    static myTasks(): Observable<OwnedTask[]> {
        const caller = currentContext()?.principal?.id;
        return tasks.pipe(map(all => all.filter(task => task.owner === caller)));
    }
}
```

`@authorize()` turns anonymous callers away before `myTasks` runs, so `caller` is always an authenticated ID. Each subscriber gets their own filtered stream: when a task owned by `ada` is added, only Ada's subscription emits a new list with it. The filter runs in the producer, so rows the caller must not see never enter the result at all. Never leave that filtering to the client.

## Subscribe from a client

A browser can subscribe over direct server-sent events or a direct WebSocket, and the `@cratis/arc` client subscribes through generated proxies over a multiplexed hub. [Subscribe to an observable query](/arc/backend/typescript/queries/subscribing-to-observable-queries/) shows each one.

## Without decorators

`defineObservableQuery` takes an `observe` callback instead of a decorated method. This complete low-level example needs no build step on Node.js 22.19 or later, which strips types by default; run it with `node observable.ts` inside the workspace:

```typescript title="observable.ts"
import express from 'express';
import { z } from 'zod';
import { ArcServer, defineObservableQuery } from '@cratis/arc.core';
import { BehaviorSubject } from 'rxjs';
import { cratisArc } from '@cratis/arc.express';

const numbers = new BehaviorSubject<number[]>([1]);
const query = defineObservableQuery({
    name: 'Numbers',
    schema: z.object({}),
    observe: () => numbers
});
const server = new ArcServer({ observableQueries: [query] });
const app = express();
const adapter = cratisArc(server);
app.use(adapter);
const listener = app.listen(3000, '127.0.0.1');
adapter.injectWebSocket(listener);
let nextNumber = 2;
const timer = setInterval(() => numbers.next([nextNumber++]), 1000);

process.once('SIGINT', () => {
    clearInterval(timer);
    void (async () => {
        try {
            await adapter.close(listener); // Drain WebSockets and SSE, then close the listener.
        } finally { await server.dispose(); }
    })();
});
```

`curl http://127.0.0.1:3000/api/numbers` answers 200 with `data: [1]` or a later number. `observe` may resolve services through `currentServices()`.

## Recap

One route serves a snapshot and a stream. Return a source that has a current value when you have one, emit `undefined` for "nothing there", and let errors be errors. Paging and sorting apply to every emission; authorization applies once, when the subscription opens, and emission guards cover what changes after that. Filter rows by the caller inside the source.

## Next step

[Subscribe to an observable query](/arc/backend/typescript/queries/subscribing-to-observable-queries/) connects a browser or the `@cratis/arc` client to the route you just declared.

## Related

- [Subscribe to an observable query](/arc/backend/typescript/queries/subscribing-to-observable-queries/)
- [Multiplexed observable queries](/arc/backend/typescript/queries/observable-query-demultiplexer/)
- [Observable emission guards](/arc/backend/typescript/queries/observable-query-emission-guards/)
- [Testing observable queries](/arc/backend/typescript/testing/observable-queries/)
- [MongoDB change streams](/arc/backend/typescript/mongodb/observing-collections/)