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 { beginCatalogBatch, invalidateCatalogBatch, publishCatalogBatch, type BatchProcessor, type CatalogBatchEntry } from './catalog/catalog_batch.js';
9 : import type { SessionConnection, SessionRegistryBackend } from '../storage/storage_backend.js';
10 : import { STORAGE_CATALOG_CHANGES, STORAGE_CATALOG_FUNCTIONS, STORAGE_CATALOG_RELATIONS } from '../storage/storage_backend.js';
11 :
12 1 : const LOG_CTX = "connection_restore";
13 :
14 : /// The stored params of a connection, as the app manifest or a session file holds them
15 : export type StoredConnection = Pick<SessionConnection, 'params' | 'paramsDraft'> & {
16 : /// The name of the connection, the app manifest keeps it, an imported session brings it along
17 : name?: string;
18 : /// Does the connection cache query results? The app manifest keeps it, absent for the default of the connector.
19 : cacheQueryResults?: boolean;
20 : };
21 :
22 : /// Create the state of a connection from its stored params.
23 : /// The connection is neither registered nor written, see `destroyUnregisteredConnection`.
24 1 : export function connectionStateFromStorage(core: DashQL, connectionId: string, stored: StoredConnection): ConnectionState {
25 1 : const [connectorInfo, details] = decodeConnectionFromProto(stored.params as any, connectionId);
26 1 : const state = restoreConnectionState(core, connectionId, connectorInfo, details, stored.paramsDraft);
27 1 : return {
28 1 : ...state,
29 1 : ...(stored.name == null ? {} : { name: stored.name }),
30 1 : ...(stored.cacheQueryResults == null ? {} : { cacheQueryResults: stored.cacheQueryResults }),
31 1 : };
32 1 : }
33 :
34 : /// Destroy a connection that was created but never registered
35 1 : export function destroyUnregisteredConnection(connection: ConnectionState): void {
36 1 : invalidateCatalogBatch(connection.catalog);
37 1 : connection.catalog.dropDescriptor(connection.catalogRelationScript.catalog_entry_id);
38 1 : connection.catalog.dropDescriptor(connection.catalogFunctionScript.catalog_entry_id);
39 1 : connection.catalogRelationScript.destroy();
40 1 : connection.catalogFunctionScript.destroy();
41 1 : connection.catalog.destroy();
42 1 : }
43 :
44 : /// What restoring a catalog needs
45 : export interface CatalogRestoreContext {
46 : /// The core instance
47 : core: DashQL;
48 : /// The logger
49 : logger: Logger;
50 : /// The processor of catalog batches, the core processes them if absent
51 : processor?: BatchProcessor;
52 : }
53 :
54 : /// The outcome of restoring the catalog of a connection from its files
55 : export interface ConnectionCatalogRestore {
56 : /// False if the restoration failed, the connection refreshes its catalog on connect
57 : succeeded: boolean;
58 : /// The files whose relations or changes were dropped, they are cleared
59 : staleFiles: string[];
60 : }
61 :
62 : /// Restore the catalog of a connection from the files of the connection.
63 : ///
64 : /// The relations and changes of a connection whose database doesn't outlive the app describe a
65 : /// database that is gone, they are dropped and their files are reported as stale. Such a catalog is
66 : /// described by a refresh on connect. A restored catalog is kept until an explicit refresh.
67 1 : export async function restoreConnectionCatalog(
68 1 : ctx: CatalogRestoreContext,
69 1 : registry: SessionRegistryBackend,
70 1 : connection: ConnectionState,
71 1 : logContext: Record<string, string>,
72 1 : ): Promise<ConnectionCatalogRestore> {
73 1 : const { connectionId, catalog, catalogRelationScript, catalogFunctionScript } = connection;
74 1 : const durable = connectionCatalogIsDurable(connection.params);
75 1 : const staleFiles: string[] = [];
76 1 : try {
77 1 : const validate = beginCatalogBatch(catalog);
78 1 : const [relations, functions, changes] = await Promise.all([
79 1 : registry.loadConnectionFile(connectionId, STORAGE_CATALOG_RELATIONS),
80 1 : registry.loadConnectionFile(connectionId, STORAGE_CATALOG_FUNCTIONS),
81 1 : registry.loadConnectionFile(connectionId, STORAGE_CATALOG_CHANGES),
82 1 : ]);
83 1 : validate();
84 1 : if (!durable) {
85 : // Clear the dropped relations and changes that older versions of the app wrote
86 1 : if (relations) staleFiles.push(STORAGE_CATALOG_RELATIONS);
87 1 : if (changes) staleFiles.push(STORAGE_CATALOG_CHANGES);
88 1 : }
89 :
90 1 : const catalogScripts: CatalogBatchEntry[] = [];
91 1 : if (durable && relations && relations.trim().length > 0) {
92 1 : ctx.logger.info("Catalog schema loaded", { ...logContext, schemaLength: relations.length.toString() }, LOG_CTX);
93 1 : catalogScripts.push({ script: catalogRelationScript, text: relations });
94 1 : }
95 1 : if (functions && functions.trim().length > 0) {
96 1 : ctx.logger.info("Catalog functions loaded", { ...logContext, functionsLength: functions.length.toString() }, LOG_CTX);
97 1 : catalogScripts.push({ script: catalogFunctionScript, text: functions });
98 1 : }
99 1 : if (catalogScripts.length > 0) {
100 1 : await publishCatalogBatch(ctx.core, catalog, catalogScripts, { processor: ctx.processor }, validate);
101 : // Only restored relations make a catalog that is kept
102 1 : if (durable) {
103 1 : connection.catalogUpdates.restoredAt = new Date();
104 : }
105 1 : }
106 1 : if (durable && changes != null && changes.trim().length > 0) {
107 1 : restoreCatalogChanges(ctx.logger, connection, changes, logContext);
108 : }
109 1 : return { succeeded: true, staleFiles };
110 1 : } catch (error) {
111 1 : ctx.logger.warn("Failed to restore connection catalog, will refresh on connect", {
112 1 : ...logContext,
113 1 : error: stringifyError(error),
114 1 : }, LOG_CTX);
115 1 : return { succeeded: false, staleFiles };
116 1 : }
117 1 : }
118 :
119 : /// Apply the catalog changes of executed statements on top of the restored relation script.
120 : /// Unreadable changes are dropped, the next catalog refresh describes the relations anyway.
121 1 : function restoreCatalogChanges(logger: Logger, connection: ConnectionState, changesSql: string, logContext: Record<string, string>): void {
122 1 : try {
123 1 : connection.catalog.applyCatalogChangesSql(connection.catalogRelationScript.catalog_entry_id, buffers.catalog.CatalogLayer.CONNECTION, changesSql);
124 1 : connection.catalogChangesSql = changesSql;
125 1 : logger.info("Catalog changes restored", { ...logContext, changesLength: changesSql.length.toString() }, LOG_CTX);
126 1 : } catch (error) {
127 0 : logger.warn("Failed to restore the catalog changes", { ...logContext, error: stringifyError(error) }, LOG_CTX);
128 0 : }
129 1 : }
|
