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