feat: Support running EventQL queries.

Author: goloroden

README.md

@@ -188,6 +188,42 @@ for await (const event of client.readEvents('/books/42', {
 controller.abort();
 ```
 
+### Running EventQL Queries
+
+To run an EventQL query, call the `runEventQlQuery` function and provide the query as argument. The function returns an asynchronous iterator, which you can use e.g. inside a `for await` loop:
+
+```typescript
+for await (const row of client.runEventQlQuery(`
+  FROM e IN events
+  PROJECT INTO e
+`)) {
+  // ...
+}
+```
+
+#### Aborting a Query
+
+If you need to abort a query use `break` or `return` within the `await for` loop. However, this only works if there is currently an iteration going on.
+
+To abort the query independently of that, hand over an abort signal as second argument when calling `runEventQlQuery`, and abort the appropriate AbortController:
+
+```typescript
+const controller = new AbortController();
+
+for await (const row of client.runEventQlQuery(`
+  FROM e IN events
+  PROJECT INTO e
+`, controller.signal)) {
+  // ...
+}
+
+// Somewhere else, abort the controller, which will cause
+// reading to end.
+controller.abort();
+```
+
+*Note that each row returned by the iterator matches the projection specified in your query.*
+
 ### Observing Events
 
 To observe all events of a subject, call the `observeEvents` function with the subject as the first argument and an options object as the second argument. Set the `recursive` option to `false`. This ensures that only events of the given subject are returned, not events of nested subjects.

src/Client.ts

@@ -11,6 +11,7 @@ import { isStreamCloudEvent } from './stream/isStreamCloudEvent.js';
 import { isStreamError } from './stream/isStreamError.js';
 import { isStreamEventType } from './stream/isStreamEventType.js';
 import { isStreamHeartbeat } from './stream/isStreamHeartbeat.js';
+import { isStreamRow } from './stream/isStreamRow.js';
 import { isStreamSubject } from './stream/isStreamSubject.js';
 import { hasShapeOf } from './types/hasShapeOf.js';
 
@@ -190,6 +191,74 @@ class Client {
 		})();
 	}
 
+	public runEventQlQuery(query: string, signal?: AbortSignal): AsyncGenerator<unknown, void, void> {
+		const url = this.#getUrl('/api/v1/run-eventql-query');
+		const apiToken = this.#apiToken;
+
+		return (async function* () {
+			const internalAbortController = new AbortController();
+			const combinedSignal = signal ?? internalAbortController.signal;
+			const shouldAbortInternally = !signal;
+
+			let removeAbortListener: (() => void) | undefined;
+
+			if (signal && !signal.aborted) {
+				const onAbort = () => internalAbortController.abort();
+				signal.addEventListener('abort', onAbort, { once: true });
+				removeAbortListener = () => signal.removeEventListener('abort', onAbort);
+			}
+
+			try {
+				const response = await fetch(url, {
+					method: 'post',
+					headers: {
+						authorization: `Bearer ${apiToken}`,
+						'content-type': 'application/json',
+					},
+					body: JSON.stringify({ query }),
+					signal: combinedSignal,
+				});
+
+				if (response.status !== 200) {
+					throw new Error(
+						`Failed to run EventQL query, got HTTP status code '${response.status}', expected '200'.`,
+					);
+				}
+				if (!response.body) {
+					throw new Error('Failed to run EventQL query.');
+				}
+
+				for await (const line of readNdJsonStream(response.body, combinedSignal)) {
+					if (isStreamHeartbeat(line)) {
+						continue;
+					}
+					if (isStreamError(line)) {
+						throw new Error(`${line.payload.error}.`);
+					}
+					if (isStreamRow(line)) {
+						const row = line.payload;
+						yield row;
+						continue;
+					}
+
+					throw new Error('Failed to run EventQL query.');
+				}
+			} catch (error) {
+				if (error instanceof DOMException && error.name === 'AbortError') {
+					return;
+				}
+				throw error;
+			} finally {
+				if (removeAbortListener) {
+					removeAbortListener();
+				}
+				if (shouldAbortInternally) {
+					internalAbortController.abort();
+				}
+			}
+		})();
+	}
+
 	public observeEvents(
 		subject: string,
 		options: ObserveEventsOptions,

src/stream/StreamRow.ts

@@ -0,0 +1,6 @@
+interface StreamRow {
+	type: 'row';
+	payload: unknown;
+}
+
+export type { StreamRow };

src/stream/isStreamRow.ts

@@ -0,0 +1,22 @@
+import { hasShapeOf } from '../types/hasShapeOf.js';
+import type { StreamRow } from './StreamRow.js';
+
+const blueprint: Omit<StreamRow, 'payload'> = {
+	type: 'row',
+};
+
+const isStreamRow = (line: unknown): line is StreamRow => {
+	if (!hasShapeOf(line, blueprint)) {
+		return false;
+	}
+	if (line.type !== 'row') {
+		return false;
+	}
+	if (!('payload' in line)) {
+		return false;
+	}
+
+	return true;
+};
+
+export { isStreamRow };

test/runEventQlQuery.test.ts

@@ -0,0 +1,69 @@
+import assert from 'node:assert/strict';
+import { afterEach, beforeEach, suite, test } from 'node:test';
+import type { EventCandidate } from '../src/EventCandidate.js';
+import { EventSourcingDbContainer } from '../src/EventSourcingDbContainer.js';
+
+suite('runEventQlQuery', { timeout: 30_000 }, () => {
+	let container: EventSourcingDbContainer;
+
+	beforeEach(async () => {
+		container = new EventSourcingDbContainer();
+		await container.start();
+	});
+
+	afterEach(async () => {
+		await container.stop();
+	});
+
+	test('reads no rows if the query does not return any rows.', async (): Promise<void> => {
+		const client = container.getClient();
+
+		let didReadRows = false;
+		for await (const _row of client.runEventQlQuery('FROM e IN events PROJECT INTO e')) {
+			didReadRows = true;
+		}
+
+		assert.equal(didReadRows, false);
+	});
+
+	test('reads all rows the query returns.', async (): Promise<void> => {
+		const client = container.getClient();
+
+		const firstEvent: EventCandidate = {
+			source: 'https://www.eventsourcingdb.io',
+			subject: '/test',
+			type: 'io.eventsourcingdb.test',
+			data: {
+				value: 23,
+			},
+		};
+
+		const secondEvent: EventCandidate = {
+			source: 'https://www.eventsourcingdb.io',
+			subject: '/test',
+			type: 'io.eventsourcingdb.test',
+			data: {
+				value: 42,
+			},
+		};
+
+		await client.writeEvents([firstEvent, secondEvent]);
+
+		const rowsRead: unknown[] = [];
+		for await (const row of client.runEventQlQuery('FROM e IN events PROJECT INTO e')) {
+			rowsRead.push(row);
+		}
+
+		assert.equal(rowsRead.length, 2);
+
+		// biome-ignore lint/suspicious/noExplicitAny: Here, the cast is okay.
+		const firstRow = rowsRead[0] as any;
+		assert.equal(firstRow.id, '0');
+		assert.equal(firstRow.data.value, 23);
+
+		// biome-ignore lint/suspicious/noExplicitAny: Here, the cast is okay.
+		const secondRow = rowsRead[1] as any;
+		assert.equal(secondRow.id, '1');
+		assert.equal(secondRow.data.value, 42);
+	});
+});