Line data Source code
1 : import type { DashQL, FlatBufferPtr, buffers } from '@ankoh/dashql-core';
2 : import type { DataFileData, SessionData, ScriptData, StorageBackend } from './storage_backend.js';
3 : import {
4 : isSessionRegistryBackend,
5 : STORAGE_CATALOG_FUNCTIONS,
6 : STORAGE_CATALOG_RELATIONS,
7 : STORAGE_CONNECTIONS_FOLDER,
8 : STORAGE_SESSION_FILE,
9 : StorageBackendType,
10 : validateDataFileName,
11 : } from './storage_backend.js';
12 : import type { MigrationFolder } from './migration/migration_folder.js';
13 : import { migrateSessionFolder, SessionMigrationResult } from './migration/session_migration.js';
14 : import { SESSION_CATALOG_COPY_FORMAT_VERSION } from './migration/versions.js';
15 : import {
16 : describeSessionValidationError,
17 : isValidUuid,
18 : SessionValidationError,
19 : validateSessionData,
20 : } from './session_validation.js';
21 : import { reconcileSessionScripts, remapSessionScripts, sanitizeSessionFiles, sessionAttachments, sessionConnectionIdsOf } from './session_files.js';
22 :
23 : /// The durable, portable subset of a session. Derived cache files and storage-location metadata
24 : /// are intentionally not represented here.
25 : export interface SessionBundle {
26 : /// The session file. Its script list names every script of `scripts` in feed order and its
27 : /// attachment list names every file of `data`.
28 : session: SessionData;
29 : /// The catalogs of the connections of the session. Sessions don't store a catalog, bundles carry
30 : /// the catalogs of their connections so that a connection new to the app starts with one.
31 : catalogs: SessionBundleCatalog[];
32 : /// The script files
33 : scripts: ScriptData[];
34 : /// The attachments
35 : data?: DataFileData[];
36 : }
37 :
38 : /// The catalog of a connection that a bundle carries
39 : export interface SessionBundleCatalog {
40 : /// The connection
41 : connectionId: string;
42 : /// The catalog relations
43 : relations: string | null;
44 : /// The catalog functions
45 : functions: string | null;
46 : }
47 :
48 : /// The path of a catalog file of a connection in a bundle, e.g. `connections/<id>/dashql-catalog-relations.sql`.
49 : /// The app ignores the folder of a session folder except when the session is imported.
50 1 : export function bundleCatalogPath(connectionId: string, fileName: string): string {
51 1 : return `${STORAGE_CONNECTIONS_FOLDER}/${connectionId}/${fileName}`;
52 1 : }
53 :
54 : /// The format version of the session file of a folder, null if it has none
55 1 : async function readFormatVersion(folder: MigrationFolder): Promise<unknown> {
56 1 : return (await readSessionField(folder, 'formatVersion')) ?? null;
57 1 : }
58 :
59 : /// Read a top-level field of the session file, undefined when the file is unreadable
60 1 : async function readSessionField(folder: MigrationFolder, field: string): Promise<unknown> {
61 1 : try {
62 1 : const text = await folder.readText(STORAGE_SESSION_FILE);
63 1 : return text == null ? undefined : (JSON.parse(text) as Record<string, unknown>)?.[field];
64 1 : } catch {
65 0 : return undefined;
66 0 : }
67 1 : }
68 :
69 : /// Is a path of a bundle the path of a catalog file of a connection?
70 1 : export function isBundleCatalogPath(path: string): boolean {
71 1 : const parts = path.split('/');
72 1 : return parts.length === 3 && parts[0] === STORAGE_CONNECTIONS_FOLDER
73 1 : && (parts[2] === STORAGE_CATALOG_RELATIONS || parts[2] === STORAGE_CATALOG_FUNCTIONS);
74 1 : }
75 :
76 : /// The session file and the catalogs of a session bundle
77 : export interface SessionBundleHead extends Pick<SessionBundle, 'session' | 'catalogs'> {
78 : /// False when unusable entries were removed from the file lists of the session file
79 : filesValid: boolean;
80 : }
81 :
82 : /// Complete a bundle from its top-level files and the files of its folders.
83 : ///
84 : /// The file lists of the session file are reconciled with the files: scripts without a file are
85 : /// dropped and script files without an entry are appended on the connection of the script before them.
86 1 : export function assembleSessionBundle(
87 1 : head: Pick<SessionBundle, 'session' | 'catalogs'>,
88 1 : scripts: ScriptData[],
89 1 : data: DataFileData[] = [],
90 1 : ): SessionBundle {
91 1 : const reconciled = reconcileSessionScripts(head.session.scripts, scripts.map(script => script.name), firstConnectionId(head.session));
92 1 : return {
93 1 : session: {
94 1 : ...head.session,
95 1 : scripts: reconciled.scripts,
96 1 : attachments: sessionAttachments(data.map(file => file.name)),
97 1 : },
98 1 : catalogs: head.catalogs,
99 1 : scripts,
100 1 : data,
101 1 : };
102 1 : }
103 :
104 : /// Reconcile the file lists of a stored session file with the stored files, writes the session file
105 : /// only when the lists changed. Must not race with writes of the session file.
106 1 : export async function reconcileStoredSessionFiles(backend: StorageBackend, sessionId: string): Promise<SessionData> {
107 1 : const [session, scripts, dataFiles] = await Promise.all([
108 1 : backend.loadSession(sessionId),
109 1 : backend.loadScripts(sessionId),
110 1 : backend.listDataFiles(sessionId),
111 1 : ]);
112 1 : const sanitized = sanitizeSessionFiles(session);
113 1 : const reconciled = reconcileSessionScripts(sanitized.scripts, scripts.map(script => script.name), firstConnectionId(session));
114 1 : const attachments = sessionAttachments(dataFiles.map(file => file.name));
115 1 : const next: SessionData = { ...session, scripts: reconciled.scripts, attachments };
116 1 : if (!sanitized.valid || reconciled.changed || !sessionAttachmentsEqual(session.attachments, attachments)) {
117 1 : await backend.saveSessionManifest(sessionId, next);
118 : }
119 1 : return next;
120 1 : }
121 :
122 : /// The first connection of a session, script files without an entry run on it when no script precedes them
123 1 : function firstConnectionId(session: SessionData): string {
124 1 : return session.connections[0]?.connectionId ?? '';
125 1 : }
126 :
127 : /// Do two attachment lists name the same files in the same order?
128 1 : function sessionAttachmentsEqual(a: SessionData['attachments'] | undefined, b: SessionData['attachments']): boolean {
129 1 : return a != null && a.length === b.length && a.every((file, i) => file.name === b[i].name);
130 1 : }
131 :
132 : /// Migrate the top-level files of a session to the current format and read them with the catalogs
133 : /// of the connections in `connections/<id>/`.
134 : ///
135 : /// The folder is usually held in memory, e.g. the files of an import or an `OverlayFolder` over a
136 : /// folder that must stay untouched. Errors start with `errorPrefix`, e.g. "Invalid ZIP".
137 1 : export async function readSessionBundleHead(folder: MigrationFolder, errorPrefix: string): Promise<SessionBundleHead> {
138 : // Older sessions copied the catalog of their default connection into the top level of the
139 : // folder. It is read before the migration removes the copy and the default connection.
140 1 : const migration = await migrateSessionFolder(folder, SESSION_CATALOG_COPY_FORMAT_VERSION);
141 1 : const hasCatalogCopy = migration !== SessionMigrationResult.Missing
142 1 : && await readFormatVersion(folder) === SESSION_CATALOG_COPY_FORMAT_VERSION;
143 1 : const [legacyRelations, legacyFunctions, legacyConnectionId] = !hasCatalogCopy ? [null, null, null] : await Promise.all([
144 1 : folder.readText(STORAGE_CATALOG_RELATIONS),
145 1 : folder.readText(STORAGE_CATALOG_FUNCTIONS),
146 1 : readSessionField(folder, 'defaultConnectionId'),
147 1 : ]);
148 1 : await migrateSessionFolder(folder);
149 1 : if (migration === SessionMigrationResult.Missing) {
150 1 : throw new Error(`${errorPrefix}: missing ${STORAGE_SESSION_FILE}`);
151 : }
152 1 : const text = await folder.readText(STORAGE_SESSION_FILE);
153 1 : if (text == null) {
154 : // Metadata of an unsupported version under an older name
155 0 : throw new Error(`${errorPrefix}: invalid ${STORAGE_SESSION_FILE}: ${describeSessionValidationError(SessionValidationError.UnsupportedFormatVersion)}`);
156 : }
157 1 : let parsed: unknown;
158 1 : try {
159 1 : parsed = JSON.parse(text);
160 1 : } catch {
161 0 : throw new Error(`${errorPrefix}: invalid ${STORAGE_SESSION_FILE}`);
162 0 : }
163 1 : if (parsed == null || typeof parsed !== 'object' || Array.isArray(parsed)) {
164 0 : throw new Error(`${errorPrefix}: invalid ${STORAGE_SESSION_FILE}`);
165 : }
166 1 : const parsedSession = parsed as SessionData;
167 1 : const validation = validateSessionData(parsedSession);
168 1 : if (!validation.ok) {
169 1 : throw new Error(`${errorPrefix}: invalid ${STORAGE_SESSION_FILE}: ${describeSessionValidationError(validation.error)}`);
170 : }
171 1 : const files = sanitizeSessionFiles(parsedSession);
172 1 : const session: SessionData = { ...parsedSession, scripts: files.scripts, attachments: files.attachments };
173 1 : const catalogs: SessionBundleCatalog[] = [];
174 1 : for (const { connectionId } of session.connections) {
175 1 : const [relations, functions] = await Promise.all([
176 1 : folder.readText(bundleCatalogPath(connectionId, STORAGE_CATALOG_RELATIONS)),
177 1 : folder.readText(bundleCatalogPath(connectionId, STORAGE_CATALOG_FUNCTIONS)),
178 1 : ]);
179 1 : if (relations != null || functions != null) {
180 1 : catalogs.push({ connectionId, relations, functions });
181 1 : } else if (connectionId === (typeof legacyConnectionId === 'string' ? legacyConnectionId : session.connections[0].connectionId)
182 1 : && (legacyRelations != null || legacyFunctions != null)) {
183 1 : catalogs.push({ connectionId, relations: legacyRelations, functions: legacyFunctions });
184 : }
185 1 : }
186 1 : return { session, catalogs, filesValid: files.valid };
187 1 : }
188 :
189 : export interface SessionCatalogValidationResult {
190 : /// The bundle without the malformed catalog files
191 : bundle: SessionBundle;
192 : /// The paths of the malformed catalog files in the bundle
193 : invalidFiles: string[];
194 : }
195 :
196 : /// Give the connections of a bundle fresh ids, the scripts and the catalogs follow them
197 1 : export function regenerateBundleConnectionIds<B extends Pick<SessionBundle, 'session' | 'catalogs'>>(bundle: B): B {
198 1 : const { session } = bundle;
199 1 : const connectionIds = new Map(session.connections.map(connection => [connection.connectionId, crypto.randomUUID()]));
200 1 : return {
201 1 : ...bundle,
202 1 : session: {
203 1 : ...session,
204 1 : connections: session.connections.map(connection => ({ ...connection, connectionId: connectionIds.get(connection.connectionId)! })),
205 1 : scripts: remapSessionScripts(session.scripts, connectionIds),
206 1 : },
207 1 : catalogs: renameBundleCatalogs(bundle.catalogs, connectionIds),
208 1 : };
209 1 : }
210 :
211 : /// Keep only the connections of a bundle that its scripts use, in the order the scripts first use them,
212 : /// and their catalogs.
213 : /// A session without scripts keeps the connection its first script will run on.
214 1 : export function pruneBundleConnections<B extends Pick<SessionBundle, 'session' | 'catalogs'>>(bundle: B): B {
215 1 : const { session } = bundle;
216 1 : const used = sessionConnectionIdsOf(session.scripts, session.connections[0]?.connectionId ?? null);
217 1 : const copies = new Map(session.connections.map(connection => [connection.connectionId, connection]));
218 1 : return {
219 1 : ...bundle,
220 1 : session: { ...session, connections: used.flatMap(connectionId => copies.get(connectionId) ?? []) },
221 1 : catalogs: bundle.catalogs.filter(catalog => used.includes(catalog.connectionId)),
222 1 : };
223 1 : }
224 :
225 : /// Rename the connections of the catalogs of a bundle (connectionId -> new id).
226 : /// Connections that become one keep the first catalog.
227 1 : export function renameBundleCatalogs(catalogs: readonly SessionBundleCatalog[], connectionIds: ReadonlyMap<string, string>): SessionBundleCatalog[] {
228 1 : const renamed = new Map<string, SessionBundleCatalog>();
229 1 : for (const catalog of catalogs) {
230 1 : const connectionId = connectionIds.get(catalog.connectionId) ?? catalog.connectionId;
231 1 : if (!renamed.has(connectionId)) renamed.set(connectionId, { ...catalog, connectionId });
232 1 : }
233 1 : return [...renamed.values()];
234 1 : }
235 :
236 : /// Parse catalog files on scratch scripts and omit malformed SQL before it can reach storage.
237 : /// Analysis is intentionally excluded: imported catalogs may reference types or objects that are
238 : /// only available once their connection has been restored.
239 1 : export function filterInvalidSessionCatalogs(
240 1 : core: DashQL,
241 1 : bundle: SessionBundle,
242 1 : ): SessionCatalogValidationResult {
243 1 : const invalidFiles: string[] = [];
244 1 : let changed = false;
245 1 : const catalogs = bundle.catalogs.map(catalog => {
246 1 : const validate = (sql: string | null, fileName: string) => {
247 1 : if (sql == null || parsesWithoutErrors(core, sql)) return sql;
248 1 : invalidFiles.push(bundleCatalogPath(catalog.connectionId, fileName));
249 1 : changed = true;
250 1 : return null;
251 1 : };
252 1 : return {
253 1 : ...catalog,
254 1 : relations: validate(catalog.relations, STORAGE_CATALOG_RELATIONS),
255 1 : functions: validate(catalog.functions, STORAGE_CATALOG_FUNCTIONS),
256 1 : };
257 1 : });
258 1 : return { bundle: changed ? { ...bundle, catalogs } : bundle, invalidFiles };
259 1 : }
260 :
261 1 : function parsesWithoutErrors(core: DashQL, sql: string): boolean {
262 1 : const catalog = core.createCatalog();
263 1 : const script = core.createScript(catalog);
264 1 : let parsed: FlatBufferPtr<buffers.parser.ParsedScript> | null = null;
265 1 : try {
266 1 : script.replaceText(sql);
267 1 : script.parse();
268 1 : parsed = script.getParsed();
269 1 : const reader = parsed.read();
270 1 : return reader.scannerErrorsLength() === 0 && reader.parserErrorsLength() === 0;
271 1 : } catch {
272 0 : return false;
273 1 : } finally {
274 1 : script.destroy();
275 1 : catalog.destroy();
276 1 : }
277 1 : }
278 :
279 : export interface SessionBundleWriteOptions {
280 : /// Destination UUID. When omitted, preserve the source session's authoritative UUID.
281 : targetSessionId?: string;
282 : /// Add " (copy)" to a nonblank source name. Intended for an explicit Create-new operation.
283 : suffixNameWithCopy?: boolean;
284 : /// The caller guarantees the destination did not exist before this write. Only then may a
285 : /// failed whole-bundle write remove the destination as rollback.
286 : targetIsFresh?: boolean;
287 : }
288 :
289 : /// Read the portable durable session data from a backend. The catalogs of the connections of the
290 : /// session are only read for exports that ask for them, and only from a backend that holds the connections.
291 1 : export async function readSessionBundle(
292 1 : sessionId: string,
293 1 : backend: StorageBackend,
294 1 : withCatalog: boolean = false,
295 1 : ): Promise<SessionBundle> {
296 1 : const [stored, scripts] = await Promise.all([
297 1 : backend.loadSession(sessionId),
298 1 : backend.loadScripts(sessionId),
299 1 : ]);
300 1 : const registry = withCatalog && isSessionRegistryBackend(backend) ? backend : null;
301 1 : const catalogs: SessionBundleCatalog[] = [];
302 1 : for (const { connectionId } of registry == null ? [] : stored.connections) {
303 1 : const [relations, functions] = await Promise.all([
304 1 : registry!.loadConnectionFile(connectionId, STORAGE_CATALOG_RELATIONS),
305 1 : registry!.loadConnectionFile(connectionId, STORAGE_CATALOG_FUNCTIONS),
306 1 : ]);
307 1 : if (relations != null || functions != null) catalogs.push({ connectionId, relations, functions });
308 1 : }
309 1 : const data = await Promise.all((await backend.listDataFiles(sessionId))
310 1 : .map(async file => ({ name: file.name, bytes: await backend.loadDataFile(sessionId, file.name) })));
311 1 : const files = sanitizeSessionFiles(stored);
312 1 : const session: SessionData = { ...stored, scripts: files.scripts, attachments: files.attachments };
313 1 : return assembleSessionBundle({ session, catalogs }, scripts, data);
314 1 : }
315 :
316 : /// Write a fully parsed bundle to storage. This is deliberately not a replacement transaction:
317 : /// callers replacing an existing session must stage and coordinate that operation separately.
318 1 : export async function writeSessionBundle(
319 1 : bundle: SessionBundle,
320 1 : backend: StorageBackend,
321 1 : options: SessionBundleWriteOptions = {},
322 1 : ): Promise<string> {
323 1 : const validation = validateSessionData(bundle.session);
324 1 : if (!validation.ok) {
325 0 : throw new Error(`Invalid source session: ${describeSessionValidationError(validation.error)}`);
326 : }
327 1 : const sessionId = options.targetSessionId ?? bundle.session.sessionId;
328 1 : if (!isValidUuid(sessionId)) {
329 0 : throw new Error(`Invalid target session id: ${sessionId}`);
330 : }
331 1 : if (options.targetIsFresh && options.targetSessionId == null) {
332 0 : throw new Error('targetIsFresh requires an explicit targetSessionId');
333 : }
334 1 : for (const file of bundle.data ?? []) validateDataFileName(file.name);
335 :
336 1 : const session: SessionData = {
337 1 : ...bundle.session,
338 1 : sessionId,
339 1 : scripts: reconcileSessionScripts(bundle.session.scripts ?? [], bundle.scripts.map(script => script.name),
340 1 : firstConnectionId(bundle.session)).scripts,
341 1 : attachments: sessionAttachments((bundle.data ?? []).map(file => file.name)),
342 1 : };
343 1 : if (backend.getBackendType() === StorageBackendType.OPFS) {
344 1 : delete session.sessionPath;
345 1 : delete session.storageType;
346 1 : delete session.nativePath;
347 1 : }
348 1 : if (options.suffixNameWithCopy) {
349 0 : const name = session.name?.trim();
350 0 : if (name) {
351 0 : session.name = `${name} (copy)`;
352 : }
353 0 : }
354 :
355 1 : try {
356 1 : await backend.saveSessionManifest(sessionId, session);
357 1 : for (const script of bundle.scripts) {
358 1 : await backend.saveScript(sessionId, script.name, script.sql);
359 : }
360 1 : for (const file of bundle.data ?? []) {
361 1 : await backend.saveDataFile(sessionId, file.name, file.bytes);
362 : }
363 1 : } catch (error) {
364 1 : if (options.targetIsFresh) {
365 1 : try {
366 1 : await backend.deleteSession(sessionId);
367 1 : } catch {
368 : // Preserve the write error when best-effort rollback also fails.
369 0 : }
370 : }
371 1 : throw error;
372 1 : }
373 :
374 1 : return sessionId;
375 1 : }
376 :
377 : /// Seed the catalogs of the connections from a bundle that was written as session `sessionId`.
378 : ///
379 : /// A connection that has a catalog keeps it, the catalog of a bundle only describes a connection that
380 : /// is new to the app. Backends without connections are left alone.
381 1 : export async function seedConnectionCatalogs(
382 1 : backend: StorageBackend,
383 1 : bundle: Pick<SessionBundle, 'session' | 'catalogs'>,
384 1 : sessionId: string,
385 1 : ): Promise<void> {
386 1 : if (!isSessionRegistryBackend(backend)) return;
387 1 : for (const catalog of bundle.catalogs) {
388 1 : const files = [
389 1 : [STORAGE_CATALOG_RELATIONS, catalog.relations],
390 1 : [STORAGE_CATALOG_FUNCTIONS, catalog.functions],
391 1 : ] as const;
392 1 : const existing = await Promise.all(files.map(([fileName]) => backend.loadConnectionFile(catalog.connectionId, fileName)));
393 1 : if (existing.some(text => text != null)) continue;
394 1 : for (const [fileName, text] of files) {
395 1 : if (text == null) continue;
396 1 : await backend.saveConnectionFile(catalog.connectionId, fileName, text);
397 1 : }
398 1 : }
399 1 : }
400 :
401 : /// Compare all portable durable content. Used to verify a staged replacement before touching the
402 : /// live session UUID.
403 1 : export function sessionBundlesEqual(a: SessionBundle, b: SessionBundle): boolean {
404 1 : const { data: aData = [], ...aText } = a;
405 1 : const { data: bData = [], ...bText } = b;
406 1 : return JSON.stringify(aText) === JSON.stringify(bText)
407 1 : && aData.length === bData.length
408 1 : && aData.every((file, i) => file.name === bData[i].name
409 1 : && file.bytes.byteLength === bData[i].bytes.byteLength
410 1 : && file.bytes.every((byte, j) => byte === bData[i].bytes[j]));
411 1 : }
|
