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