Line data Source code
1 : import * as core from '@ankoh/dashql-core';
2 : import * as connection from '@ankoh/dashql-jsonschema/connection.js';
3 :
4 : import type { HyperConnectorAction } from './hyper/hyper_connection_state.js';
5 : import type { SalesforceConnectionStateAction } from './salesforce/salesforce_connection_state.js';
6 : import { CatalogUpdateTaskState, reduceCatalogAction } from './catalog/catalog_update_state.js';
7 : import { generateCatalogScriptHeader, CatalogSource } from './catalog/catalog_sql_generator.js';
8 : import { generateFunctionScriptHeader } from './catalog/catalog_function_sql_generator.js';
9 : import { VariantKind } from '../utils/variant.js';
10 : import {
11 : CONNECTOR_INFOS,
12 : ConnectorInfo,
13 : getConnectorInfoForParams,
14 : ConnectorType,
15 : } from './connector_info.js';
16 : import {
17 : createQueryExecutionMetrics,
18 : EXECUTE_QUERY,
19 : QUERY_CANCELLED,
20 : QUERY_FAILED,
21 : QUERY_PROGRESS_UPDATED,
22 : QUERY_STATEMENT_STARTED,
23 : QUERY_STATEMENT_SUCCEEDED,
24 : QUERY_RECEIVED_BATCH,
25 : QUERY_RUNNING,
26 : QUERY_RECEIVED_ALL_BATCHES,
27 : QueryExecutionAction,
28 : QUERY_SUCCEEDED,
29 : QUERY_PROCESSED_RESULTS,
30 : QUERY_PREPARING,
31 : QUERY_SENDING,
32 : QUERY_PROCESSING_RESULTS,
33 : QUERY_CACHE_RECORDED,
34 : QUERY_CACHE_DELETED,
35 : QueryExecutionState,
36 : } from '../query/query_execution_state.js';
37 : import { reduceQueryAction } from '../query/query_execution_state.js';
38 : import type { TrinoConnectorAction } from './trino/trino_connection_state.js';
39 : import { ConnectionStateDetailsVariant, createConnectionStateDetails, isSalesforceDetails } from './connection_state_details.js';
40 : import { Logger, stringifyError } from '../platform/logger/logger.js';
41 : import { invalidateCatalogBatch } from './catalog/catalog_batch.js';
42 : import { connectionParamsEqual, connectionParamsWithoutDraftRows, getConnectionParamsFromStateDetails, type ConnectionParams } from './connection_params.js';
43 : import { HYPER_CHANNEL_SETUP_STARTED } from './hyper/hyper_connection_state.js';
44 : import { OAUTH_STARTED, TRINO_CHANNEL_SETUP_STARTED } from './trino/trino_connection_state.js';
45 : import { SETUP_STARTED } from './salesforce/salesforce_connection_state.js';
46 : import { cancelConnectionSetup } from './connection_setup_cancellation.js';
47 : import { dateToTimestamp } from './proto_helper.js';
48 : import type { DetailedError } from '../utils/error.js';
49 : import { catalogNameDefaults, connectionParamsOf, connectorOfDetails, connectorParams } from './connector_descriptor.js';
50 :
51 1 : const LOG_CTX = 'connection_state';
52 :
53 : export interface CatalogUpdates {
54 : /// The running tasks
55 : tasksRunning: Map<number, CatalogUpdateTaskState>;
56 : /// The finished tasks
57 : tasksFinished: Map<number, CatalogUpdateTaskState>;
58 : /// Restored at a certain time
59 : restoredAt: Date | null;
60 : /// The id of the running or most recently completed full refresh.
61 : /// Set at UPDATE_CATALOG start; used to decide whether to kick off
62 : /// an auto-refresh and to surface the currently-displayed refresh task.
63 : currentFullRefresh: number | null;
64 : /// The id of the most recently *completed*, non-superseded full refresh
65 : /// (succeeded, failed, or cancelled). Only advances on completion, so components
66 : /// can depend on it to react when the catalog script has been updated.
67 : lastFullRefresh: number | null;
68 : }
69 :
70 : export interface ConnectionState {
71 : /// Runtime identity of the connection.
72 : connectionId: string;
73 :
74 : /// The connection state contains many references into the Wasm heap.
75 : /// It therefore makes sense that connection state users resolve the "right" module through here.
76 : instance: core.DashQL;
77 : /// Whether the database belongs to a committed session or has connected at least once.
78 : /// Draft manifest writes are independent of activation; catalog writes require activation.
79 : active: boolean;
80 :
81 : /// The connection state
82 : connectionStatus: ConnectionStatus;
83 : /// The connection health
84 : connectionHealth: ConnectionHealth;
85 : /// The connection info
86 : connectorInfo: ConnectorInfo;
87 : /// The name the user gave the connection, the label falls back to the connector without it
88 : name?: string;
89 : /// Does the connection cache query results? Absent for the default of the connector.
90 : cacheQueryResults?: boolean;
91 :
92 : /// The connection details
93 : details: ConnectionStateDetailsVariant;
94 : /// Last successful parameters (initial defaults until the first successful setup).
95 : params: ConnectionParams;
96 : /// Local editable settings, independent of the runtime channel and OAuth state.
97 : paramsDraft?: ConnectionParams;
98 : /// Immutable input of the current setup attempt; never persisted.
99 : setupAttemptParams?: ConnectionParams;
100 : /// Draft at attempt start, including raw form rows, for exact successful-discard comparison.
101 : setupAttemptDraft?: ConnectionParams;
102 : /// The connection statistics
103 : metrics: connection.ConnectionMetrics;
104 :
105 : /// The catalog
106 : catalog: core.DashQLCatalog;
107 : /// The catalog updates
108 : catalogUpdates: CatalogUpdates;
109 : /// The catalog relation script (consolidated SQL for all relations)
110 : catalogRelationScript: core.DashQLScript;
111 : /// The catalog function script (consolidated SQL for all function declarations)
112 : catalogFunctionScript: core.DashQLScript;
113 : /// The catalog changes of executed statements since the last catalog refresh, as SQL.
114 : /// They are applied to the relation descriptor on top of the relation script.
115 : catalogChangesSql: string;
116 :
117 : /// The queries that are currently running
118 : queriesActive: Map<number, QueryExecutionState>;
119 : /// The active queries ordered
120 : queriesActiveOrdered: number[];
121 : /// The queries that finished (succeeded, failed, cancelled)
122 : queriesFinished: Map<number, QueryExecutionState>;
123 : /// The finished queries ordered
124 : queriesFinishedOrdered: number[];
125 :
126 : /// The snapshot of query ids that are active or finished
127 : snapshotQueriesActiveFinished: number;
128 : }
129 :
130 1 : export enum ConnectionStatus {
131 1 : NOT_STARTED,
132 :
133 : // Generate setup status
134 1 : HEALTH_CHECK_STARTED,
135 1 : HEALTH_CHECK_FAILED,
136 1 : HEALTH_CHECK_CANCELLED,
137 1 : HEALTH_CHECK_SUCCEEDED,
138 :
139 : // Channel setup
140 1 : CHANNEL_SETUP_STARTED,
141 1 : CHANNEL_SETUP_FAILED,
142 1 : CHANNEL_SETUP_CANCELLED,
143 1 : CHANNEL_READY,
144 :
145 : // Generate OAuth
146 1 : AUTH_STARTED,
147 1 : AUTH_CANCELLED,
148 1 : AUTH_FAILED,
149 1 : PKCE_GENERATION_STARTED,
150 1 : PKCE_GENERATED,
151 1 : WAITING_FOR_OAUTH_CODE_VIA_WINDOW,
152 1 : WAITING_FOR_OAUTH_CODE_VIA_LINK,
153 1 : OAUTH_CODE_RECEIVED,
154 1 : ACCESS_TOKEN_REQUESTED,
155 1 : ACCESS_TOKEN_RECEIVED,
156 :
157 : // Salesforce OAuth
158 1 : DATA_CLOUD_TOKEN_REQUESTED,
159 1 : DATA_CLOUD_TOKEN_RECEIVED,
160 1 : CORE_ACCESS_TOKEN_REQUESTED,
161 1 : CORE_ACCESS_TOKEN_RECEIVED,
162 1 : }
163 :
164 1 : export enum ConnectionHealth {
165 1 : NOT_STARTED = 0,
166 1 : CONNECTING = 1,
167 1 : CANCELLED = 2,
168 1 : ONLINE = 3,
169 1 : FAILED = 4,
170 1 : }
171 :
172 : /// Is a connection connected or connecting? An open connection keeps its protocol and can't be merged.
173 1 : export function isConnectionOpen(health: ConnectionHealth | undefined): boolean {
174 1 : return health === ConnectionHealth.CONNECTING || health === ConnectionHealth.ONLINE;
175 1 : }
176 :
177 1 : export function printConnectionHealth(health: ConnectionHealth) {
178 1 : switch (health) {
179 1 : case ConnectionHealth.NOT_STARTED:
180 1 : return "Not Started";
181 0 : case ConnectionHealth.CONNECTING:
182 0 : return "Connecting";
183 0 : case ConnectionHealth.CANCELLED:
184 0 : return "Cancelled";
185 0 : case ConnectionHealth.ONLINE:
186 0 : return "Online";
187 0 : case ConnectionHealth.FAILED:
188 0 : return "Failed";
189 1 : }
190 1 : }
191 :
192 : export type ConnectionStateWithoutId = Omit<ConnectionState, "connectionId">;
193 :
194 1 : export const DELETE_CONNECTION = Symbol('DELETE_CONNECTION');
195 1 : export const RESET_CONNECTION = Symbol('RESET_CONNECTION');
196 1 : export const SWITCH_CONNECTOR_TYPE = Symbol('SWITCH_CONNECTOR_TYPE');
197 1 : export const SET_CATALOG_SCRIPT = Symbol('SET_CATALOG_SCRIPT');
198 1 : export const APPLY_CATALOG_CHANGES = Symbol('APPLY_CATALOG_CHANGES');
199 1 : export const UPDATE_CATALOG = Symbol('UPDATE_CATALOG');
200 : export const CATALOG_UPDATE_STARTED = Symbol('CATALOG_UPDATE_STARTED');
201 1 : export const CATALOG_UPDATE_REGISTER_QUERY = Symbol('CATALOG_UPDATE_REGISTER_QUERY');
202 1 : export const CATALOG_UPDATE_SCHEMA_SCRIPT = Symbol('CATALOG_UPDATE_SCHEMA_SCRIPT');
203 1 : export const CATALOG_UPDATE_SUCCEEDED = Symbol('CATALOG_UPDATE_SUCCEEDED');
204 1 : export const CATALOG_UPDATE_PARTIALLY_SUCCEEDED = Symbol('CATALOG_UPDATE_PARTIALLY_SUCCEEDED');
205 1 : export const CATALOG_UPDATE_FAILED = Symbol('CATALOG_UPDATE_FAILED');
206 1 : export const CATALOG_UPDATE_CANCELLED = Symbol('CATALOG_UPDATE_CANCELLED');
207 :
208 1 : export const HEALTH_CHECK_STARTED = Symbol('HEALTH_CHECK_STARTED');
209 1 : export const HEALTH_CHECK_CANCELLED = Symbol('HEALTH_CHECK_CANCELLED');
210 1 : export const HEALTH_CHECK_SUCCEEDED = Symbol('HEALTH_CHECK_SUCCEEDED');
211 1 : export const HEALTH_CHECK_FAILED = Symbol('HEALTH_CHECK_FAILED');
212 1 : export const SET_CONNECTION_DRAFT = Symbol('SET_CONNECTION_DRAFT');
213 1 : export const DISCARD_CONNECTION_DRAFT = Symbol('DISCARD_CONNECTION_DRAFT');
214 1 : export const COMMIT_CONNECTION_PARAMS = Symbol('COMMIT_CONNECTION_PARAMS');
215 1 : export const RENAME_CONNECTION = Symbol('RENAME_CONNECTION');
216 1 : export const SET_CONNECTION_QUERY_RESULT_CACHING = Symbol('SET_CONNECTION_QUERY_RESULT_CACHING');
217 :
218 : export type CatalogAction =
219 : | VariantKind<typeof SET_CATALOG_SCRIPT, core.DashQLScript>
220 : | VariantKind<typeof APPLY_CATALOG_CHANGES, core.buffers.catalog.CatalogChangeT[]>
221 : | VariantKind<typeof UPDATE_CATALOG, [number, CatalogUpdateTaskState]>
222 : | VariantKind<typeof CATALOG_UPDATE_REGISTER_QUERY, [number, number]>
223 : | VariantKind<typeof CATALOG_UPDATE_SCHEMA_SCRIPT, [number]>
224 : | VariantKind<typeof CATALOG_UPDATE_CANCELLED, [number, Error]>
225 : | VariantKind<typeof CATALOG_UPDATE_FAILED, [number, Error]>
226 : | VariantKind<typeof CATALOG_UPDATE_SUCCEEDED, [number]>
227 : | VariantKind<typeof CATALOG_UPDATE_PARTIALLY_SUCCEEDED, [number, Error]>
228 : ;
229 :
230 : /// The health check of a connection, every connector runs it the same way
231 : export type HealthCheckAction =
232 : | VariantKind<typeof HEALTH_CHECK_STARTED, null>
233 : | VariantKind<typeof HEALTH_CHECK_CANCELLED, null>
234 : | VariantKind<typeof HEALTH_CHECK_FAILED, DetailedError>
235 : | VariantKind<typeof HEALTH_CHECK_SUCCEEDED, null>
236 : ;
237 :
238 : export type ConnectionAction =
239 : | VariantKind<typeof SET_CONNECTION_DRAFT, ConnectionParams>
240 : | VariantKind<typeof DISCARD_CONNECTION_DRAFT, null>
241 : | VariantKind<typeof COMMIT_CONNECTION_PARAMS, ConnectionParams>
242 : | VariantKind<typeof RENAME_CONNECTION, string>
243 : | VariantKind<typeof SET_CONNECTION_QUERY_RESULT_CACHING, boolean>
244 : | VariantKind<typeof DELETE_CONNECTION, null>
245 : | VariantKind<typeof RESET_CONNECTION, null>
246 : | VariantKind<typeof SWITCH_CONNECTOR_TYPE, ConnectorType>
247 : | HealthCheckAction
248 : | CatalogAction
249 : | QueryExecutionAction
250 : | HyperConnectorAction
251 : | TrinoConnectorAction
252 : | SalesforceConnectionStateAction
253 : ;
254 :
255 : /// Does a connection cache query results? The connector decides unless the user chose.
256 1 : export function connectionCachesQueryResults(state: Pick<ConnectionState, 'cacheQueryResults' | 'connectorInfo'>): boolean {
257 1 : return state.cacheQueryResults ?? state.connectorInfo.features.queryResultCaching;
258 1 : }
259 :
260 1 : export function reduceConnectionState(state: ConnectionState, action: ConnectionAction, logger: Logger): ConnectionState {
261 1 : switch (action.type) {
262 1 : case SET_CONNECTION_DRAFT: {
263 1 : if (getConnectorInfoForParams(action.value)?.connectorType !== state.connectorInfo.connectorType) return state;
264 1 : const paramsDraft = structuredClone(action.value);
265 1 : if (connectionParamsEqual(state.paramsDraft, paramsDraft)) return state;
266 1 : return { ...state, paramsDraft };
267 1 : }
268 : // COMMIT_CONNECTION_PARAMS makes saved params the params of the connection.
269 : // The draft is cleared unless it changed while saving.
270 1 : case COMMIT_CONNECTION_PARAMS: {
271 1 : const params = connectionParamsWithoutDraftRows(structuredClone(action.value));
272 1 : const draftSaved = state.paramsDraft == null
273 1 : || connectionParamsEqual(connectionParamsWithoutDraftRows(state.paramsDraft), params);
274 1 : if (connectionParamsEqual(state.params, params) && (state.paramsDraft == null || !draftSaved)) return state;
275 1 : return { ...state, params, paramsDraft: draftSaved ? undefined : state.paramsDraft };
276 1 : }
277 : // A blank name clears the name
278 1 : case RENAME_CONNECTION: {
279 1 : const name = action.value.trim() || undefined;
280 1 : if (name === state.name) return state;
281 1 : return { ...state, name };
282 1 : }
283 : // The connection keeps the choice, also when it matches the default of the connector
284 1 : case SET_CONNECTION_QUERY_RESULT_CACHING: {
285 1 : if (state.cacheQueryResults === action.value) return state;
286 1 : return { ...state, cacheQueryResults: action.value };
287 : }
288 1 : case DISCARD_CONNECTION_DRAFT: {
289 1 : if (state.connectionHealth === ConnectionHealth.CONNECTING) return state;
290 1 : const details = {
291 1 : ...state.details,
292 1 : value: {
293 1 : ...state.details.value,
294 1 : proto: {
295 1 : ...state.details.value.proto,
296 1 : setupError: undefined,
297 1 : channelError: undefined,
298 1 : schemaResolutionError: undefined,
299 1 : healthCheckError: undefined,
300 1 : },
301 1 : },
302 1 : } as unknown as ConnectionStateDetailsVariant;
303 1 : return { ...state, paramsDraft: undefined, details };
304 1 : }
305 1 : case SET_CATALOG_SCRIPT:
306 1 : return {
307 1 : ...state,
308 1 : catalogRelationScript: action.value,
309 1 : };
310 :
311 1 : case APPLY_CATALOG_CHANGES:
312 1 : return applyCatalogChanges(state, action.value, logger);
313 :
314 1 : case UPDATE_CATALOG:
315 1 : case CATALOG_UPDATE_REGISTER_QUERY:
316 1 : case CATALOG_UPDATE_SCHEMA_SCRIPT:
317 1 : case CATALOG_UPDATE_CANCELLED:
318 1 : case CATALOG_UPDATE_SUCCEEDED:
319 1 : case CATALOG_UPDATE_PARTIALLY_SUCCEEDED:
320 1 : case CATALOG_UPDATE_FAILED:
321 1 : return reduceCatalogAction(state, action);
322 :
323 1 : case EXECUTE_QUERY:
324 1 : case QUERY_PREPARING:
325 1 : case QUERY_RUNNING:
326 1 : case QUERY_SENDING:
327 1 : case QUERY_PROGRESS_UPDATED:
328 1 : case QUERY_STATEMENT_STARTED:
329 1 : case QUERY_STATEMENT_SUCCEEDED:
330 1 : case QUERY_RECEIVED_BATCH:
331 1 : case QUERY_RECEIVED_ALL_BATCHES:
332 1 : case QUERY_PROCESSING_RESULTS:
333 1 : case QUERY_PROCESSED_RESULTS:
334 1 : case QUERY_SUCCEEDED:
335 1 : case QUERY_CANCELLED:
336 1 : case QUERY_FAILED:
337 1 : case QUERY_CACHE_RECORDED:
338 1 : case QUERY_CACHE_DELETED:
339 1 : return reduceQueryAction(state, action);
340 :
341 : // RESET_CONNECTION stops the connection and clears its catalog
342 1 : case RESET_CONNECTION: {
343 : // Pending catalog publishers are aborted before the catalog is cleared
344 1 : const newState = stopConnection(state, action, 'reset');
345 1 : state.catalog.clear();
346 1 : return newState;
347 1 : }
348 :
349 : // SWITCH_CONNECTOR_TYPE changes the connection's type in-place.
350 : // Only allowed when the connection hasn't been configured yet.
351 1 : case SWITCH_CONNECTOR_TYPE: {
352 1 : if (state.connectionStatus !== ConnectionStatus.NOT_STARTED || state.active || state.paramsDraft != null) {
353 1 : return state;
354 : }
355 1 : const newType = action.value;
356 1 : if (state.connectorInfo.connectorType === newType) {
357 1 : return state;
358 : }
359 1 : const newInfo = CONNECTOR_INFOS[newType as number];
360 1 : const newDetails = createConnectionStateDetails(newType);
361 1 : return {
362 1 : ...state,
363 1 : connectorInfo: newInfo,
364 1 : details: newDetails,
365 1 : params: getConnectionParamsFromStateDetails(newDetails)!,
366 1 : paramsDraft: undefined,
367 1 : setupAttemptParams: undefined,
368 1 : setupAttemptDraft: undefined,
369 1 : };
370 1 : }
371 :
372 :
373 : /// DELETE_CONNECTION stops the connection and destroys its catalog.
374 : /// The registry refuses to delete connections that sessions reference.
375 1 : case DELETE_CONNECTION: {
376 1 : const newState = stopConnection(state, action, 'deleted');
377 :
378 : // Text-only source scripts do not own their published descriptors.
379 1 : try {
380 1 : state.catalog.dropDescriptor(state.catalogRelationScript.catalog_entry_id);
381 1 : } catch (e) {
382 : // Descriptor may have already been dropped - ignore error
383 1 : }
384 1 : state.catalogRelationScript.destroy();
385 1 : try {
386 1 : state.catalog.dropDescriptor(state.catalogFunctionScript.catalog_entry_id);
387 1 : } catch (e) {
388 : // Descriptor may have already been dropped - ignore error
389 1 : }
390 1 : state.catalogFunctionScript.destroy();
391 :
392 : // Delete the conneciton catalog
393 1 : state.catalog.destroy();
394 :
395 1 : return newState;
396 1 : }
397 :
398 1 : default: {
399 : // Capture setup input before connector reducers replace runtime setupParams.
400 1 : let setupAttemptParams = state.setupAttemptParams;
401 1 : let setupAttemptDraft = state.setupAttemptDraft;
402 1 : switch (action.type) {
403 1 : case HYPER_CHANNEL_SETUP_STARTED:
404 1 : setupAttemptParams = structuredClone({ hyper: action.value });
405 1 : setupAttemptDraft = state.paramsDraft;
406 1 : break;
407 1 : case OAUTH_STARTED:
408 1 : case TRINO_CHANNEL_SETUP_STARTED:
409 1 : setupAttemptParams = structuredClone({ trino: action.value });
410 1 : setupAttemptDraft = state.setupAttemptParams == null ? state.paramsDraft : state.setupAttemptDraft;
411 1 : break;
412 1 : case SETUP_STARTED:
413 1 : setupAttemptParams = structuredClone(connectionParamsOf(connectorOfDetails(state.details), action.value));
414 1 : setupAttemptDraft = state.paramsDraft;
415 : break;
416 1 : }
417 1 : let newState = reduceHealthCheck(state, action) ?? reduceConnectorDetails(state, action);
418 1 : if (newState == null) {
419 1 : throw new Error(`failed to apply state action: ${String(action.type)}`);
420 : }
421 1 : newState = { ...newState, setupAttemptParams, setupAttemptDraft };
422 1 : if (newState.connectionHealth === ConnectionHealth.ONLINE
423 1 : && state.connectionHealth !== ConnectionHealth.ONLINE && setupAttemptParams != null) {
424 1 : newState = {
425 1 : ...newState,
426 1 : params: connectionParamsWithoutDraftRows(setupAttemptParams),
427 1 : paramsDraft: connectionParamsEqual(state.paramsDraft, setupAttemptDraft)
428 1 : && (setupAttemptDraft == null || connectionParamsEqual(connectionParamsWithoutDraftRows(setupAttemptDraft), connectionParamsWithoutDraftRows(setupAttemptParams)))
429 1 : ? undefined : state.paramsDraft,
430 1 : setupAttemptParams: undefined,
431 1 : setupAttemptDraft: undefined,
432 1 : };
433 1 : const connector = connectorOfDetails(newState.details);
434 1 : if (isSalesforceDetails(newState.details) && connector.paramsKey in setupAttemptParams) {
435 1 : const salesforceParams = connectorParams(connector, setupAttemptParams) as connection.SalesforceConnectionParams;
436 1 : newState.params = connectionParamsOf(connector, {
437 1 : ...salesforceParams,
438 1 : login: newState.details.value.proto.setupParams?.login ?? salesforceParams.login,
439 1 : });
440 1 : }
441 1 : } else if (newState.connectionHealth === ConnectionHealth.FAILED || newState.connectionHealth === ConnectionHealth.CANCELLED) {
442 1 : newState.setupAttemptParams = undefined;
443 1 : newState.setupAttemptDraft = undefined;
444 1 : }
445 :
446 : // Embedded connectors become online as soon as their local channel is ready, without a
447 : // health-check action. Activate every database on its first transition to online.
448 1 : if (newState.connectionHealth === ConnectionHealth.ONLINE && !newState.active) {
449 1 : newState = { ...newState, active: true };
450 : }
451 :
452 1 : return newState;
453 1 : }
454 1 : }
455 1 : }
456 :
457 : /// Apply a health check action, null for other actions
458 1 : function reduceHealthCheck(state: ConnectionState, action: ConnectionAction): ConnectionState | null {
459 1 : const now = dateToTimestamp(new Date());
460 1 : switch (action.type) {
461 1 : case HEALTH_CHECK_STARTED:
462 1 : return withConnectorProto({ ...state, connectionStatus: ConnectionStatus.HEALTH_CHECK_STARTED, connectionHealth: ConnectionHealth.CONNECTING },
463 1 : proto => ({ ...proto, setupTimings: { ...proto.setupTimings, healthCheckStartedAt: now } }));
464 1 : case HEALTH_CHECK_FAILED:
465 1 : return withConnectorProto({ ...state, connectionStatus: ConnectionStatus.HEALTH_CHECK_FAILED, connectionHealth: ConnectionHealth.FAILED },
466 1 : proto => ({ ...proto, setupTimings: { ...proto.setupTimings, healthCheckFailedAt: now }, healthCheckError: action.value }));
467 1 : case HEALTH_CHECK_CANCELLED:
468 1 : return withConnectorProto({ ...state, connectionStatus: ConnectionStatus.HEALTH_CHECK_CANCELLED, connectionHealth: ConnectionHealth.CANCELLED },
469 1 : proto => ({ ...proto, setupTimings: { ...proto.setupTimings, healthCheckCancelledAt: now } }));
470 1 : case HEALTH_CHECK_SUCCEEDED:
471 1 : return withConnectorProto({ ...state, connectionStatus: ConnectionStatus.HEALTH_CHECK_SUCCEEDED, connectionHealth: ConnectionHealth.ONLINE },
472 1 : proto => ({ ...proto, setupTimings: { ...proto.setupTimings, healthCheckSucceededAt: now } }));
473 1 : default:
474 1 : return null;
475 1 : }
476 1 : }
477 :
478 : /// The setup timings and errors that the details of every connector keep
479 : type ConnectorProto = Pick<ConnectionStateDetailsVariant['value']['proto'], 'setupTimings' | 'healthCheckError'>;
480 :
481 : /// Update the setup timings and errors of the connector details
482 1 : function withConnectorProto(state: ConnectionState, patch: (proto: ConnectorProto) => ConnectorProto): ConnectionState {
483 1 : const details = state.details.value;
484 1 : return {
485 1 : ...state,
486 1 : details: { ...state.details, value: { ...details, proto: { ...details.proto, ...patch(details.proto) } } } as ConnectionStateDetailsVariant,
487 1 : };
488 1 : }
489 :
490 : /// Let the reducer of the connector apply an action to the connection details
491 1 : function reduceConnectorDetails(state: ConnectionState, action: ConnectionAction): ConnectionState | null {
492 1 : return connectorOfDetails(state.details).reduce(state, action);
493 1 : }
494 :
495 : /// Stop a connection that is reset or deleted.
496 : /// Cancels the setup and the running catalog updates and drops the query executions.
497 1 : function stopConnection(state: ConnectionState, action: ConnectionAction, reason: 'reset' | 'deleted'): ConnectionState {
498 1 : cancelConnectionSetup(state.connectionId);
499 1 : invalidateCatalogBatch(state.catalog);
500 1 : for (const task of state.catalogUpdates.tasksRunning.values()) {
501 1 : task.cancellation.abort(new DOMException(`Connection was ${reason}`, 'AbortError'));
502 : }
503 : // The queries run on the channel of the stopped connection, the executor stops them at the backend
504 1 : for (const query of state.queriesActive.values()) {
505 1 : query.cancellation.abort(new DOMException(`Connection was ${reason}`, 'AbortError'));
506 : }
507 1 : const cleaned: ConnectionState = {
508 1 : ...state,
509 1 : connectionStatus: ConnectionStatus.NOT_STARTED,
510 1 : connectionHealth: ConnectionHealth.NOT_STARTED,
511 1 : setupAttemptParams: undefined,
512 1 : setupAttemptDraft: undefined,
513 1 : metrics: createConnectionMetrics(),
514 1 : catalogUpdates: {
515 1 : tasksRunning: new Map(),
516 1 : tasksFinished: new Map(),
517 1 : currentFullRefresh: null,
518 1 : lastFullRefresh: null,
519 1 : restoredAt: null,
520 1 : },
521 1 : queriesActive: new Map(),
522 1 : queriesActiveOrdered: [],
523 1 : queriesFinished: new Map(),
524 1 : queriesFinishedOrdered: [],
525 1 : snapshotQueriesActiveFinished: state.snapshotQueriesActiveFinished + 1,
526 1 : };
527 : // Cleaning up the details is best-effort, the connector may ignore the action
528 1 : return reduceConnectorDetails(cleaned, action) ?? cleaned;
529 1 : }
530 :
531 : /// Apply the catalog changes of executed statements to the relation descriptor and record them as SQL.
532 : /// The changes are qualified with the default database and schema of the connection.
533 : ///
534 : /// Known limitation: the changes don't distinguish temporary relations. A `CREATE TEMP TABLE` is
535 : /// recorded and persisted like any other relation, so it outlives the database session it belongs to
536 : /// and is restored with the catalog after a reconnect.
537 1 : function applyCatalogChanges(state: ConnectionState, changes: core.buffers.catalog.CatalogChangeT[], logger: Logger): ConnectionState {
538 1 : if (changes.length === 0) return state;
539 1 : try {
540 1 : const qualified = state.instance.qualifyCatalogChanges(changes, catalogNameDefaults(state.details));
541 1 : state.catalog.applyCatalogChanges(state.catalogRelationScript.catalog_entry_id, core.buffers.catalog.CatalogLayer.CONNECTION, qualified);
542 : // A running refresh may have read the metadata before the statements ran, publishing it would undo the changes
543 1 : invalidateCatalogBatch(state.catalog);
544 1 : const catalogChangesSql = state.instance.appendCatalogChangesSql(state.catalogChangesSql, qualified);
545 1 : logger.info('Applied catalog changes of executed statements', {
546 1 : connectionId: state.connectionId,
547 1 : changes: qualified.length.toString(),
548 1 : }, LOG_CTX);
549 1 : return { ...state, catalogChangesSql };
550 1 : } catch (e: unknown) {
551 1 : logger.error('Failed to apply catalog changes of executed statements', {
552 1 : connectionId: state.connectionId,
553 1 : error: stringifyError(e),
554 1 : }, LOG_CTX);
555 1 : return state;
556 1 : }
557 1 : }
558 :
559 :
560 1 : export function createConnectionMetrics(): connection.ConnectionMetrics {
561 1 : return createQueryExecutionMetrics();
562 1 : }
563 :
564 : /// The optional properties of a new connection state
565 : export interface ConnectionStateOptions {
566 : /// Does the connection belong to a committed session or was it connected once?
567 : active?: boolean;
568 : /// The params, derived from the details by default
569 : params?: ConnectionParams;
570 : /// The draft of the params
571 : paramsDraft?: ConnectionParams;
572 : }
573 :
574 : /// Create the state of a connection with an empty catalog
575 1 : export function createConnectionState(dql: core.DashQL, info: ConnectorInfo, details: ConnectionStateDetailsVariant, options: ConnectionStateOptions = {}): ConnectionStateWithoutId {
576 1 : let catalog: core.DashQLCatalog | null = null;
577 1 : let catalogRelationScript: core.DashQLScript | null = null;
578 1 : let catalogFunctionScript: core.DashQLScript | null = null;
579 1 : try {
580 1 : catalog = dql.createCatalog();
581 1 : catalogRelationScript = dql.createScript(catalog);
582 1 : catalogRelationScript.replaceText(generateCatalogScriptHeader(CatalogSource.Unknown));
583 1 : catalogFunctionScript = dql.createScript(catalog);
584 1 : catalogFunctionScript.replaceText(generateFunctionScriptHeader(CatalogSource.Unknown));
585 1 : } catch (error) {
586 1 : catalogFunctionScript?.destroy();
587 1 : catalogRelationScript?.destroy();
588 1 : catalog?.destroy();
589 1 : throw error;
590 1 : }
591 1 : return {
592 1 : instance: dql,
593 1 : active: options.active ?? false,
594 1 : connectionStatus: ConnectionStatus.NOT_STARTED,
595 1 : connectionHealth: ConnectionHealth.NOT_STARTED,
596 1 : connectorInfo: info,
597 1 : metrics: createConnectionMetrics(),
598 1 : details,
599 1 : params: options.params == null ? getConnectionParamsFromStateDetails(details)! : structuredClone(options.params),
600 1 : paramsDraft: options.paramsDraft == null ? undefined : structuredClone(options.paramsDraft),
601 1 : catalog,
602 1 : catalogUpdates: {
603 1 : tasksRunning: new Map(),
604 1 : tasksFinished: new Map(),
605 1 : currentFullRefresh: null,
606 1 : lastFullRefresh: null,
607 1 : restoredAt: null,
608 1 : },
609 1 : catalogRelationScript,
610 1 : catalogFunctionScript,
611 1 : catalogChangesSql: '',
612 1 : snapshotQueriesActiveFinished: 1,
613 1 : queriesActive: new Map(),
614 1 : queriesActiveOrdered: [],
615 1 : queriesFinished: new Map(),
616 1 : queriesFinishedOrdered: [],
617 1 : };
618 1 : }
|
