Line data Source code
1 : import { HttpClient, HttpFetchResult } from '../../platform/http/http_client.js';
2 : import { Logger, LoggableException } from '../../platform/logger/logger.js';
3 :
4 1 : const LOG_CTX = "hyperdb_http_client";
5 :
6 : // ---------------------------------------------------------------------------
7 : // Shared V3 API types
8 : // ---------------------------------------------------------------------------
9 :
10 : export type CompletionStatus =
11 : | "RUNNING_OR_UNSPECIFIED"
12 : | "FINISHED"
13 : | "RESULTS_PRODUCED";
14 :
15 : export interface ColumnDefinition {
16 : name?: string;
17 : type?: string;
18 : nullable?: boolean;
19 : }
20 :
21 : export interface ExecutionStats {
22 : wallClockTime: number;
23 : rowsProcessed: number;
24 : }
25 :
26 : export interface QueryStatus {
27 : queryId: string;
28 : completionStatus: CompletionStatus;
29 : chunkCount: number | null;
30 : rowCount: number | null;
31 : progress: number | null;
32 : expirationTime: string | null;
33 : executionStats: ExecutionStats | null;
34 : }
35 :
36 : /// The order in which a query advances through the completion states
37 1 : const COMPLETION_STATUS_RANK: Record<CompletionStatus, number> = {
38 1 : RUNNING_OR_UNSPECIFIED: 0,
39 1 : RESULTS_PRODUCED: 1,
40 1 : FINISHED: 2,
41 1 : };
42 :
43 : /// Read a number that the server may serialize as a JSON string
44 1 : function readNumber(value: unknown): number | null {
45 1 : const n = typeof value === "string" ? Number(value) : value;
46 1 : return typeof n === "number" && Number.isFinite(n) ? n : null;
47 1 : }
48 :
49 : /// Read a query status from a status header or a status response.
50 : /// The status response serializes chunkCount and rowCount as strings.
51 1 : export function readQueryStatus(raw: any): QueryStatus {
52 1 : const stats = raw.executionStats;
53 1 : return {
54 1 : queryId: raw.queryId,
55 1 : completionStatus: raw.completionStatus ?? "RUNNING_OR_UNSPECIFIED",
56 1 : chunkCount: readNumber(raw.chunkCount),
57 1 : rowCount: readNumber(raw.rowCount),
58 1 : progress: readNumber(raw.progress),
59 1 : expirationTime: raw.expirationTime ?? null,
60 1 : executionStats: stats ? {
61 1 : wallClockTime: readNumber(stats.wallClockTime) ?? 0,
62 1 : rowsProcessed: readNumber(stats.rowsProcessed) ?? 0,
63 1 : } : null,
64 1 : };
65 1 : }
66 :
67 1 : function maxOrNull(a: number | null, b: number | null): number | null {
68 1 : return a == null ? b : b == null ? a : Math.max(a, b);
69 1 : }
70 :
71 : /// Merge a status into the current one.
72 : /// Responses of parallel requests arrive in any order, an older status must not roll back a newer one.
73 1 : export function mergeQueryStatus(current: QueryStatus, update: QueryStatus): QueryStatus {
74 1 : const currentRank = COMPLETION_STATUS_RANK[current.completionStatus] ?? 0;
75 1 : const updateRank = COMPLETION_STATUS_RANK[update.completionStatus] ?? 0;
76 1 : const a = current.executionStats;
77 1 : const b = update.executionStats;
78 1 : return {
79 1 : queryId: current.queryId,
80 1 : completionStatus: updateRank >= currentRank ? update.completionStatus : current.completionStatus,
81 1 : chunkCount: maxOrNull(current.chunkCount, update.chunkCount),
82 1 : rowCount: maxOrNull(current.rowCount, update.rowCount),
83 1 : progress: maxOrNull(current.progress, update.progress),
84 1 : expirationTime: update.expirationTime ?? current.expirationTime,
85 1 : executionStats: a && b ? {
86 1 : wallClockTime: Math.max(a.wallClockTime, b.wallClockTime),
87 1 : rowsProcessed: Math.max(a.rowsProcessed, b.rowsProcessed),
88 1 : } : (b ?? a),
89 1 : };
90 1 : }
91 :
92 : // ---------------------------------------------------------------------------
93 : // Request types
94 : // ---------------------------------------------------------------------------
95 :
96 : export interface ExecuteQueryRequest {
97 : sql: string;
98 : transferMode?: "ADAPTIVE" | string;
99 : settings?: Record<string, unknown>;
100 : resultRange?: {
101 : rowLimit?: number;
102 : byteLimit: number;
103 : };
104 : queryRowLimit?: number;
105 : paramStyle?: "NAMED" | string;
106 : databases?: {
107 : path: string;
108 : alias?: string;
109 : }[];
110 : }
111 :
112 : export interface GetQueryStatusParams {
113 : queryId: string;
114 : waitTimeMs?: number;
115 : }
116 :
117 : export interface GetQueryChunkParams {
118 : queryId: string;
119 : chunkId: number;
120 : omitSchema?: boolean;
121 : }
122 :
123 : export interface GetQueryRowsParams {
124 : queryId: string;
125 : offset: number;
126 : limit?: number;
127 : byteLimit?: number;
128 : omitSchema?: boolean;
129 : }
130 :
131 : // ---------------------------------------------------------------------------
132 : // Optional request headers
133 : // ---------------------------------------------------------------------------
134 :
135 : export interface HyperHttpRequestHeaders {
136 : workload?: string;
137 : externalClientContext?: string;
138 : adaptiveTimeout?: string;
139 : }
140 :
141 : // ---------------------------------------------------------------------------
142 : // Auth provider
143 : // ---------------------------------------------------------------------------
144 :
145 : export interface HyperHttpAuthProvider {
146 : getAuthHeaders(): Promise<Record<string, string>>;
147 : }
148 :
149 : // ---------------------------------------------------------------------------
150 : // Error
151 : // ---------------------------------------------------------------------------
152 :
153 : export interface QueryErrorResponse {
154 : error: string;
155 : message: string;
156 : details?: {
157 : customerHint?: string;
158 : customerDetail?: string;
159 : systemDetail?: string;
160 : errorSource?: string;
161 : position?: {
162 : errorBeginCharacterOffset: string;
163 : errorEndCharacterOffset: string;
164 : };
165 : };
166 : }
167 :
168 1 : export class HyperHttpError extends LoggableException {
169 1 : public readonly httpStatus: number;
170 1 : public readonly errorResponse: QueryErrorResponse | null;
171 :
172 1 : constructor(httpStatus: number, errorResponse: QueryErrorResponse | null, diagnostics: Record<string, string | null> = {}) {
173 1 : super(errorResponse?.message || errorResponse?.error || `HTTP ${httpStatus}`, diagnostics, LOG_CTX);
174 1 : this.httpStatus = httpStatus;
175 1 : this.errorResponse = errorResponse;
176 1 : }
177 1 : }
178 :
179 : /// The maximum length of a raw header value in the error diagnostics
180 1 : const MAX_DIAGNOSTIC_LENGTH = 2000;
181 : /// How deep to search nested objects for an error
182 1 : const MAX_ERROR_SEARCH_DEPTH = 3;
183 :
184 1 : function truncate(text: string, maxLength: number): string {
185 1 : return text.length <= maxLength ? text : `${text.substring(0, maxLength)}...`;
186 1 : }
187 :
188 1 : function nonEmptyString(value: unknown): string | undefined {
189 1 : return typeof value === 'string' && value.trim() ? value : undefined;
190 1 : }
191 :
192 : /// Read a Hyper ErrorInfo serialized as JSON, in camelCase or snake_case
193 1 : function readErrorInfo(raw: any, httpStatus: number): QueryErrorResponse | null {
194 1 : const primaryMessage = nonEmptyString(raw.primaryMessage ?? raw.primary_message);
195 1 : if (!primaryMessage) return null;
196 1 : const position = raw.position;
197 1 : const begin = position?.errorBeginCharacterOffset ?? position?.error_begin_character_offset;
198 1 : const end = position?.errorEndCharacterOffset ?? position?.error_end_character_offset;
199 1 : return {
200 1 : error: nonEmptyString(raw.sqlstate) ?? `HTTP ${httpStatus}`,
201 1 : message: primaryMessage,
202 1 : details: {
203 1 : customerHint: nonEmptyString(raw.customerHint ?? raw.customer_hint),
204 1 : customerDetail: nonEmptyString(raw.customerDetail ?? raw.customer_detail),
205 1 : systemDetail: nonEmptyString(raw.systemDetail ?? raw.system_detail),
206 1 : errorSource: nonEmptyString(raw.errorSource ?? raw.error_source),
207 1 : position: begin != null && end != null ? {
208 1 : errorBeginCharacterOffset: String(begin),
209 1 : errorEndCharacterOffset: String(end),
210 1 : } : undefined,
211 1 : },
212 1 : };
213 1 : }
214 :
215 : /// Read an error from a JSON error body or the JSON status header.
216 : /// Recognizes the V3 error envelope, Salesforce error arrays, Hyper ErrorInfo objects,
217 : /// google.rpc.Status objects with packed details, and any of these nested in other objects.
218 1 : export function readErrorResponse(raw: unknown, httpStatus: number, depth: number = 0): QueryErrorResponse | null {
219 1 : if (raw == null || typeof raw !== 'object' || depth > MAX_ERROR_SEARCH_DEPTH) return null;
220 1 : if (Array.isArray(raw)) {
221 1 : for (const entry of raw) {
222 1 : const error = readErrorResponse(entry, httpStatus, depth + 1);
223 1 : if (error) return error;
224 1 : }
225 0 : return null;
226 0 : }
227 1 : const obj = raw as Record<string, any>;
228 :
229 : // Hyper ErrorInfo
230 1 : const errorInfo = readErrorInfo(obj, httpStatus);
231 1 : if (errorInfo) return errorInfo;
232 :
233 : // google.rpc.Status, prefer a packed ErrorInfo over the terse message
234 1 : if (Array.isArray(obj.details)) {
235 1 : for (const detail of obj.details) {
236 1 : if (detail && typeof detail === 'object') {
237 1 : const info = readErrorInfo(detail, httpStatus);
238 1 : if (info) return info;
239 1 : }
240 : }
241 : }
242 :
243 : // V3 error envelope or Salesforce error
244 1 : const message = nonEmptyString(obj.message);
245 1 : const error = nonEmptyString(obj.error);
246 1 : const errorCode = nonEmptyString(obj.errorCode);
247 1 : if (message || error || errorCode) {
248 1 : return {
249 1 : ...obj,
250 1 : message: message ?? '',
251 1 : error: error ?? errorCode ?? `HTTP ${httpStatus}`,
252 1 : details: obj.details != null && typeof obj.details === 'object' && !Array.isArray(obj.details) ? obj.details : undefined,
253 1 : } as QueryErrorResponse;
254 : }
255 :
256 : // An error nested in another object, e.g. the error of a query status
257 1 : for (const value of Object.values(obj)) {
258 1 : const nested = readErrorResponse(value, httpStatus, depth + 1);
259 1 : if (nested) return nested;
260 1 : }
261 1 : return null;
262 1 : }
263 :
264 : // ---------------------------------------------------------------------------
265 : // Client
266 : // ---------------------------------------------------------------------------
267 :
268 1 : export class HyperDatabaseHttpClient {
269 1 : httpClient: HttpClient;
270 : /// The Hyper endpoint URL (the Salesforce Data Cloud instance URL).
271 1 : endpointUrl: URL;
272 1 : auth: HyperHttpAuthProvider;
273 1 : logger: Logger;
274 :
275 1 : constructor(httpClient: HttpClient, endpointUrl: URL, auth: HyperHttpAuthProvider, logger: Logger) {
276 1 : this.httpClient = httpClient;
277 1 : this.endpointUrl = endpointUrl;
278 1 : this.auth = auth;
279 1 : this.logger = logger;
280 1 : }
281 :
282 1 : private async buildHeaders(extra?: HyperHttpRequestHeaders): Promise<Headers> {
283 1 : const authHeaders = await this.auth.getAuthHeaders();
284 1 : const headers = new Headers({
285 1 : "Accept": "application/vnd.apache.arrow.stream",
286 1 : "Content-Type": "application/json",
287 1 : ...authHeaders,
288 1 : });
289 1 : if (extra?.workload) {
290 0 : headers.set("x-hyperdb-workload", extra.workload);
291 : }
292 1 : if (extra?.externalClientContext) {
293 1 : headers.set("x-hyperdb-external-client-context", extra.externalClientContext);
294 : }
295 1 : if (extra?.adaptiveTimeout) {
296 0 : headers.set("X-hyperdb-adaptive-timeout", extra.adaptiveTimeout);
297 : }
298 1 : return headers;
299 1 : }
300 :
301 1 : private parseStatusHeader(response: HttpFetchResult): QueryStatus | null {
302 1 : const raw = response.headers.get("x-hyperdb-status");
303 1 : if (!raw) return null;
304 1 : return readQueryStatus(JSON.parse(raw));
305 1 : }
306 :
307 1 : private async throwIfError(response: HttpFetchResult): Promise<void> {
308 1 : if (response.status >= 400) {
309 1 : let errorBody: QueryErrorResponse | null = null;
310 1 : let errorOrigin = 'none';
311 1 : let bodyType = 'unreadable';
312 1 : let bodyKeys = '';
313 1 : try {
314 1 : const body = await response.json();
315 1 : bodyType = body == null ? 'null' : Array.isArray(body) ? 'array' : typeof body;
316 1 : bodyKeys = body != null && typeof body === 'object'
317 1 : ? Object.keys(Array.isArray(body) ? (body[0] ?? {}) : body)
318 1 : .filter(key => ['message', 'error', 'errorCode', 'details', 'detail'].includes(key)).join(',')
319 0 : : '';
320 1 : errorBody = readErrorResponse(body, response.status);
321 1 : if (errorBody) errorOrigin = 'body';
322 1 : } catch {
323 : // Response body may not be valid JSON
324 0 : }
325 : // The body may be empty and the error is only in the status header
326 1 : const rawStatus = response.headers.get('x-hyperdb-status');
327 1 : if (errorBody == null && rawStatus) {
328 1 : try {
329 1 : errorBody = readErrorResponse(JSON.parse(rawStatus), response.status);
330 1 : if (errorBody) errorOrigin = 'x-hyperdb-status';
331 1 : } catch {
332 : // Status header may not be valid JSON
333 0 : }
334 : }
335 1 : const diagnostics = {
336 1 : status: response.status.toString(),
337 1 : statusText: response.statusText,
338 1 : contentType: response.headers.get('content-type'),
339 1 : hyperdbStatusPresent: (rawStatus != null).toString(),
340 1 : hyperdbStatus: rawStatus != null ? truncate(rawStatus, MAX_DIAGNOSTIC_LENGTH) : null,
341 1 : bodyType,
342 1 : bodyKeys,
343 1 : hyperErrorRecognized: (errorBody != null).toString(),
344 1 : hyperErrorOrigin: errorOrigin,
345 1 : };
346 1 : this.logger.warn('Hyper HTTP error response', diagnostics, LOG_CTX);
347 1 : throw new HyperHttpError(response.status, errorBody, diagnostics);
348 1 : }
349 1 : }
350 :
351 : /// POST /api/v3/query
352 1 : async executeQuery(request: ExecuteQueryRequest, extra?: HyperHttpRequestHeaders, abort?: AbortSignal): Promise<{ status: QueryStatus | null; response: HttpFetchResult }> {
353 1 : const url = new URL("/api/v3/query", this.endpointUrl);
354 1 : const headers = await this.buildHeaders(extra);
355 1 : const response = await this.httpClient.fetch(url, {
356 1 : method: "POST",
357 1 : headers,
358 1 : body: JSON.stringify(request),
359 1 : signal: abort,
360 1 : });
361 1 : await this.throwIfError(response);
362 1 : const status = this.parseStatusHeader(response);
363 1 : return { status, response };
364 1 : }
365 :
366 : /// GET /api/v3/query/{queryId}
367 1 : async getQueryStatus(params: GetQueryStatusParams, abort?: AbortSignal): Promise<QueryStatus> {
368 1 : const url = new URL(`/api/v3/query/${encodeURIComponent(params.queryId)}`, this.endpointUrl);
369 1 : if (params.waitTimeMs !== undefined) {
370 0 : url.searchParams.set("waitTimeMs", params.waitTimeMs.toString());
371 : }
372 1 : const headers = await this.buildHeaders();
373 1 : const response = await this.httpClient.fetch(url, {
374 1 : method: "GET",
375 1 : headers,
376 1 : signal: abort,
377 1 : });
378 1 : await this.throwIfError(response);
379 1 : return readQueryStatus(await response.json());
380 1 : }
381 :
382 : /// DELETE /api/v3/query/{queryId}
383 1 : async cancelQuery(queryId: string, abort?: AbortSignal): Promise<void> {
384 1 : const url = new URL(`/api/v3/query/${encodeURIComponent(queryId)}`, this.endpointUrl);
385 1 : const headers = await this.buildHeaders();
386 1 : const response = await this.httpClient.fetch(url, {
387 1 : method: "DELETE",
388 1 : headers,
389 1 : signal: abort,
390 1 : });
391 1 : await this.throwIfError(response);
392 1 : }
393 :
394 : /// GET /api/v3/query/{queryId}/chunk/{chunkId}
395 1 : async getQueryChunk(params: GetQueryChunkParams, extra?: HyperHttpRequestHeaders, abort?: AbortSignal): Promise<{ status: QueryStatus | null; response: HttpFetchResult }> {
396 1 : const url = new URL(`/api/v3/query/${encodeURIComponent(params.queryId)}/chunk/${params.chunkId}`, this.endpointUrl);
397 1 : if (params.omitSchema) {
398 1 : url.searchParams.set("omitSchema", "true");
399 : }
400 1 : const headers = await this.buildHeaders(extra);
401 1 : const response = await this.httpClient.fetch(url, {
402 1 : method: "GET",
403 1 : headers,
404 1 : signal: abort,
405 1 : });
406 1 : await this.throwIfError(response);
407 1 : const status = this.parseStatusHeader(response);
408 1 : return { status, response };
409 1 : }
410 :
411 : /// GET /api/v3/query/{queryId}/row
412 1 : async getQueryRows(params: GetQueryRowsParams, extra?: HyperHttpRequestHeaders, abort?: AbortSignal): Promise<HttpFetchResult> {
413 1 : const url = new URL(`/api/v3/query/${encodeURIComponent(params.queryId)}/row`, this.endpointUrl);
414 1 : url.searchParams.set("offset", params.offset.toString());
415 1 : if (params.limit !== undefined) {
416 0 : url.searchParams.set("limit", params.limit.toString());
417 : }
418 1 : if (params.byteLimit !== undefined) {
419 0 : url.searchParams.set("byteLimit", params.byteLimit.toString());
420 : }
421 1 : if (params.omitSchema) {
422 0 : url.searchParams.set("omitSchema", "true");
423 : }
424 1 : const headers = await this.buildHeaders(extra);
425 1 : const response = await this.httpClient.fetch(url, {
426 1 : method: "GET",
427 1 : headers,
428 1 : signal: abort,
429 1 : });
430 1 : await this.throwIfError(response);
431 1 : return response;
432 1 : }
433 1 : }
|
