Line data Source code
1 : import * as buffers from '@ankoh/dashql-core/buffers';
2 : import type { DashQL } from '@ankoh/dashql-core';
3 : import type { Logger } from '../platform/logger/logger.js';
4 : import { stringifyError } from '../platform/logger/logger.js';
5 : import type { ConnectionState } from './connection_state.js';
6 : import { decodeConnectionFromProto, restoreConnectionState } from './connection_import.js';
7 : import { connectionCatalogIsDurable } from './connection_params.js';
8 : import { SALESFORCE_DATA_CLOUD_CONNECTOR } from './connector_info.js';
9 : import { beginCatalogBatch, invalidateCatalogBatch, publishCatalogBatch, type BatchProcessor, type CatalogBatchEntry } from './catalog/catalog_batch.js';
10 : import type { SessionConnection, SessionRegistryBackend } from '../storage/storage_backend.js';
11 : import { STORAGE_CATALOG_CHANGES, STORAGE_CATALOG_FUNCTIONS, STORAGE_CATALOG_RELATIONS, STORAGE_CATALOG_STATE } from '../storage/storage_backend.js';
12 : import { packRelationName, parseCatalogStateFile, readRelationName } from './catalog/catalog_sql_generator.js';
13 : import { mergeRelationNames } from './catalog/catalog_update_state.js';
14 :
15 1 : const LOG_CTX = "connection_restore";
16 :
17 : /// The stored params of a connection, as the app manifest or a session file holds them
18 : export type StoredConnection = Pick<SessionConnection, 'params' | 'paramsDraft'> & {
19 : /// The name of the connection, the app manifest keeps it, an imported session brings it along
20 : name?: string;
21 : /// Does the connection cache query results? The app manifest keeps it, absent for the default of the connector.
22 : cacheQueryResults?: boolean;
23 : };
24 :
25 : /// Create the state of a connection from its stored params.
26 : /// The connection is neither registered nor written, see `destroyUnregisteredConnection`.
27 1 : export function connectionStateFromStorage(core: DashQL, connectionId: string, stored: StoredConnection): ConnectionState {
28 1 : const [connectorInfo, details] = decodeConnectionFromProto(stored.params as any, connectionId);
29 1 : const state = restoreConnectionState(core, connectionId, connectorInfo, details, stored.paramsDraft);
30 1 : return {
31 1 : ...state,
32 1 : ...(stored.name == null ? {} : { name: stored.name }),
33 1 : ...(stored.cacheQueryResults == null ? {} : { cacheQueryResults: stored.cacheQueryResults }),
34 1 : };
35 1 : }
36 :
37 : /// Destroy a connection that was created but never registered
38 1 : export function destroyUnregisteredConnection(connection: ConnectionState): void {
39 1 : invalidateCatalogBatch(connection.catalog);
40 1 : connection.catalog.dropDescriptor(connection.catalogRelationEntryId);
41 1 : connection.catalog.dropDescriptor(connection.catalogFunctionScript.catalog_entry_id);
42 1 : connection.catalogFunctionScript.destroy();
43 1 : connection.catalog.destroy();
44 1 : }
45 :
46 : /// What restoring a catalog needs
47 : export interface CatalogRestoreContext {
48 : /// The core instance
49 : core: DashQL;
50 : /// The logger
51 : logger: Logger;
52 : /// The processor of catalog batches, the core processes them if absent
53 : processor?: BatchProcessor;
54 : }
55 :
56 : /// The outcome of restoring the catalog of a connection from its files
57 : export interface ConnectionCatalogRestore {
58 : /// False if the restoration failed, the connection refreshes its catalog on connect
59 : succeeded: boolean;
60 : /// The files to write: those whose relations or changes were dropped are cleared
61 : files: Array<{ fileName: string; text: string }>;
62 : }
63 :
64 : /// Restore the catalog of a connection from the files of the connection.
65 : ///
66 : /// The relations of a connection whose database doesn't outlive the app describe a database that is
67 : /// gone, they are dropped and their file is reported as stale. Such a catalog is described by a refresh
68 : /// on connect. A restored catalog is kept until an explicit refresh.
69 : ///
70 : /// The relations the connection didn't answer about yet are asked about again once it is online, and those it reported as
71 : /// gone stay gone. Older versions of the app kept the changes of executed statements in a file of their own. The relations
72 : /// of that file count as not answered, the file is cleared.
73 1 : export async function restoreConnectionCatalog(
74 1 : ctx: CatalogRestoreContext,
75 1 : registry: SessionRegistryBackend,
76 1 : connection: ConnectionState,
77 1 : logContext: Record<string, string>,
78 1 : ): Promise<ConnectionCatalogRestore> {
79 1 : const { connectionId, catalog, catalogRelationEntryId, catalogFunctionScript } = connection;
80 1 : const durable = connectionCatalogIsDurable(connection.params);
81 1 : const files: Array<{ fileName: string; text: string }> = [];
82 1 : try {
83 1 : const validate = beginCatalogBatch(catalog);
84 1 : const [relations, functions, changes, state] = await Promise.all([
85 1 : registry.loadConnectionFile(connectionId, STORAGE_CATALOG_RELATIONS),
86 1 : registry.loadConnectionFile(connectionId, STORAGE_CATALOG_FUNCTIONS),
87 1 : registry.loadConnectionFile(connectionId, STORAGE_CATALOG_CHANGES),
88 1 : registry.loadConnectionFile(connectionId, STORAGE_CATALOG_STATE),
89 1 : ]);
90 1 : validate();
91 : // Clear the dropped relations that older versions of the app wrote
92 1 : if (!durable && relations) files.push({ fileName: STORAGE_CATALOG_RELATIONS, text: '' });
93 :
94 1 : const catalogScripts: CatalogBatchEntry[] = [];
95 1 : if (durable && relations && relations.trim().length > 0) {
96 1 : ctx.logger.info("Catalog schema loaded", { ...logContext, schemaLength: relations.length.toString() }, LOG_CTX);
97 : // The batch reads the file once, the catalog keeps the descriptor and not the text
98 1 : catalogScripts.push({ entryId: catalogRelationEntryId, text: relations });
99 1 : }
100 1 : if (functions && functions.trim().length > 0) {
101 1 : ctx.logger.info("Catalog functions loaded", { ...logContext, functionsLength: functions.length.toString() }, LOG_CTX);
102 1 : catalogScripts.push({ script: catalogFunctionScript, text: functions });
103 1 : }
104 1 : if (catalogScripts.length > 0) {
105 1 : await publishCatalogBatch(ctx.core, catalog, catalogScripts, { processor: ctx.processor }, validate);
106 : // Only restored relations make a catalog that is kept
107 1 : if (durable && relations && relations.trim().length > 0) {
108 1 : connection.catalogUpdates.described = true;
109 : }
110 1 : }
111 1 : const { pending, gone } = parseCatalogStateFile(state);
112 1 : if (durable && gone.length > 0) {
113 1 : catalog.replaceTables(catalogRelationEntryId, buffers.catalog.CatalogLayer.CONNECTION, { relations: gone.map(packRelationName), tables: [] });
114 : }
115 : // Nothing that runs on Salesforce changes its relations, and it has no lookup for them
116 1 : if (durable && connection.details.type !== SALESFORCE_DATA_CLOUD_CONNECTOR) {
117 1 : const changed = changes?.trim() ? ctx.core.touchedRelations(changes).map(name => readRelationName(name)) : [];
118 1 : connection.catalogUpdates.touchedRelations = mergeRelationNames(pending, changed);
119 1 : }
120 : // The relations of the old file move to the new one before the old one is cleared
121 1 : if (changes) {
122 1 : files.push({ fileName: STORAGE_CATALOG_STATE, text: JSON.stringify({ pending: connection.catalogUpdates.touchedRelations, gone }) });
123 1 : files.push({ fileName: STORAGE_CATALOG_CHANGES, text: '' });
124 1 : }
125 1 : return { succeeded: true, files };
126 1 : } catch (error) {
127 1 : ctx.logger.warn("Failed to restore connection catalog, will refresh on connect", {
128 1 : ...logContext,
129 1 : error: stringifyError(error),
130 1 : }, LOG_CTX);
131 1 : return { succeeded: false, files };
132 1 : }
133 1 : }
|
