Line data Source code
1 : import * as arrow from 'apache-arrow';
2 :
3 : import type { HttpClient, HttpFetchResult } from '../../platform/http/http_client.js';
4 : import { parseTrinoTimestampMs } from '../trino/arrow_conversion.js';
5 : import { buildClientFunctionRequest, followRequest, ClientFunctionRequest, requestUrl, STANDARD_OPTIONS } from './clfn_request.js';
6 : import {
7 : findClientFunction,
8 : ClientFunctionCallDescription,
9 : ClientFunctionColumn,
10 : ClientFunctionError,
11 : ClientFunction,
12 : ClientFunctionPaging,
13 : ClientFunctionPending,
14 : ClientFunctionSpec,
15 : ClientFunctionValueSource,
16 : jsonPointerSegments,
17 : pendingOf,
18 : readJsonPointer,
19 : } from './clfn_spec.js';
20 :
21 : /// Where a host of a spec is, and what every request to it carries
22 : export interface ClientFunctionHostTarget {
23 : baseUrl: URL;
24 : headers: Record<string, string>;
25 : }
26 :
27 : /// What a connection provides for its client functions
28 : export interface ClientFunctionEnvironment {
29 : /// The HTTP client
30 : http: HttpClient;
31 : /// Resolve a host that the spec names. Async, since a token may need a refresh or an exchange.
32 : resolveHost(host: string): Promise<ClientFunctionHostTarget>;
33 : /// The values of the variables in the base paths
34 : variables?(): Record<string, string>;
35 : /// Sleep while a pending result polls, for tests
36 : sleep?(ms: number, abort?: AbortSignal): Promise<void>;
37 : /// The current time in milliseconds, for tests
38 : now?(): number;
39 : }
40 :
41 : /// The result table of a call
42 : export interface ClientFunctionResult {
43 : schema: arrow.Schema;
44 : rows: unknown[][];
45 : }
46 :
47 : /// The first and the longest wait between two polls
48 1 : const POLL_MIN_MS = 1000;
49 1 : const POLL_MAX_MS = 10000;
50 : /// The value of `_poll` for a request that is repeated
51 1 : const POLL_SAME = 'same';
52 :
53 : /// The Arrow type of a column
54 1 : function columnType(type: string | undefined): arrow.DataType {
55 1 : switch (type) {
56 1 : case 'TIMESTAMPTZ': return new arrow.TimestampMillisecond('UTC');
57 1 : case 'DATE': return new arrow.DateDay();
58 1 : case 'INTEGER': return new arrow.Int32();
59 1 : case 'BIGINT': return new arrow.Int64();
60 1 : case 'DOUBLE': return new arrow.Float64();
61 1 : case 'BOOLEAN': return new arrow.Bool();
62 1 : default: return new arrow.Utf8();
63 1 : }
64 1 : }
65 :
66 : /// Convert a value to its column type, undefined if it can't be. The value then goes to `_raw`.
67 1 : function convertValue(value: unknown, type: string | undefined): unknown {
68 1 : if (value === null || value === undefined) {
69 1 : return null;
70 : }
71 1 : switch (type) {
72 1 : case 'INTEGER': {
73 1 : const n = typeof value === 'string' && value.trim() !== '' ? Number(value) : value;
74 1 : return typeof n === 'number' && Number.isInteger(n) && Math.abs(n) <= 0x7FFFFFFF ? n : undefined;
75 1 : }
76 1 : case 'BIGINT': {
77 1 : if (typeof value === 'number' && Number.isSafeInteger(value)) return value;
78 1 : return typeof value === 'string' && /^-?\d+$/.test(value) ? value : undefined;
79 : }
80 1 : case 'DOUBLE': {
81 1 : const n = typeof value === 'string' && value.trim() !== '' ? Number(value) : value;
82 1 : return typeof n === 'number' && Number.isFinite(n) ? n : undefined;
83 1 : }
84 1 : case 'BOOLEAN':
85 1 : return typeof value === 'boolean' ? value : undefined;
86 : // What the Arrow batch can read: a value that passes here and fails there would be lost
87 1 : case 'TIMESTAMPTZ':
88 1 : return typeof value === 'string' && parseTrinoTimestampMs(value) !== null ? value : undefined;
89 : // The day as written, whatever time and zone follow it
90 1 : case 'DATE': {
91 1 : const day = typeof value === 'string' ? value.slice(0, 10) : '';
92 1 : return /^\d{4}-\d{2}-\d{2}$/.test(day) && !Number.isNaN(Date.parse(day)) ? day : undefined;
93 1 : }
94 1 : case 'JSON':
95 1 : return JSON.stringify(value);
96 1 : default:
97 1 : return typeof value === 'object' ? undefined : String(value);
98 1 : }
99 1 : }
100 :
101 : /// One page of a response
102 : interface Page {
103 : /// The status
104 : status: number;
105 : /// The rows of the page, the fields as the API sent them
106 : items: unknown[];
107 : /// The fields outside the rows
108 : meta: Record<string, unknown> | null;
109 : /// The parsed body
110 : body: unknown;
111 : /// The headers
112 : headers: Headers;
113 : /// A body that doesn't hold the rows the spec promises. It ends the list and lands in `_raw`, so a spec that disagrees
114 : /// with a response doesn't look like an empty list.
115 : unexpected?: unknown;
116 : }
117 :
118 : /// Read the entry with a `rel` of a `Link` header
119 1 : function readLink(link: string | null, rel: string): string | null {
120 1 : const wanted = rel.toLowerCase();
121 1 : for (const entry of (link ?? '').split(',')) {
122 1 : const match = /<([^>]*)>\s*;\s*(.*)/.exec(entry.trim());
123 1 : const found = match === null ? null : /(?:^|;)\s*rel="?([^";]*)"?\s*(?:;|$)/i.exec(match[2]);
124 1 : if (found !== null && found[1].trim().toLowerCase() === wanted) {
125 1 : return match![1];
126 : }
127 1 : }
128 1 : return null;
129 1 : }
130 :
131 : /// A value that the spec points to
132 1 : function readSource(source: ClientFunctionValueSource | undefined, page: Page): unknown {
133 1 : if (source === undefined) {
134 0 : return undefined;
135 : }
136 1 : if ('last_row' in source) {
137 1 : const last = page.items[page.items.length - 1];
138 1 : return last === undefined ? undefined : readJsonPointer(last, source.last_row);
139 1 : }
140 1 : if ('link' in source) {
141 1 : return readLink(page.headers.get('Link'), source.link);
142 : }
143 1 : if ('header' in source) {
144 0 : return page.headers.get(source.header);
145 : }
146 1 : if ('body' in source) {
147 1 : return readJsonPointer(page.body, source.body);
148 : }
149 : // A mistake in a spec would end the list without a word otherwise
150 1 : throw new ClientFunctionError(`the spec reads a value from the unknown source ${JSON.stringify(source)}`);
151 1 : }
152 :
153 : /// Does the list end after a page? Only a documented rule ends it early, so an empty page is the default.
154 1 : function isLastPage(paging: ClientFunctionPaging, page: Page, position: number, size: number | null): boolean {
155 1 : const end = paging.end ?? 'empty_page';
156 1 : if (end === 'empty_page') {
157 1 : return page.items.length === 0;
158 : }
159 1 : if (end === 'short_page') {
160 1 : return size === null ? page.items.length === 0 : page.items.length < size;
161 : }
162 1 : if ('has_more' in end) {
163 1 : return readJsonPointer(page.body, end.has_more) !== true;
164 : }
165 1 : const total = Number(readJsonPointer(page.body, end.total));
166 1 : if (Number.isNaN(total)) {
167 0 : return page.items.length === 0;
168 : }
169 : // A total counts the pages of a page number, and the rows of an offset
170 1 : return paging.kind === 'page' ? position - (paging.start ?? 0) + 1 >= total : position + page.items.length >= total;
171 1 : }
172 :
173 : /// Is a value absent or empty?
174 1 : function isEmptyValue(value: unknown): boolean {
175 1 : return value === null || value === undefined || value === '';
176 1 : }
177 :
178 : /// The value for the next page, null on the last page
179 1 : function nextValue(paging: ClientFunctionPaging | undefined, page: Page, position: number, size: number | null): string | null {
180 1 : if (paging === undefined) {
181 1 : return null;
182 : }
183 1 : if (paging.kind === 'url' || paging.kind === 'token') {
184 1 : const value = readSource(paging.from, page);
185 1 : if (isEmptyValue(value)) {
186 1 : return null;
187 : }
188 1 : if (paging.end !== undefined && isLastPage(paging, page, position, size)) {
189 1 : return null;
190 : }
191 1 : return String(value);
192 1 : }
193 1 : if (isLastPage(paging, page, position, size)) {
194 1 : return null;
195 : }
196 : // Pages can be shorter than asked for, an offset follows the rows that arrived
197 1 : return String(paging.kind === 'offset' ? position + page.items.length : position + 1);
198 1 : }
199 :
200 : /// Parse a response body. What isn't JSON stays text, it ends up in `_raw`.
201 1 : function parseBody(text: string): unknown {
202 1 : if (text === '') {
203 1 : return null;
204 : }
205 1 : try {
206 1 : return JSON.parse(text);
207 1 : } catch {
208 1 : return text;
209 1 : }
210 1 : }
211 :
212 : /// The error message of a response
213 1 : function errorMessage(spec: ClientFunctionSpec, fn: ClientFunction, body: unknown, fallback: string): string {
214 1 : const pointers = fn.errors?.message ?? spec.errors?.message ?? [];
215 1 : for (const pointer of pointers) {
216 1 : const message = readJsonPointer(body, pointer);
217 1 : if (typeof message !== 'string' || message === '') {
218 1 : continue;
219 : }
220 1 : if (fn.errors?.message_is_json ?? spec.errors?.message_is_json) {
221 1 : const inner = parseBody(message);
222 1 : const nested = readJsonPointer(inner, '/message');
223 1 : if (typeof nested === 'string') {
224 1 : return nested;
225 : }
226 1 : }
227 1 : return message;
228 1 : }
229 1 : return typeof body === 'string' && body !== '' ? body : fallback;
230 1 : }
231 :
232 : /// Is a value an object?
233 1 : function isObject(value: unknown): value is Record<string, unknown> {
234 1 : return typeof value === 'object' && value !== null && !Array.isArray(value);
235 1 : }
236 :
237 : /// An object without the value at a path and without the objects that this leaves empty, null if nothing is left
238 1 : function withoutPath(object: Record<string, unknown>, [segment, ...rest]: string[]): Record<string, unknown> | null {
239 1 : const { [segment]: child, ...others } = object;
240 1 : const inner = rest.length > 0 && isObject(child) ? withoutPath(child, rest) : null;
241 1 : const out = inner === null ? others : { ...others, [segment]: inner };
242 1 : return Object.keys(out).length > 0 ? out : null;
243 1 : }
244 :
245 : /// Read the page of a finished response
246 1 : function readPage(fn: ClientFunction, status: number, body: unknown, headers: Headers): Page {
247 1 : if (fn.raw || fn.rows === undefined) {
248 1 : return { status, items: body === null ? [{}] : [body], meta: null, body, headers };
249 : }
250 1 : const items = readJsonPointer(body, fn.rows);
251 : // An explicit null and an empty object are an empty list
252 1 : const isEmpty = items === null || body === null || (typeof body === 'object' && Object.keys(body).length === 0);
253 1 : if (!Array.isArray(items) && !isEmpty) {
254 1 : return { status, items: [], meta: null, body, headers, unexpected: body };
255 : }
256 1 : const meta = fn.rows !== '' && isObject(body) ? withoutPath(body, jsonPointerSegments(fn.rows)) : null;
257 1 : return { status, items: Array.isArray(items) ? items : [], meta, body, headers };
258 1 : }
259 :
260 : /// The options of a call with `_next` applied, if that sets an option
261 1 : function optionsWithNext(fn: ClientFunction, options: Record<string, unknown>): Record<string, unknown> {
262 1 : const next = options['_next'] ?? null;
263 1 : const paging = fn.paging;
264 1 : if (next === null || paging === undefined || paging.kind === 'url' || paging.option === undefined) {
265 1 : return options;
266 : }
267 1 : const numeric = paging.kind === 'offset' || paging.kind === 'page';
268 1 : return { ...options, [paging.option]: numeric ? Number(next) : next };
269 1 : }
270 :
271 : /// The first request of a call, which a run sends and `EXPLAIN` prints. A followed `_next` or `_poll` URL replaces the path
272 : /// and query, the headers of options still apply.
273 1 : function firstRequest(spec: ClientFunctionSpec, fn: ClientFunction, options: Record<string, unknown>, variables: Record<string, string>): ClientFunctionRequest {
274 1 : const label = `${spec.namespace}.${fn.name}`;
275 1 : const next = options['_next'] ?? null;
276 1 : const poll = options['_poll'] ?? null;
277 : // Only a request that repeats has this value, anything else would send the request again, even a write
278 1 : const pending = pendingOf(spec, fn);
279 1 : if (poll === POLL_SAME && (pending === null || !('repeat' in pending.poll))) {
280 1 : throw new ClientFunctionError(`${label} polls a URL, _poll takes the value of its result`);
281 : }
282 1 : const followed = next !== null && fn.paging?.kind === 'url' ? next : poll !== null && poll !== POLL_SAME ? poll : null;
283 1 : if (followed === null) {
284 1 : return buildClientFunctionRequest(spec, fn, optionsWithNext(fn, options), variables);
285 : }
286 1 : if (fn.raw === true) {
287 0 : throw new ClientFunctionError(`${label} can't follow a URL`);
288 : }
289 1 : if (fn.host === undefined) {
290 0 : throw new ClientFunctionError(`${label} has no host`);
291 : }
292 1 : const headers: [string, string][] = [];
293 1 : for (const option of fn.options ?? []) {
294 1 : if (option.in === 'header' && options[option.name] !== undefined && options[option.name] !== null) {
295 1 : headers.push([option.api_name ?? option.name, String(options[option.name])]);
296 : }
297 : }
298 1 : return followRequest({ method: 'GET', host: fn.host, path: '', query: [], headers, body: null, followUrl: null }, String(followed));
299 1 : }
300 :
301 : /// Run calls of one function
302 1 : class ClientFunctionCallRun {
303 1 : readonly fn: ClientFunction;
304 1 : readonly pending: ClientFunctionPending | null;
305 1 : readonly columns: ClientFunctionColumn[];
306 :
307 1 : constructor(readonly spec: ClientFunctionSpec, description: ClientFunctionCallDescription, readonly env: ClientFunctionEnvironment, readonly abort?: AbortSignal) {
308 1 : this.fn = findClientFunction(spec, description.function);
309 1 : this.pending = pendingOf(spec, this.fn);
310 : // A raw call has no columns, its response is in `_raw`
311 1 : this.columns = this.fn.raw ? [] : (this.fn.columns ?? []);
312 1 : }
313 :
314 : /// The schema of the result
315 1 : schema(): arrow.Schema {
316 1 : const fields = this.columns.map(c => new arrow.Field(c.name, columnType(c.type), true));
317 1 : fields.push(new arrow.Field('_status', new arrow.Int32(), true));
318 1 : fields.push(new arrow.Field('_next', new arrow.Utf8(), true));
319 1 : if (this.pending !== null) {
320 1 : fields.push(new arrow.Field('_poll', new arrow.Utf8(), true));
321 : }
322 1 : fields.push(new arrow.Field('_meta', new arrow.Utf8(), true));
323 1 : fields.push(new arrow.Field('_raw', new arrow.Utf8(), true));
324 1 : return new arrow.Schema(fields);
325 1 : }
326 :
327 : /// A row with only the standard columns
328 1 : private standardRow(status: number | null, next: string | null, poll: string | null, meta: unknown, raw: unknown): unknown[] {
329 1 : const row: unknown[] = this.columns.map(() => null);
330 1 : row.push(status, next);
331 1 : if (this.pending !== null) {
332 1 : row.push(poll);
333 : }
334 1 : row.push(meta === null || meta === undefined ? null : JSON.stringify(meta));
335 1 : row.push(raw === null || raw === undefined ? null : JSON.stringify(raw));
336 1 : return row;
337 1 : }
338 :
339 : /// A row of an item: the listed fields in their columns, everything else in `_raw`
340 1 : private itemRow(item: unknown, page: Page, next: string | null): unknown[] {
341 1 : if (typeof item !== 'object' || item === null || Array.isArray(item)) {
342 0 : return this.standardRow(page.status, next, null, page.meta, item);
343 : }
344 1 : const rest: Record<string, unknown> = { ...item };
345 1 : const row: unknown[] = this.columns.map(column => {
346 1 : const apiName = column.api_name ?? column.name;
347 1 : const converted = convertValue(rest[apiName], column.type);
348 : // A value that can't be converted stays in `_raw`, so a spec that disagrees doesn't fail the statement
349 1 : if (converted !== undefined) {
350 1 : delete rest[apiName];
351 : }
352 1 : return converted ?? null;
353 1 : });
354 1 : row.push(page.status, next);
355 1 : if (this.pending !== null) {
356 1 : row.push(null);
357 : }
358 1 : row.push(page.meta === null ? null : JSON.stringify(page.meta));
359 1 : row.push(Object.keys(rest).length > 0 ? JSON.stringify(rest) : null);
360 1 : return row;
361 1 : }
362 :
363 : /// Send a request
364 1 : private async send(request: ClientFunctionRequest): Promise<HttpFetchResult> {
365 1 : const { baseUrl, headers } = await this.env.resolveHost(request.host);
366 1 : const url = requestUrl(request, baseUrl);
367 1 : const merged = new Headers(headers);
368 1 : for (const [name, value] of request.headers) {
369 1 : merged.set(name, value);
370 : }
371 1 : const init: RequestInit = { method: request.method, headers: merged, signal: this.abort };
372 1 : if (request.body !== null) {
373 1 : init.body = request.body;
374 : }
375 1 : return this.env.http.fetch(url, init);
376 1 : }
377 :
378 : /// Send a request and poll while it's pending, up to the time limit.
379 : /// Returns the response, or the value to poll if the call is still running when the time runs out.
380 1 : private async sendAndWait(request: ClientFunctionRequest, waitSeconds: number): Promise<{ response: HttpFetchResult } | { poll: string; status: number }> {
381 1 : const now = this.env.now ?? Date.now;
382 1 : const sleep = this.env.sleep ?? ((ms: number, abort?: AbortSignal) => abortableSleep(ms, abort));
383 1 : const deadline = now() + waitSeconds * 1000;
384 1 : let current = request;
385 1 : let delay = POLL_MIN_MS;
386 1 : for (;;) {
387 1 : const response = await this.send(current);
388 1 : if (this.pending === null || response.status !== this.pending.status) {
389 1 : return { response };
390 : }
391 1 : const rule = this.pending.poll;
392 1 : if (!('header' in rule) && !('repeat' in rule)) {
393 0 : throw new ClientFunctionError(`${this.spec.namespace}.${this.fn.name} has the unknown poll rule ${JSON.stringify(rule)}`);
394 : }
395 1 : let poll = POLL_SAME;
396 1 : if ('header' in rule) {
397 1 : const value = response.headers.get(rule.header);
398 1 : if (isEmptyValue(value)) {
399 1 : throw new ClientFunctionError(`${this.spec.namespace}.${this.fn.name} answered ${response.status} without a ${rule.header} header`, response.status);
400 : }
401 1 : poll = value!;
402 1 : }
403 1 : const remaining = deadline - now();
404 1 : if (remaining <= 0) {
405 1 : return { poll, status: response.status };
406 : }
407 1 : const retryAfter = Number(response.headers.get('Retry-After'));
408 1 : const wait = Math.min(Number.isFinite(retryAfter) && retryAfter > 0 ? retryAfter * 1000 : delay, remaining);
409 1 : await sleep(wait, this.abort);
410 1 : delay = Math.min(delay * 2, POLL_MAX_MS);
411 1 : if (poll !== POLL_SAME) {
412 1 : current = followRequest(request, poll);
413 : }
414 1 : }
415 1 : }
416 :
417 : /// Read a response as a page, or fail
418 1 : private async finish(response: HttpFetchResult): Promise<Page> {
419 1 : const body = parseBody(await response.text());
420 1 : if (this.fn.errors?.empty_on?.includes(response.status)) {
421 1 : return { status: response.status, items: [], meta: null, body: null, headers: response.headers };
422 : }
423 1 : const fallback = `${this.spec.namespace}.${this.fn.name} failed with the status ${response.status}`;
424 1 : if (response.status < 200 || response.status >= 300) {
425 1 : throw new ClientFunctionError(errorMessage(this.spec, this.fn, body, fallback), response.status);
426 : }
427 1 : const errorIf = this.fn.errors?.error_if;
428 1 : if (errorIf !== undefined && !isEmptyValue(readJsonPointer(body, errorIf))) {
429 1 : throw new ClientFunctionError(errorMessage(this.spec, this.fn, body, fallback), response.status);
430 : }
431 1 : return readPage(this.fn, response.status, body, response.headers);
432 1 : }
433 :
434 : /// Run the call
435 1 : async run(options: Record<string, unknown>): Promise<ClientFunctionResult> {
436 1 : const { fn, spec, env } = this;
437 1 : const paging = fn.paging;
438 1 : const next = options['_next'] ?? null;
439 1 : const poll = options['_poll'] ?? null;
440 1 : const pages = Number(options['_pages'] ?? 1);
441 1 : const wait = Number(options['_wait'] ?? 0);
442 1 : if (!Number.isInteger(pages) || pages < 1) {
443 0 : throw new ClientFunctionError('_pages takes a positive number');
444 : }
445 1 : if (!Number.isFinite(wait) || wait < 0) {
446 0 : throw new ClientFunctionError('_wait takes a number of seconds');
447 : }
448 1 : if (next !== null && poll !== null) {
449 0 : throw new ClientFunctionError('_next and _poll continue different things, use one of them');
450 : }
451 1 : if (poll !== null && this.pending === null) {
452 1 : throw new ClientFunctionError(`${spec.namespace}.${fn.name} has no option _poll`);
453 : }
454 1 : if ((next !== null || pages > 1) && paging === undefined) {
455 1 : throw new ClientFunctionError(`${spec.namespace}.${fn.name} doesn't page`);
456 : }
457 :
458 1 : const send: Record<string, unknown> = { ...optionsWithNext(fn, options) };
459 1 : const numeric = paging?.kind === 'offset' || paging?.kind === 'page';
460 1 : const variables = env.variables?.() ?? {};
461 1 : const build = (values: Record<string, unknown>) => buildClientFunctionRequest(spec, fn, values, variables);
462 1 : let request = firstRequest(spec, fn, options, variables);
463 :
464 1 : const start = paging?.start ?? 0;
465 1 : const size = paging?.size !== undefined && typeof options[paging.size] === 'number' ? options[paging.size] as number : null;
466 1 : let position = numeric ? (paging!.option !== undefined && typeof send[paging!.option!] === 'number' ? send[paging!.option!] as number : start) : 0;
467 :
468 1 : const pageRows: { page: Page }[] = [];
469 1 : let lastNext: string | null = null;
470 1 : for (let i = 0; i < pages; ++i) {
471 1 : const outcome = await this.sendAndWait(request, wait);
472 1 : if (!('response' in outcome)) {
473 1 : if (i > 0) {
474 0 : throw new ClientFunctionError(`${spec.namespace}.${fn.name}: a later page is still running, use _wait`, outcome.status);
475 : }
476 1 : return { schema: this.schema(), rows: [this.standardRow(outcome.status, null, outcome.poll, null, null)] };
477 1 : }
478 1 : const page = await this.finish(outcome.response);
479 1 : pageRows.push({ page });
480 1 : const value = page.unexpected === undefined ? nextValue(paging, page, position, size) : null;
481 : // A list that doesn't move on would repeat forever
482 1 : if (value !== null && value === (i === 0 ? next : lastNext)) {
483 1 : throw new ClientFunctionError(`${spec.namespace}.${fn.name} returned the same next value twice`);
484 : }
485 1 : lastNext = value;
486 1 : if (value === null || i + 1 === pages) {
487 1 : break;
488 : }
489 : // The next page of this call
490 1 : position = paging!.kind === 'offset' || paging!.kind === 'page' ? Number(value) : position;
491 1 : if (paging!.kind === 'url') {
492 1 : request = followRequest(request, value);
493 1 : } else {
494 1 : send[paging!.option!] = numeric ? Number(value) : value;
495 1 : request = build(send);
496 1 : }
497 1 : }
498 :
499 : // `_next` on every row holds the value after the last page. No row can't carry it, so an empty page with a next
500 : // value still has one.
501 1 : const rows: unknown[][] = [];
502 1 : for (const { page } of pageRows) {
503 1 : for (const item of page.items) {
504 1 : rows.push(this.itemRow(item, page, lastNext));
505 : }
506 1 : if (page.unexpected !== undefined) {
507 1 : rows.push(this.standardRow(page.status, lastNext, null, null, page.unexpected));
508 : }
509 1 : }
510 1 : if (rows.length === 0 && (lastNext !== null || paging === undefined)) {
511 1 : const last = pageRows[pageRows.length - 1].page;
512 1 : rows.push(this.standardRow(last.status, lastNext, null, last.meta, null));
513 1 : }
514 1 : return { schema: this.schema(), rows };
515 1 : }
516 1 : }
517 :
518 : /// Sleep until the time is over or the signal aborts
519 1 : function abortableSleep(ms: number, abort?: AbortSignal): Promise<void> {
520 1 : return new Promise((resolve, reject) => {
521 1 : abort?.throwIfAborted();
522 1 : const timer = setTimeout(resolve, ms);
523 1 : abort?.addEventListener('abort', () => {
524 0 : clearTimeout(timer);
525 0 : reject(abort.reason);
526 1 : }, { once: true });
527 1 : });
528 1 : }
529 :
530 : /// Run a call. A call that writes is sent like any other: the planner leaves a script with one out of the runs of the
531 : /// session and of input changes, see docs/design/client_functions.md.
532 1 : export async function runClientFunctionCall(spec: ClientFunctionSpec, description: ClientFunctionCallDescription, env: ClientFunctionEnvironment, abort?: AbortSignal): Promise<ClientFunctionResult> {
533 1 : requireValues(description);
534 1 : const run = new ClientFunctionCallRun(spec, description, env, abort);
535 1 : return run.run(description.options);
536 1 : }
537 :
538 : /// A plan puts the values of its inputs in a description and the shell has none, so a description here has them. Its options
539 : /// would be sent with a null otherwise.
540 1 : function requireValues(description: ClientFunctionCallDescription): void {
541 1 : const binding = description.bindings?.[0];
542 1 : if (binding !== undefined) {
543 1 : throw new ClientFunctionError(`the parameter :${binding.param} has no value`);
544 : }
545 1 : }
546 :
547 : /// The result of `EXPLAIN CALL`: the resolved request. It sends nothing, needs no credentials, and shows the name of the
548 : /// host.
549 1 : export async function explainClientFunctionCall(spec: ClientFunctionSpec, description: ClientFunctionCallDescription, env: Pick<ClientFunctionEnvironment, 'variables'> = {}): Promise<ClientFunctionResult> {
550 1 : const fn = findClientFunction(spec, description.function);
551 1 : requireValues(description);
552 1 : const options = description.options;
553 : // A base path shows its variables if the binding isn't there to fill them
554 1 : const variables = new Proxy(env.variables?.() ?? {}, {
555 1 : get: (target, name: string) => target[name] ?? `{${name}}`,
556 1 : });
557 1 : const request = firstRequest(spec, fn, options, variables);
558 1 : const standard = Object.fromEntries(Object.entries(options).filter(([name]) => (STANDARD_OPTIONS as readonly string[]).includes(name)));
559 1 : const text = (value: unknown) => value === null ? null : String(value);
560 1 : const query = request.query.map(([name, value]) => `${name}=${value}`).join('&');
561 1 : const fields: [string, string | null][] = [
562 1 : ['method', request.method],
563 1 : ['host', request.host],
564 1 : ['path', request.followUrl ?? request.path],
565 1 : ['query', query === '' ? null : query],
566 1 : ['headers', request.headers.length > 0 ? JSON.stringify(Object.fromEntries(request.headers)) : null],
567 1 : ['body', text(request.body)],
568 1 : ['standard_options', Object.keys(standard).length > 0 ? JSON.stringify(standard) : null],
569 1 : ];
570 1 : return {
571 1 : schema: new arrow.Schema(fields.map(([name]) => new arrow.Field(name, new arrow.Utf8(), true))),
572 1 : rows: [fields.map(([, value]) => value)],
573 1 : };
574 1 : }
|
