Line data    Source code

       1              : import { getOPFSRoot } from '../../platform/opfs_root.js';
       2              : import type { ParquetDatabase } from '../../platform/hyperdb/hyperdb_parquet_export.js';
       3              : import { STORAGE_ATTACHMENTS_FOLDER, STORAGE_SESSIONS_FOLDER } from '../../storage/storage_backend.js';
       4              : import { attachmentPath, attachmentPrefix } from './session_attachments.js';
       5              : 
       6              : /// The mounts of the attachments of the sessions in the embedded Hyper.
       7              : ///
       8              : /// The embedded Hyper reads the attachments of a session from the OPFS folder `sessions/<id>/attachments` (a native
       9              : /// session mirrors its attachments there), mounted read-only at `/mnt/external/attachments/<id>`. A session is
      10              : /// mounted before a plan scans its attachments. A change of the attachments marks the mount stale, the next use
      11              : /// refreshes it, so the database looks up the files it opened again.
      12              : interface AttachmentMount {
      13              :     /// The database that mounted the folder
      14              :     database: ParquetDatabase;
      15              :     /// Did the attachments change since the mount or its last refresh?
      16              :     stale: boolean;
      17              : }
      18              : 
      19            1 : const MOUNTS = new Map<string, AttachmentMount>();
      20              : /// Mounts and refreshes run one after the other
      21            1 : let pending: Promise<unknown> = Promise.resolve();
      22              : 
      23            1 : function serialize<T>(f: () => Promise<T>): Promise<T> {
      24            1 :     const next = pending.then(f, f);
      25            1 :     pending = next.catch(() => {});
      26            1 :     return next;
      27            1 : }
      28              : 
      29              : /// Mount the attachments of a session in a database, refresh the mount if they changed. Returns the mount path.
      30            1 : export function mountSessionAttachments(
      31            1 :     database: ParquetDatabase,
      32            1 :     sessionId: string,
      33            1 :     getRoot: () => Promise<FileSystemDirectoryHandle> = getOPFSRoot,
      34            1 : ): Promise<string> {
      35            1 :     return serialize(async () => {
      36            1 :         const path = attachmentPrefix(sessionId);
      37            1 :         let mount = MOUNTS.get(sessionId);
      38            1 :         if (mount != null && mount.database !== database) {
      39            0 :             await mount.database.removeExternalDirectory(path).catch(() => {});
      40            0 :             MOUNTS.delete(sessionId);
      41            0 :             mount = undefined;
      42            0 :         }
      43            1 :         if (mount == null) {
      44            1 :             const sessions = await (await getRoot()).getDirectoryHandle(STORAGE_SESSIONS_FOLDER, { create: true });
      45            1 :             const session = await sessions.getDirectoryHandle(sessionId, { create: true });
      46            1 :             const folder = await session.getDirectoryHandle(STORAGE_ATTACHMENTS_FOLDER, { create: true });
      47            1 :             await database.registerExternalDirectory(folder, path, 'read');
      48            1 :             MOUNTS.set(sessionId, { database, stale: false });
      49            1 :         } else if (mount.stale) {
      50              :             // Changes during the refresh mark the mount stale again
      51            1 :             mount.stale = false;
      52            1 :             try {
      53            1 :                 await database.refreshExternalDirectory(path);
      54            1 :             } catch (error) {
      55            0 :                 mount.stale = true;
      56            0 :                 throw error;
      57            0 :             }
      58            1 :         }
      59            1 :         return path;
      60            1 :     });
      61            1 : }
      62              : 
      63              : /// The attachments of a session changed, the next use of the mount refreshes it
      64            1 : export function markSessionAttachmentsChanged(sessionId: string): void {
      65            1 :     const mount = MOUNTS.get(sessionId);
      66            1 :     if (mount != null) mount.stale = true;
      67            1 : }
      68              : 
      69              : /// Remove the mount of the attachments of a session
      70            1 : export function unmountSessionAttachments(sessionId: string): Promise<void> {
      71            1 :     return serialize(async () => {
      72            1 :         const mount = MOUNTS.get(sessionId);
      73            1 :         if (mount == null) return;
      74            1 :         MOUNTS.delete(sessionId);
      75            1 :         await mount.database.removeExternalDirectory(attachmentPrefix(sessionId)).catch(() => {});
      76            1 :     });
      77            1 : }
      78              : 
      79              : /// Read the columns of a Parquet attachment with the embedded Hyper
      80            1 : export async function readAttachmentColumns(database: ParquetDatabase, sessionId: string, fileName: string): Promise<string[]> {
      81            1 :     await mountSessionAttachments(database, sessionId);
      82            1 :     const connection = await database.connect({ pooled: true });
      83            1 :     try {
      84            1 :         const path = attachmentPath(sessionId, fileName).replace(/'/g, "''");
      85            1 :         const table = await connection.query(`select * from external('${path}', format => 'parquet') limit 0`);
      86            1 :         return table.schema.fields.map(field => field.name);
      87            1 :     } finally {
      88            1 :         await connection.close();
      89            1 :     }
      90            1 : }