Line data Source code
1 : import * as Immutable from 'immutable';
2 :
3 : import { Logger, stringifyError } from '../platform/logger/logger.js';
4 : import { VariantKind } from '../utils/variant.js';
5 : import type { StorageBackend, SessionData, SessionMetadata as StorageSessionMetadata, SessionConnection, ConnectionEntry, ConnectionParams, SessionRegistryBackend, SessionScript } from './storage_backend.js';
6 : import { ExternalScriptChangeError, STORAGE_SCRIPTS_FOLDER, STORAGE_SESSION_FILE, STORAGE_CONNECTIONS_FOLDER, isSessionRegistryBackend } from './storage_backend.js';
7 : import { SESSION_FORMAT_VERSION } from './migration/versions.js';
8 : import { sessionAttachments, withSessionScripts } from './session_files.js';
9 :
10 1 : const LOG_CTX = 'storage_writer';
11 :
12 : /// Order-independent deep equality for plain JSON values (objects, arrays, primitives).
13 : /// Session manifests are plain JSON (no Dates/Maps/functions), so this is sufficient.
14 1 : function jsonDeepEqual(a: any, b: any): boolean {
15 1 : if (a === b) {
16 1 : return true;
17 : }
18 1 : if (typeof a !== typeof b || a === null || b === null || typeof a !== 'object') {
19 1 : return false;
20 : }
21 1 : if (Array.isArray(a) || Array.isArray(b)) {
22 1 : if (!Array.isArray(a) || !Array.isArray(b) || a.length !== b.length) {
23 1 : return false;
24 : }
25 1 : return a.every((v, i) => jsonDeepEqual(v, b[i]));
26 1 : }
27 1 : const aKeys = Object.keys(a);
28 1 : const bKeys = Object.keys(b);
29 1 : if (aKeys.length !== bKeys.length) {
30 1 : return false;
31 : }
32 1 : return aKeys.every(k => Object.prototype.hasOwnProperty.call(b, k) && jsonDeepEqual(a[k], b[k]));
33 1 : }
34 :
35 : /// Compare the manifest fields the writer owns (formatVersion, sessionId, name, connections, scripts, attachments, metadata).
36 : /// Other fields (storageType/nativePath/sessionPath) are display- or registry-only and are not
37 : /// written into the session file, so they're deliberately ignored.
38 : ///
39 : /// Both sides are round-tripped through JSON first so this compares exactly what would be
40 : /// persisted: serialization drops `undefined`-valued keys, so a freshly built params object with
41 : /// `{ foo: undefined }` must compare equal to the reloaded `{}` it serializes to.
42 1 : function sessionManifestEquals(a: SessionData, b: SessionData): boolean {
43 1 : const project = (s: SessionData) => JSON.parse(JSON.stringify({
44 1 : formatVersion: s.formatVersion,
45 1 : sessionId: s.sessionId,
46 1 : name: s.name,
47 1 : connections: s.connections,
48 1 : scripts: s.scripts,
49 1 : attachments: s.attachments,
50 1 : metadata: s.metadata,
51 1 : }));
52 1 : return jsonDeepEqual(project(a), project(b));
53 1 : }
54 :
55 : export const DEBOUNCE_DURATION_SESSION_WRITE = 100;
56 : export const DEBOUNCE_DURATION_SCRIPT_WRITE = 100;
57 :
58 1 : export const WRITE_SESSION_MANIFEST = Symbol('WRITE_SESSION_MANIFEST');
59 1 : export const WRITE_SESSION_PROPERTIES = Symbol('WRITE_SESSION_PROPERTIES');
60 :
61 : /// The session fields that are owned by the user, not derived from scripts or connections.
62 : /// They are written together since writes of the same manifest replace each other while debounced.
63 : export interface SessionProperties {
64 : /// The session name, null if the user never named the session
65 : name: string | null;
66 : /// Does the session stay ephemeral? A write without it promotes the session
67 : ephemeral?: boolean;
68 : }
69 : /// A session that is written as a whole, its scripts first and its manifest last
70 : export interface SessionReplacement {
71 : /// The session id
72 : sessionId: string;
73 : /// The session name, null keeps the name on disk
74 : name: string | null;
75 : /// The script files, their text is read when the write runs. The script list of the session file
76 : /// is written by `WRITE_SESSION_SCRIPTS` alone, it is the only writer of the list.
77 : scripts: Array<{ fileName: string; readText: () => string }>;
78 : /// A connection of a session that is not stored yet, the script list follows with the others
79 : connection: SessionConnection | null;
80 : /// The file name the session was created from, if any
81 : originalFileName: string | undefined;
82 : }
83 1 : export const REPLACE_SESSION_SCRIPTS = Symbol('REPLACE_SESSION_SCRIPTS');
84 1 : export const WRITE_CONNECTION_ENTRY = Symbol('WRITE_CONNECTION_ENTRY');
85 1 : export const DELETE_CONNECTION_ENTRY = Symbol('DELETE_CONNECTION_ENTRY');
86 1 : export const WRITE_CONNECTION_FILE = Symbol('WRITE_CONNECTION_FILE');
87 1 : export const PATCH_SESSION_CONNECTION = Symbol('PATCH_SESSION_CONNECTION');
88 1 : export const WRITE_SCRIPT = Symbol('WRITE_SCRIPT');
89 1 : export const DELETE_SCRIPT = Symbol('DELETE_SCRIPT');
90 1 : export const RENAME_SCRIPT = Symbol('RENAME_SCRIPT');
91 1 : export const WRITE_SESSION_SCRIPTS = Symbol('WRITE_SESSION_SCRIPTS');
92 1 : export const REFRESH_SESSION_ATTACHMENTS = Symbol('REFRESH_SESSION_ATTACHMENTS');
93 :
94 : /// The script list of a session
95 : export interface SessionScriptList {
96 : /// The scripts in feed order with their connections
97 : scripts: SessionScript[];
98 : /// Copies of the connections of the scripts that the session file may not hold yet
99 : connections: SessionConnection[];
100 : }
101 :
102 : export type StorageWriteTaskVariant =
103 : | VariantKind<typeof WRITE_SESSION_MANIFEST, [string, SessionConnection[]] | [string, SessionConnection[], { ephemeral: boolean }]>
104 : | VariantKind<typeof WRITE_SESSION_PROPERTIES, [string, SessionProperties]>
105 : | VariantKind<typeof REPLACE_SESSION_SCRIPTS, SessionReplacement>
106 : | VariantKind<typeof WRITE_CONNECTION_ENTRY, ConnectionEntry>
107 : | VariantKind<typeof DELETE_CONNECTION_ENTRY, string>
108 : /// The text of a file, or how to read it when the write runs, null if there is nothing to write anymore
109 : | VariantKind<typeof WRITE_CONNECTION_FILE, [string, string, string | (() => string | null)]>
110 : | VariantKind<typeof PATCH_SESSION_CONNECTION, [string, SessionConnection] | [string, SessionConnection, string[]]>
111 : | VariantKind<typeof WRITE_SCRIPT, [string, string, string]>
112 : | VariantKind<typeof DELETE_SCRIPT, [string, string]>
113 : | VariantKind<typeof RENAME_SCRIPT, [string, string, string]>
114 : | VariantKind<typeof WRITE_SESSION_SCRIPTS, [string, SessionScriptList]>
115 : | VariantKind<typeof REFRESH_SESSION_ATTACHMENTS, string>
116 : ;
117 :
118 : /// The key of a write. Writes with the same key coalesce, the latest task wins.
119 : ///
120 : /// Keys are logical paths, they are not the paths of the files:
121 : /// - Session keys start with the session UUID, e.g. `<sessionId>/scripts/01_query.sql`. The
122 : /// per-session operations (pause, pending writes, cancellation) select keys by this prefix.
123 : /// - Keys of the app storage start with `@`, e.g. `@manifest/connections/<id>` for a connection
124 : /// entry of the app manifest and `@connections/<id>/<file>` for the files of a connection in
125 : /// `connections/<id>/`. A UUID never starts with `@`, so these keys never match the prefix of a
126 : /// session and are never paused or cancelled with one. `storageWriteChainKey` orders them.
127 : export type StorageWriteKey = string;
128 : /// The manifest is keyed on its real file path (`<sessionId>/dashql-session.json`) so the key it is
129 : /// *scheduled* under matches the path its *completed* write is recorded under — otherwise the manifest
130 : /// shows up as two rows in the stats view (a phantom `<sessionId>/` schedule row plus the real
131 : /// file's write row). Keying both on the file also means a scheduled-but-not-yet-flushed manifest
132 : /// write coalesces onto the same statistics row as its completion.
133 1 : export const groupSessionManifestWrites = (sessionId: string) => `${sessionId}/${STORAGE_SESSION_FILE}`;
134 1 : export const groupSessionWrites = (sessionId: string) => `${sessionId}/${STORAGE_SCRIPTS_FOLDER}`;
135 : /// The connection entries are written into the app manifest, one key per connection.
136 1 : export const groupConnectionEntryWrites = (connectionId: string) => `@manifest/${STORAGE_CONNECTIONS_FOLDER}/${connectionId}`;
137 : /// The files of a connection, e.g. its catalog relations
138 1 : export const groupConnectionFileWrites = (connectionId: string, fileName: string) => `@${STORAGE_CONNECTIONS_FOLDER}/${connectionId}/${fileName}`;
139 : /// The copy of a connection's params in a session. Copies of different connections must not
140 : /// coalesce, but they serialize with every other write of the session.
141 1 : export const groupSessionConnectionCopyWrites = (sessionId: string, connectionId: string) => `${sessionId}/${STORAGE_SESSION_FILE}:connection/${connectionId}`;
142 : /// The script list of a session and the connections it uses. It is patched into the session file and
143 : /// must not coalesce with the other writes of the session file.
144 1 : export const groupSessionScriptListWrites = (sessionId: string) => `${sessionId}/${STORAGE_SESSION_FILE}:scripts`;
145 : /// The attachment list of a session
146 1 : export const groupSessionAttachmentListWrites = (sessionId: string) => `${sessionId}/${STORAGE_SESSION_FILE}:attachments`;
147 : /// A cached query result of a session and the marker of its last access
148 1 : export const groupScriptWrites = (sessionId: string, fileName: string) =>
149 1 : `${sessionId}/${STORAGE_SCRIPTS_FOLDER}/${fileName}`;
150 1 : export const groupScriptDeletes = (sessionId: string, scriptName: string) =>
151 1 : `${sessionId}/${STORAGE_SCRIPTS_FOLDER}/${scriptName}:delete`;
152 : /// A page/script rename lives in its own `:rename` keyspace, keyed by the *source* path. Keeping it
153 : /// off the write/delete keyspaces means a later content write (or delete) of the destination never
154 : /// coalesces onto — and so never clobbers — a still-pending rename of the same name. The action lives
155 : /// in a `:rename` *suffix* rather than a prefix so the key still starts with the session id and
156 : /// sorts/scopes like every other file key.
157 1 : export const groupScriptRenames = (sessionId: string, oldScriptName: string) =>
158 1 : `${sessionId}/${STORAGE_SCRIPTS_FOLDER}/${oldScriptName}:rename`;
159 :
160 : /// Whether a statistics key belongs to a given session. Every write key is a path rooted at the
161 : /// session id (`<sessionId>/…`), with the `:delete`/`:rename` action namespaces living in a suffix
162 : /// (see the group* helpers above), so the session owns the key when it is the session id itself or a
163 : /// descendant of it. Used to scope the storage-writer stats view to the active session.
164 1 : export function storageWriteKeyBelongsToSession(key: StorageWriteKey, sessionId: string): boolean {
165 1 : return key === sessionId || key.startsWith(`${sessionId}/`);
166 1 : }
167 :
168 : /// Strip the session-id prefix from a write key for display, keeping any `:delete`/`:rename` suffix
169 : /// so the action stays legible. Once the stats view is scoped to a single session the session id is
170 : /// redundant on every row, so e.g. `<sessionId>/scripts/page-1/01.sql` renders as
171 : /// `scripts/page-1/01.sql` and `<sessionId>/scripts/page-1:rename` as `scripts/page-1:rename`.
172 : /// The bare session key collapses to an empty string. Keys that don't belong to the session are
173 : /// returned unchanged.
174 1 : export function storageWriteKeyWithinSession(key: StorageWriteKey, sessionId: string): string {
175 1 : if (!storageWriteKeyBelongsToSession(key, sessionId)) {
176 0 : return key;
177 : }
178 1 : return key === sessionId ? '' : key.slice(sessionId.length + 1);
179 1 : }
180 :
181 : /// The outcome of a scheduled write
182 : export type StorageWriteResult =
183 : /// The task was executed (a task may still skip writing content that is already persisted)
184 : | { status: 'written' }
185 : /// A newer task with the same key replaced the task before it was executed
186 : | { status: 'superseded' }
187 : /// The task was discarded before it was executed
188 : | { status: 'cancelled' }
189 : /// The writer or the session is paused, the task stays pending and is executed on resume
190 : | { status: 'deferred' }
191 : /// Executing the task failed
192 : | { status: 'failed'; error: unknown };
193 :
194 : /// The chain key of a write key, tasks with the same chain key are executed one after another.
195 : /// Session keys chain per session (`<sessionId>/…`), so mutations and their index rebuilds cannot
196 : /// overlap. Global keys start with `@` and chain per entity
197 : /// (`@manifest/…`, `@connections/<id>/…`).
198 1 : export function storageWriteChainKey(key: StorageWriteKey): string {
199 1 : const parts = key.split('/');
200 1 : if (key.startsWith('@connections/') && parts.length > 1) {
201 1 : return `${parts[0]}/${parts[1]}`;
202 : }
203 1 : return parts[0];
204 1 : }
205 :
206 : interface AsyncStorageWriteTask {
207 : /// The latest task
208 : latestTask: StorageWriteTaskVariant;
209 : /// Resolve the caller of the latest task
210 : resolveLatestTask: (result: StorageWriteResult) => void;
211 : /// The time when this task was first scheduled
212 : scheduledAt: Date;
213 : /// The debounce duration in milliseconds
214 : debounceDurationMs: number;
215 : /// The timer
216 : timer: ReturnType<typeof setTimeout>;
217 : }
218 :
219 : export interface StorageWriterStatistics {
220 : /// The number of writes we scheduled
221 : totalScheduledWrites: number;
222 : /// The number of writes we performed
223 : totalWrites: number;
224 : /// The number of bytes written in total to storage
225 : totalWrittenBytes: number;
226 : /// The accumulated time spent writing
227 : totalWriteTime: number;
228 : /// The last write
229 : lastWrite: Date | null;
230 : }
231 :
232 : export type StorageWriteStatisticsMap = Immutable.Map<StorageWriteKey, StorageWriterStatistics>;
233 : export type StorageWriteStatisticsSubscriber = (stats: StorageWriteStatisticsMap) => void;
234 :
235 : /// A session whose files are not available, e.g. a native folder that was moved or deleted
236 1 : export class SessionUnavailableError extends Error {
237 : /// The session id
238 1 : readonly sessionId: string;
239 :
240 1 : constructor(sessionId: string, cause: unknown) {
241 1 : super(`session ${sessionId} is not available: ${stringifyError(cause)}`);
242 1 : this.name = 'SessionUnavailableError';
243 1 : this.sessionId = sessionId;
244 1 : }
245 1 : }
246 :
247 : /// Subscriber for scripts that changed outside the app
248 : export type ExternalScriptChangeSubscriber = (sessionId: string) => void;
249 :
250 : /// Replace the copies of a connection's params in a session.
251 : /// References of the replaced connections (connections that were merged into this one) are rewritten
252 : /// to the connection id. Returns null when the session doesn't reference the connection or the copies
253 : /// are up to date.
254 1 : export function patchSessionConnection(session: SessionData, copy: SessionConnection, replaced: readonly string[] = []): SessionData | null {
255 1 : const connectionId = copy.connectionId;
256 1 : const references = (id: string) => id === connectionId || replaced.includes(id);
257 1 : const next: SessionData = {
258 1 : ...session,
259 1 : connections: dedupeConnections(session.connections.map(connection =>
260 1 : references(connection.connectionId) ? copy : connection)),
261 1 : scripts: session.scripts.map(script => replaced.includes(script.connectionId) ? { ...script, connectionId } : script),
262 1 : };
263 1 : return sessionManifestEquals(session, next) ? null : next;
264 1 : }
265 :
266 : /// Drop connections that an earlier connection of the list references already.
267 : /// Consolidated connections can turn distinct references into the same one.
268 1 : export function dedupeConnections<T extends { connectionId: string }>(connections: T[]): T[] {
269 1 : const seen = new Set<string>();
270 1 : return connections.filter(connection => {
271 1 : if (seen.has(connection.connectionId)) return false;
272 1 : seen.add(connection.connectionId);
273 1 : return true;
274 1 : });
275 1 : }
276 :
277 : /// Insert or replace a connection entry, keeping its creation time
278 1 : export function upsertConnectionEntry(connections: ConnectionEntry[] | undefined, entry: ConnectionEntry): ConnectionEntry[] | null {
279 1 : const entries = connections ?? [];
280 1 : const index = entries.findIndex(e => e.id === entry.id);
281 1 : const prev = index >= 0 ? entries[index] : null;
282 1 : const next: ConnectionEntry = {
283 1 : ...entry,
284 1 : ...(entry.createdAt == null && prev?.createdAt != null ? { createdAt: prev.createdAt } : {}),
285 1 : };
286 1 : if (prev != null && jsonDeepEqual(JSON.parse(JSON.stringify(prev)), JSON.parse(JSON.stringify(next)))) {
287 0 : return null;
288 : }
289 1 : return index >= 0 ? entries.map((e, i) => (i === index ? next : e)) : [...entries, next];
290 1 : }
291 :
292 1 : export class StorageWriter {
293 : /// The logger
294 1 : logger: Logger;
295 : /// The storage backend
296 1 : backend: StorageBackend;
297 : /// The pending tasks
298 1 : pendingTasks: Map<StorageWriteKey, AsyncStorageWriteTask>;
299 : /// The statistics
300 1 : statistics: StorageWriteStatisticsMap;
301 : /// The listeners for
302 1 : statisticsSubscribers: Set<StorageWriteStatisticsSubscriber>;
303 : /// Is the writer paused? While paused, debounced timers don't process tasks.
304 1 : paused: boolean;
305 : /// Sessions temporarily held by external reload/conflict handling.
306 1 : pausedSessions: Set<string>;
307 : /// Deleted sessions, writes for them are discarded until a session with the id is registered again.
308 1 : deletedSessions: Set<string>;
309 : /// The executions that are currently in flight (used by flush() to await completion).
310 1 : inFlight: Set<Promise<void>>;
311 : /// Latest execution per chain key, so mutations and their index rebuilds cannot overlap.
312 1 : chainExecutions: Map<string, Promise<void>>;
313 : /// Subscribers for scripts that changed outside the app
314 1 : externalChangeSubscribers: Set<ExternalScriptChangeSubscriber>;
315 : /// Monotonic scheduled-write generation per key, used to revalidate destructive prompts.
316 1 : writeKeyGenerations: Map<string, number>;
317 1 : nextWriteGeneration: number;
318 :
319 1 : constructor(logger: Logger, backend: StorageBackend) {
320 1 : this.logger = logger;
321 1 : this.backend = backend;
322 1 : this.pendingTasks = new Map();
323 1 : this.statistics = Immutable.Map();
324 1 : this.statisticsSubscribers = new Set();
325 1 : this.paused = false;
326 1 : this.pausedSessions = new Set();
327 1 : this.deletedSessions = new Set();
328 1 : this.inFlight = new Set();
329 1 : this.chainExecutions = new Map();
330 1 : this.externalChangeSubscribers = new Set();
331 1 : this.writeKeyGenerations = new Map();
332 1 : this.nextWriteGeneration = 1;
333 1 : }
334 :
335 : /// Pause the writer.
336 : /// Cancels all pending debounce timers so nothing is written until resume()/flush() is called.
337 : /// Tasks scheduled while paused stay pending and are re-armed on resume().
338 1 : public pause(): void {
339 1 : this.paused = true;
340 1 : for (const task of this.pendingTasks.values()) {
341 1 : clearTimeout(task.timer);
342 : }
343 1 : }
344 :
345 : /// Resume the writer, re-arming debounce timers for any tasks accumulated while paused.
346 1 : public resume(): void {
347 1 : if (!this.paused) {
348 0 : return;
349 : }
350 1 : this.paused = false;
351 1 : for (const [key, task] of this.pendingTasks) {
352 1 : if ([...this.pausedSessions].some(sessionId => storageWriteKeyBelongsToSession(key, sessionId))) {
353 0 : continue;
354 : }
355 1 : task.timer = setTimeout(() => this.processTask(key), task.debounceDurationMs);
356 1 : }
357 1 : }
358 :
359 : /// Flush the writer: process every pending task immediately and await all in-flight writes.
360 : /// Works regardless of the paused state and leaves the paused state unchanged.
361 1 : public async flush(): Promise<void> {
362 1 : const wasPaused = this.paused;
363 : // Temporarily allow processing even if paused.
364 1 : this.paused = false;
365 1 : const keys = [...this.pendingTasks.keys()];
366 1 : for (const key of keys) {
367 1 : const task = this.pendingTasks.get(key);
368 1 : if (task) {
369 1 : clearTimeout(task.timer);
370 : }
371 1 : await this.processTask(key, true);
372 1 : }
373 1 : this.paused = wasPaused;
374 : // Await any executions that were already in flight (e.g. a timer that fired just before).
375 1 : while (this.inFlight.size > 0) {
376 1 : await Promise.all([...this.inFlight]);
377 : }
378 1 : }
379 :
380 1 : public getStatistics(): StorageWriteStatisticsMap {
381 1 : return this.statistics;
382 1 : }
383 :
384 1 : public pauseSession(sessionId: string): void {
385 1 : this.pausedSessions.add(sessionId);
386 1 : for (const [key, task] of this.pendingTasks) {
387 1 : if (storageWriteKeyBelongsToSession(key, sessionId)) {
388 1 : clearTimeout(task.timer);
389 : }
390 : }
391 1 : }
392 :
393 1 : public resumeSession(sessionId: string): void {
394 1 : if (!this.pausedSessions.delete(sessionId) || this.paused) {
395 0 : return;
396 : }
397 1 : for (const [key, task] of this.pendingTasks) {
398 1 : if (storageWriteKeyBelongsToSession(key, sessionId)) {
399 1 : task.timer = setTimeout(() => this.processTask(key), task.debounceDurationMs);
400 : }
401 : }
402 1 : }
403 :
404 : /// Pending write keys below a session id. This is intentionally prefix-based: every writer key
405 : /// is rooted at the session id, including the delete/rename action namespaces.
406 1 : public getPendingKeysForSession(sessionId: string, include: (key: string) => boolean = () => true): string[] {
407 1 : return [...this.pendingTasks.keys()].filter(key => storageWriteKeyBelongsToSession(key, sessionId) && include(key));
408 1 : }
409 :
410 : /// Discard pending writes for a session, resolving their promises as not executed. Used only
411 : /// after the user explicitly chooses an externally-written disk version over local debounced
412 : /// edits. Writes already in flight cannot be cancelled and are handled by settle().
413 1 : public cancelPendingWritesForSession(sessionId: string, include: (key: string) => boolean = () => true): void {
414 1 : for (const [key, task] of this.pendingTasks) {
415 1 : if (!storageWriteKeyBelongsToSession(key, sessionId) || !include(key)) {
416 0 : continue;
417 : }
418 1 : clearTimeout(task.timer);
419 1 : this.pendingTasks.delete(key);
420 1 : task.resolveLatestTask({ status: 'cancelled' });
421 1 : }
422 1 : }
423 :
424 : /// Discard the pending writes of the files of a connection that is deleted, they would recreate its folder
425 1 : public cancelPendingConnectionFileWrites(connectionId: string): void {
426 1 : const prefix = `@${STORAGE_CONNECTIONS_FOLDER}/${connectionId}/`;
427 1 : for (const [key, task] of this.pendingTasks) {
428 1 : if (key.startsWith(prefix)) {
429 1 : clearTimeout(task.timer);
430 1 : this.pendingTasks.delete(key);
431 1 : task.resolveLatestTask({ status: 'cancelled' });
432 1 : }
433 : }
434 1 : }
435 :
436 : /// Delete a session from the storage.
437 : /// The pending writes of the session are discarded and writes that already started finish first,
438 : /// no write recreates files of the deleted session afterwards.
439 1 : public async deleteSession(sessionId: string): Promise<void> {
440 1 : this.deletedSessions.add(sessionId);
441 1 : this.cancelPendingWritesForSession(sessionId);
442 1 : await this.settle();
443 1 : try {
444 1 : await this.backend.deleteSession(sessionId);
445 1 : } catch (error) {
446 : // The session stays in the storage, it may be written again
447 0 : this.deletedSessions.delete(sessionId);
448 0 : throw error;
449 0 : }
450 1 : this.logger.info("Deleted session", { sessionId }, LOG_CTX);
451 1 : }
452 :
453 : /// A session with the id of a deleted session was registered again, e.g. by importing the same bundle.
454 : /// Writes for the id are accepted again, the writes discarded while it was deleted stay discarded.
455 1 : public reviveSession(sessionId: string): void {
456 1 : this.deletedSessions.delete(sessionId);
457 1 : }
458 :
459 : /// Is a session held, e.g. after one of its scripts changed outside the app?
460 : /// Its writes stay pending until the session resumes.
461 1 : public isSessionPaused(sessionId: string): boolean {
462 1 : return this.paused || this.pausedSessions.has(sessionId);
463 1 : }
464 :
465 : /// Write the pending tasks of a session right away and wait until they are persisted.
466 : ///
467 : /// Resolves true when every pending write of the session reached the storage. A held writer or
468 : /// session keeps its writes pending and resolves false. Unlike flush(), a held session is never
469 : /// forced, its writes would overwrite the external change. A failed write stays pending as well,
470 : /// unless a newer write of its key replaced it, and the next flush writes it again.
471 1 : public async flushSession(sessionId: string): Promise<boolean> {
472 1 : if (this.isSessionPaused(sessionId)) return false;
473 1 : const failed = new Map<StorageWriteKey, AsyncStorageWriteTask>();
474 1 : for (const key of this.getPendingKeysForSession(sessionId)) {
475 1 : const task = this.pendingTasks.get(key);
476 1 : if (task == null) continue;
477 1 : const resolve = task.resolveLatestTask;
478 1 : task.resolveLatestTask = result => {
479 1 : if (result.status === 'failed') failed.set(key, task);
480 1 : resolve(result);
481 1 : };
482 1 : clearTimeout(task.timer);
483 1 : void this.processTask(key);
484 1 : }
485 1 : await this.settle();
486 1 : for (const [key, task] of failed) {
487 1 : if (this.pendingTasks.has(key)) continue;
488 1 : task.resolveLatestTask = () => { };
489 1 : task.timer = setTimeout(() => this.processTask(key), task.debounceDurationMs);
490 1 : this.pendingTasks.set(key, task);
491 1 : }
492 : // A write that would overwrite an external change pauses the session and stays pending
493 1 : return failed.size === 0 && !this.isSessionPaused(sessionId);
494 1 : }
495 :
496 : /// Wait for writes that have already started without flushing pending debounced work.
497 1 : public async settle(): Promise<void> {
498 1 : while (this.inFlight.size > 0) {
499 1 : await Promise.all([...this.inFlight]);
500 : }
501 1 : }
502 :
503 1 : public getSessionWriteGeneration(sessionId: string, include: (key: string) => boolean = () => true): number {
504 1 : let generation = 0;
505 1 : for (const [key, value] of this.writeKeyGenerations) {
506 1 : if (storageWriteKeyBelongsToSession(key, sessionId) && include(key)) {
507 1 : generation = Math.max(generation, value);
508 : }
509 : }
510 1 : return generation;
511 1 : }
512 :
513 : /// Subscribe to script writes that are held because the script changed outside the app
514 1 : public subscribeExternalChanges(subscriber: ExternalScriptChangeSubscriber): () => void {
515 1 : this.externalChangeSubscribers.add(subscriber);
516 1 : return () => this.externalChangeSubscribers.delete(subscriber);
517 1 : }
518 :
519 1 : public subscribeStatisticsListener(listener: StorageWriteStatisticsSubscriber) {
520 1 : this.statisticsSubscribers.add(listener);
521 :
522 1 : }
523 1 : public unsubscribeStatisticsListener(listener: StorageWriteStatisticsSubscriber) {
524 1 : this.statisticsSubscribers.delete(listener);
525 1 : }
526 :
527 : /// Schedule a debounced write, resolves true when the task was executed.
528 : /// A newer write with the same key replaces a pending task and resolves its caller with false.
529 1 : public async write(key: string, task: StorageWriteTaskVariant, debounceFor: number = 0): Promise<boolean> {
530 1 : const result = await this.schedule(key, task, debounceFor);
531 1 : return result.status === 'written';
532 1 : }
533 :
534 : /// Write without debouncing and wait until the content is persisted.
535 : ///
536 : /// The task replaces any pending task with the same key and is executed right away, chained after
537 : /// running tasks of the same chain key. A paused writer or session resolves with `deferred` and
538 : /// keeps the task pending, it is executed on resume.
539 1 : public writeDurable(key: string, task: StorageWriteTaskVariant): Promise<StorageWriteResult> {
540 1 : const result = this.schedule(key, task, 0);
541 1 : const pending = this.pendingTasks.get(key)!;
542 1 : clearTimeout(pending.timer);
543 1 : if (this.isPausedKey(key)) {
544 1 : pending.resolveLatestTask({ status: 'deferred' });
545 1 : pending.resolveLatestTask = () => { };
546 1 : } else {
547 1 : void this.processTask(key);
548 : }
549 1 : return result;
550 1 : }
551 :
552 : /// Is a key held by a paused writer or a paused session?
553 1 : protected isPausedKey(key: string): boolean {
554 1 : return this.paused || [...this.pausedSessions].some(sessionId => storageWriteKeyBelongsToSession(key, sessionId));
555 1 : }
556 :
557 1 : protected schedule(key: string, task: StorageWriteTaskVariant, debounceFor: number): Promise<StorageWriteResult> {
558 : // Writes that were queued while a session was deleted must not recreate its files
559 1 : if ([...this.deletedSessions].some(sessionId => storageWriteKeyBelongsToSession(key, sessionId))) {
560 1 : return Promise.resolve({ status: 'cancelled' });
561 : }
562 1 : this.writeKeyGenerations.set(key, this.nextWriteGeneration++);
563 : // Is there a previous task with the same key?
564 1 : const prevTask = this.pendingTasks.get(key);
565 1 : let scheduledAt: Date;
566 1 : let debounceDurationMs: number = debounceFor;
567 1 : let timer: ReturnType<typeof setTimeout>;
568 1 : if (prevTask) {
569 : // Tell the former write call that the task was not executed
570 1 : prevTask.resolveLatestTask({ status: 'superseded' });
571 : // Overwrite the task
572 1 : scheduledAt = prevTask.scheduledAt;
573 1 : debounceDurationMs = prevTask.debounceDurationMs;
574 1 : timer = prevTask.timer;
575 1 : } else {
576 1 : scheduledAt = new Date();
577 1 : timer = setTimeout(() => this.processTask(key), debounceFor);
578 1 : }
579 1 : let resolveTask: ((result: StorageWriteResult) => void) | null = null;
580 1 : const result = new Promise<StorageWriteResult>(resolve => { resolveTask = resolve; });
581 :
582 : // Overwrite any previous task with the same key
583 1 : this.pendingTasks.set(key, {
584 1 : latestTask: task,
585 1 : resolveLatestTask: resolveTask!,
586 1 : scheduledAt: scheduledAt,
587 1 : debounceDurationMs,
588 1 : timer,
589 1 : });
590 1 : this.registerScheduledWrite(key);
591 1 : return result;
592 1 : }
593 :
594 1 : protected async processTask(key: string, force: boolean = false) {
595 1 : const task = this.pendingTasks.get(key);
596 1 : if (!task) {
597 1 : return;
598 : }
599 : // If a timer fires while paused, leave the task pending - it'll be re-armed on resume().
600 1 : if (!force && this.isPausedKey(key)) {
601 1 : return;
602 : }
603 1 : this.pendingTasks.delete(key);
604 1 : const chainKey = storageWriteChainKey(key);
605 1 : const previousExecution = this.chainExecutions.get(chainKey);
606 1 : const execution = (async () => {
607 1 : if (previousExecution) {
608 1 : await previousExecution;
609 : }
610 1 : try {
611 1 : await this.executeTask(key, task.latestTask);
612 1 : task.resolveLatestTask({ status: 'written' });
613 1 : } catch (e: any) {
614 1 : if (e instanceof ExternalScriptChangeError) {
615 1 : this.holdExternallyChangedWrite(key, task, e);
616 1 : return;
617 1 : }
618 1 : this.logger.error("executing write task failed", {
619 1 : key: key,
620 1 : error: stringifyError(e)
621 1 : })
622 1 : task.resolveLatestTask({ status: 'failed', error: e });
623 1 : }
624 1 : })();
625 1 : this.chainExecutions.set(chainKey, execution);
626 1 : this.inFlight.add(execution);
627 1 : try {
628 1 : await execution;
629 1 : } finally {
630 1 : this.inFlight.delete(execution);
631 1 : if (this.chainExecutions.get(chainKey) === execution) {
632 1 : this.chainExecutions.delete(chainKey);
633 : }
634 1 : }
635 1 : }
636 :
637 : /// Keep a write that would overwrite an external change pending and pause its session.
638 : /// The subscribers decide between the disk and the app, then resume the session.
639 1 : protected holdExternallyChangedWrite(key: string, task: AsyncStorageWriteTask, error: ExternalScriptChangeError) {
640 1 : this.logger.warn("script changed outside the app, holding the write", { key, sessionId: error.sessionId, file: error.fileName }, LOG_CTX);
641 1 : this.pauseSession(error.sessionId);
642 1 : if (this.pendingTasks.has(key)) {
643 : // A newer task with the same key was scheduled while this one was executing
644 0 : task.resolveLatestTask({ status: 'superseded' });
645 : } else {
646 1 : this.pendingTasks.set(key, task);
647 : }
648 1 : for (const subscriber of this.externalChangeSubscribers) {
649 1 : subscriber(error.sessionId);
650 : }
651 1 : }
652 :
653 1 : protected updateStatistics(statistics: Immutable.Map<StorageWriteKey, StorageWriterStatistics>) {
654 1 : this.statistics = statistics;
655 1 : for (const subscriber of this.statisticsSubscribers) {
656 1 : subscriber(statistics);
657 : }
658 1 : }
659 :
660 1 : protected registerScheduledWrite(key: StorageWriteKey) {
661 : const ifNotSet: StorageWriterStatistics = {
662 1 : totalScheduledWrites: 1,
663 1 : totalWrites: 0,
664 1 : totalWrittenBytes: 0,
665 1 : totalWriteTime: 0,
666 1 : lastWrite: null,
667 : };
668 1 : const stats = this.statistics.update(key, ifNotSet, (stats) => ({ ...stats, totalScheduledWrites: stats.totalScheduledWrites + 1 }));
669 1 : this.updateStatistics(stats);
670 1 : }
671 :
672 1 : protected registerWrite(key: StorageWriteKey, writtenBytes: number, writeDurationMs: number) {
673 1 : const ifNotSet: StorageWriterStatistics = {
674 1 : totalScheduledWrites: 1,
675 1 : totalWrites: 1,
676 1 : totalWrittenBytes: writtenBytes,
677 1 : totalWriteTime: writeDurationMs,
678 1 : lastWrite: new Date(),
679 1 : };
680 1 : const stats = this.statistics.update(key, ifNotSet, (stats) => ({
681 1 : ...stats,
682 1 : totalWrites: stats.totalWrites + 1,
683 1 : totalWrittenBytes: stats.totalWrittenBytes + writtenBytes,
684 1 : totalWriteTime: stats.totalWriteTime + writeDurationMs,
685 1 : lastWrite: new Date(),
686 1 : }));
687 1 : this.updateStatistics(stats);
688 1 : }
689 :
690 : /// The backend as session registry, connection writes need one
691 1 : protected requireRegistry(): SessionRegistryBackend {
692 1 : if (!isSessionRegistryBackend(this.backend)) {
693 1 : throw new Error('the storage backend does not own a session registry');
694 : }
695 1 : return this.backend;
696 1 : }
697 :
698 : /// Patch an existing session file, the write is skipped when nothing changed.
699 : /// A missing file is not recreated, a moved native folder would otherwise come back as a new one.
700 1 : private async patchSessionFile(key: string, sessionId: string, patch: (session: SessionData) => SessionData): Promise<void> {
701 1 : let existing: SessionData;
702 1 : try {
703 1 : existing = await this.backend.loadSession(sessionId);
704 1 : } catch (e: any) {
705 1 : throw new SessionUnavailableError(sessionId, e);
706 1 : }
707 1 : const next = patch(existing);
708 1 : if (sessionManifestEquals(existing, next)) return;
709 1 : this.logger.info("Patching session", { key, sessionId }, LOG_CTX);
710 1 : await this.saveSessionFile(sessionId, next);
711 1 : }
712 :
713 : /// Save the session file of a session and record the write
714 1 : private async saveSessionFile(sessionId: string, session: SessionData): Promise<void> {
715 1 : const text = JSON.stringify(session, null, 2);
716 1 : await this.timedWrite(`${sessionId}/${STORAGE_SESSION_FILE}`, text, () => this.backend.saveSessionManifest(sessionId, session));
717 1 : }
718 :
719 : /// Write a file and record the write with its duration
720 1 : private async timedWrite(path: string, text: string, write: () => Promise<void>): Promise<void> {
721 1 : const timeBefore = new Date();
722 1 : await write();
723 1 : this.registerWrite(path, text.length, new Date().getTime() - timeBefore.getTime());
724 1 : }
725 :
726 1 : protected async executeTask(key: string, task: StorageWriteTaskVariant) {
727 1 : switch (task.type) {
728 1 : case WRITE_SESSION_MANIFEST: {
729 1 : const [sessionId, connections, created] = task.value;
730 1 : this.logger.info("Writing session", {
731 1 : key,
732 1 : sessionId,
733 1 : }, LOG_CTX);
734 :
735 : // Load whatever is already on disk so we can (a) preserve the original createdAt
736 : // instead of churning a fresh timestamp on every write, and (b) skip the write
737 : // entirely when nothing actually changed. WRITE_SESSION_MANIFEST is scheduled on
738 : // every connection-state change, so most of these tasks rewrite identical content.
739 1 : let existingSession: SessionData | null = null;
740 1 : try {
741 1 : existingSession = await this.backend.loadSession(sessionId);
742 1 : } catch {
743 1 : existingSession = null;
744 1 : }
745 :
746 : // For now, create minimal session metadata
747 1 : const sessionMetadata: StorageSessionMetadata = {
748 1 : ...existingSession?.metadata,
749 1 : createdAt: existingSession?.metadata?.createdAt ?? new Date().toISOString(),
750 1 : ...(created?.ephemeral ? { ephemeral: true } : {}),
751 1 : };
752 :
753 : // sessionPath is a display-only field, recomputed from the UUID + location for
754 : // the UI; we don't persist it here. storageType/nativePath are stamped by the
755 : // composite backend, which knows the session's physical location.
756 : // The given connections replace the copies on disk, the scripts keep theirs
757 1 : const connData: SessionData = withSessionScripts({
758 1 : formatVersion: SESSION_FORMAT_VERSION,
759 1 : sessionId,
760 1 : ...(existingSession?.name ? { name: existingSession.name } : {}),
761 1 : connections: [],
762 1 : scripts: [],
763 1 : attachments: existingSession?.attachments ?? [],
764 1 : metadata: sessionMetadata,
765 1 : }, existingSession?.scripts ?? [], [...connections, ...existingSession?.connections ?? []]);
766 :
767 : // Skip the write if the persisted manifest already matches what we'd write.
768 1 : if (existingSession != null && sessionManifestEquals(existingSession, connData)) {
769 1 : this.logger.debug("Skipping session write: manifest unchanged", {
770 1 : sessionId,
771 1 : connectionIds: connections.map(connection => connection.connectionId).join(','),
772 1 : }, LOG_CTX);
773 1 : break;
774 1 : }
775 :
776 1 : await this.saveSessionFile(sessionId, connData);
777 1 : break;
778 1 : }
779 1 : case WRITE_SESSION_PROPERTIES: {
780 1 : const [sessionId, { name, ephemeral }] = task.value;
781 1 : await this.patchSessionFile(key, sessionId, existing => {
782 1 : const next: SessionData = { ...existing };
783 1 : if (name) next.name = name;
784 0 : else delete next.name;
785 1 : if (!ephemeral && next.metadata?.ephemeral != null) {
786 1 : const { ephemeral: _promoted, ...metadata } = next.metadata;
787 1 : next.metadata = metadata;
788 1 : }
789 1 : return next;
790 1 : });
791 1 : break;
792 1 : }
793 1 : case REPLACE_SESSION_SCRIPTS: {
794 1 : const session = task.value;
795 1 : const sessionId = session.sessionId;
796 1 : this.logger.info("Creating session", {
797 1 : key,
798 1 : sessionId,
799 1 : }, LOG_CTX);
800 :
801 1 : for (const script of session.scripts) {
802 1 : const sql = script.readText();
803 1 : await this.timedWrite(`${sessionId}/scripts/${script.fileName}`, sql, () => this.backend.saveScript(sessionId, script.fileName, sql));
804 1 : }
805 :
806 : // Save session last so it never references content that doesn't exist yet.
807 : // Preserve the original createdAt across rewrites; only stamp it on the first write.
808 1 : let connections: SessionData['connections'];
809 1 : let createdAt: string;
810 : // Preserve the user-supplied name across a session rewrite: this path rebuilds the
811 : // whole manifest from session scripts, which carry no name, so we must carry over
812 : // whatever the user already set on disk.
813 1 : let existingName: string | undefined;
814 1 : let existingMetadata: StorageSessionMetadata | undefined;
815 1 : let existingAttachments: SessionData['attachments'] = [];
816 1 : let existingScripts: SessionData['scripts'] = [];
817 1 : try {
818 1 : const existingSession = await this.backend.loadSession(sessionId);
819 1 : existingScripts = existingSession.scripts ?? [];
820 1 : connections = existingSession.connections;
821 1 : createdAt = existingSession.metadata?.createdAt ?? new Date().toISOString();
822 1 : existingName = existingSession.name;
823 1 : existingMetadata = existingSession.metadata;
824 1 : existingAttachments = existingSession.attachments ?? [];
825 1 : } catch {
826 1 : connections = session.connection == null ? [] : [session.connection];
827 1 : createdAt = new Date().toISOString();
828 1 : }
829 :
830 1 : const sessionMetadata: StorageSessionMetadata = {
831 1 : ...existingMetadata,
832 1 : originalFileName: session.originalFileName,
833 1 : createdAt,
834 1 : };
835 :
836 1 : const connData: SessionData = withSessionScripts({
837 1 : formatVersion: SESSION_FORMAT_VERSION,
838 1 : sessionId,
839 1 : ...(session.name ? { name: session.name } : existingName ? { name: existingName } : {}),
840 1 : connections,
841 1 : scripts: [],
842 1 : attachments: existingAttachments,
843 1 : metadata: sessionMetadata,
844 1 : }, existingScripts);
845 :
846 1 : await this.saveSessionFile(sessionId, connData);
847 1 : break;
848 1 : }
849 1 : case WRITE_SESSION_SCRIPTS: {
850 1 : const [sessionId, list] = task.value;
851 1 : await this.patchSessionFile(key, sessionId, session => withSessionScripts(session, list.scripts, list.connections));
852 1 : break;
853 1 : }
854 1 : case REFRESH_SESSION_ATTACHMENTS: {
855 1 : const sessionId = task.value;
856 1 : const attachments = sessionAttachments((await this.backend.listDataFiles(sessionId)).map(file => file.name));
857 1 : await this.patchSessionFile(key, sessionId, session => ({ ...session, attachments }));
858 1 : break;
859 1 : }
860 1 : case WRITE_CONNECTION_ENTRY: {
861 1 : const entry = task.value;
862 1 : const registry = this.requireRegistry();
863 1 : const timeBefore = new Date();
864 1 : await registry.updateConnections(connections => upsertConnectionEntry(connections, entry));
865 1 : this.registerWrite(key, JSON.stringify(entry).length, new Date().getTime() - timeBefore.getTime());
866 1 : break;
867 1 : }
868 1 : case DELETE_CONNECTION_ENTRY: {
869 1 : const connectionId = task.value;
870 1 : const registry = this.requireRegistry();
871 1 : this.logger.info("Deleting connection", { key, connectionId }, LOG_CTX);
872 1 : await registry.updateConnections(connections => {
873 1 : if (connections == null || !connections.some(e => e.id === connectionId)) return null;
874 0 : return connections.filter(e => e.id !== connectionId);
875 1 : });
876 1 : await registry.deleteConnectionFiles(connectionId);
877 1 : break;
878 1 : }
879 1 : case WRITE_CONNECTION_FILE: {
880 1 : const [connectionId, fileName, content] = task.value;
881 1 : const text = typeof content === 'function' ? content() : content;
882 1 : if (text == null) break;
883 1 : const registry = this.requireRegistry();
884 1 : await this.timedWrite(`@${STORAGE_CONNECTIONS_FOLDER}/${connectionId}/${fileName}`, text, () => registry.saveConnectionFile(connectionId, fileName, text));
885 1 : break;
886 1 : }
887 1 : case PATCH_SESSION_CONNECTION: {
888 1 : const [sessionId, copy, replaced] = task.value;
889 : // The copy is only patched into an existing session file. A missing file must not be
890 : // recreated, a moved native folder would otherwise silently come back as a new one.
891 1 : await this.patchSessionFile(key, sessionId, existing => patchSessionConnection(existing, copy, replaced) ?? existing);
892 1 : break;
893 1 : }
894 1 : case WRITE_SCRIPT: {
895 1 : const [sessionId, fileName, sql] = task.value;
896 :
897 1 : this.logger.info("Writing session script", {
898 1 : key,
899 1 : sessionId,
900 1 : file: fileName,
901 1 : }, LOG_CTX);
902 :
903 1 : await this.timedWrite(`${sessionId}/scripts/${fileName}`, sql, () => this.backend.saveScript(sessionId, fileName, sql));
904 1 : break;
905 1 : }
906 1 : case DELETE_SCRIPT: {
907 1 : const [sessionId, scriptName] = task.value;
908 1 : this.logger.info("Deleting session script", {
909 1 : task: key,
910 1 : sessionId,
911 1 : scriptName,
912 1 : }, LOG_CTX);
913 1 : await this.backend.deleteScript(sessionId, scriptName);
914 1 : break;
915 1 : }
916 1 : case RENAME_SCRIPT: {
917 1 : const [sessionId, oldScriptName, newScriptName] = task.value;
918 1 : this.logger.info("Renaming session script", {
919 1 : task: key,
920 1 : sessionId,
921 1 : oldScriptName,
922 1 : newScriptName,
923 1 : }, LOG_CTX);
924 1 : await this.backend.renameScript(sessionId, oldScriptName, newScriptName);
925 1 : break;
926 1 : }
927 1 : }
928 1 : }
929 1 : }
|
