16876454078

Committed 11 Aug 2025 09:43AM UTC coverage: 82.089% (+0.7%) from 81.362%

Build # 16876454078

Build Type

Pull #424

github

Committed by

web-flow

Commit Message

Merge 34aff9a17 into 7572ec5d6

Pull Request Pull Request #424: feat: adds an export tool and exported-data resource MCP-16

Run Details

780 of 999 branches covered (78.08%)

Branch coverage included in aggregate %.

592 of 659 new or added lines in 13 files covered. (89.83%)

12 existing lines in 2 files now uncovered.

4078 of 4919 relevant lines covered (82.9%)

66.28 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.14

/src/common/exportsManager.ts

import z from "zod";
import path from "path";
import fs from "fs/promises";
import EventEmitter from "events";
import { createWriteStream } from "fs";
import { FindCursor } from "mongodb";
import { EJSON, EJSONOptions, ObjectId } from "bson";
import { Transform } from "stream";
import { pipeline } from "stream/promises";
import { MongoLogId } from "mongodb-log-writer";
import { RWLock } from "async-rwlock";

import { UserConfig } from "./config.js";
import { LoggerBase, LogId } from "./logger.js";

export const jsonExportFormat = z.enum(["relaxed", "canonical"]);
export type JSONExportFormat = z.infer<typeof jsonExportFormat>;

interface CommonExportData {
    exportName: string;
    exportTitle: string;
    exportURI: string;
    exportPath: string;
}

interface ReadyExport extends CommonExportData {
    exportStatus: "ready";
    exportCreatedAt: number;
}

interface InProgressExport extends CommonExportData {
    exportStatus: "in-progress";
}

type StoredExport = ReadyExport | InProgressExport;

/**
 * Ideally just exportName and exportURI should be made publicly available but
 * we also make exportPath available because the export tool, also returns the
 * exportPath in its response when the MCP server is running connected to stdio
 * transport. The reasoning behind this is that a few clients, Cursor in
 * particular, as of the date of this writing (7 August 2025) cannot refer to
 * resource URIs which means they have no means to access the exported resource.
 * As of this writing, majority of the usage of our MCP server is behind STDIO
 * transport so we can assume that for most of the usages, if not all, the MCP
 * server will be running on the same machine as of the MCP client and thus we
 * can provide the local path to export so that these clients which do not still
 * support parsing resource URIs, can still work with the exported data. We
 * expect for clients to catch up and implement referencing resource URIs at
 * which point it would be safe to remove the `exportPath` from the publicly
 * exposed properties of an export.
 *
 * The editors that we would like to watch out for are Cursor and Windsurf as
 * they don't yet support working with Resource URIs.
 *
 * Ref Cursor: https://forum.cursor.com/t/cursor-mcp-resource-feature-support/50987
 * JIRA: https://jira.mongodb.org/browse/MCP-104 */
type AvailableExport = Pick<StoredExport, "exportName" | "exportTitle" | "exportURI" | "exportPath">;

export type ExportsManagerConfig = Pick<UserConfig, "exportsPath" | "exportTimeoutMs" | "exportCleanupIntervalMs"> & {
    // The maximum number of milliseconds to wait for in-flight operations to
    // settle before shutting down ExportsManager.
    activeOpsDrainTimeoutMs?: number;

    // The maximum number of milliseconds to wait before timing out queued reads
    readTimeout?: number;

    // The maximum number of milliseconds to wait before timing out queued writes
    writeTimeout?: number;
};

type ExportsManagerEvents = {
    closed: [];
    "export-expired": [string];
    "export-available": [string];
};

class OperationAbortedError extends Error {}

export class ExportsManager extends EventEmitter<ExportsManagerEvents> {
    private storedExports: Record<StoredExport["exportName"], StoredExport> = {};
    private exportsCleanupInProgress: boolean = false;
    private exportsCleanupInterval?: NodeJS.Timeout;
    private readonly shutdownController: AbortController = new AbortController();
    private readonly readTimeoutMs: number;
    private readonly writeTimeoutMs: number;
    private readonly exportLocks: Map<string, RWLock> = new Map();

    private constructor(
        private readonly exportsDirectoryPath: string,
        private readonly config: ExportsManagerConfig,
        private readonly logger: LoggerBase
    ) {
        super();
        this.readTimeoutMs = this.config.readTimeout ?? 30_0000; // 30 seconds is the default timeout for an MCP request
        this.writeTimeoutMs = this.config.writeTimeout ?? 120_000; // considering that writes can take time
    }

    public get availableExports(): AvailableExport[] {
        this.assertIsNotShuttingDown();
        return Object.values(this.storedExports)
            .filter((storedExport) => {
                return (
                    storedExport.exportStatus === "ready" &&
                    !isExportExpired(storedExport.exportCreatedAt, this.config.exportTimeoutMs)
                );
            })
            .map(({ exportName, exportTitle, exportURI, exportPath }) => ({
                exportName,
                exportTitle,
                exportURI,
                exportPath,
            }));
    }

    protected init(): void {
        if (!this.exportsCleanupInterval) {
            this.exportsCleanupInterval = setInterval(
                () => void this.cleanupExpiredExports(),
                this.config.exportCleanupIntervalMs
            );
        }
    }
    public async close(): Promise<void> {
        if (this.shutdownController.signal.aborted) {
            return;
        }
        try {
            clearInterval(this.exportsCleanupInterval);
            this.shutdownController.abort();
            await fs.rm(this.exportsDirectoryPath, { force: true, recursive: true });
            this.emit("closed");
        } catch (error) {
            this.logger.error({
                id: LogId.exportCloseError,
                context: "Error while closing ExportsManager",
                message: error instanceof Error ? error.message : String(error),
            });
        }
    }

    public async readExport(exportName: string): Promise<string> {
        try {
            this.assertIsNotShuttingDown();
            exportName = decodeURIComponent(exportName);
            return await this.withLock(
                {
                    exportName,
                    mode: "read",
                    callbackName: "readExport",
                },
                async (): Promise<string> => {
                    const exportHandle = this.storedExports[exportName];
                    if (!exportHandle) {
                        throw new Error("Requested export has either expired or does not exist!");
                    }

                    // This won't happen because of lock synchronization but
                    // keeping it here to make TS happy.
                    if (exportHandle.exportStatus === "in-progress") {
                        throw new Error("Requested export is still being generated!");
                    }

                    const { exportPath } = exportHandle;

                    return fs.readFile(exportPath, { encoding: "utf8", signal: this.shutdownController.signal });
                }
            );
        } catch (error) {
            this.logger.error({
                id: LogId.exportReadError,
                context: `Error when reading export - ${exportName}`,
                message: error instanceof Error ? error.message : String(error),
            });
            throw error;
        }
    }

    public async createJSONExport({
        input,
        exportName,
        exportTitle,
        jsonExportFormat,
    }: {
        input: FindCursor;
        exportName: string;
        exportTitle: string;
        jsonExportFormat: JSONExportFormat;
    }): Promise<AvailableExport> {
        try {
            this.assertIsNotShuttingDown();
            const exportNameWithExtension = validateExportName(ensureExtension(exportName, "json"));
            return await this.withLock(
                {
                    exportName: exportNameWithExtension,
                    mode: "write",
                    callbackName: "createJSONExport",
                },
                (): Promise<AvailableExport> => {
                    if (this.storedExports[exportNameWithExtension]) {
                        return Promise.reject(
                            new Error("Export with same name is either already available or being generated.")
                        );
                    }
                    const exportURI = `exported-data://${encodeURIComponent(exportNameWithExtension)}`;
                    const exportFilePath = path.join(this.exportsDirectoryPath, exportNameWithExtension);
                    const inProgressExport: InProgressExport = (this.storedExports[exportNameWithExtension] = {
                        exportName: exportNameWithExtension,
                        exportTitle,
                        exportPath: exportFilePath,
                        exportURI: exportURI,
                        exportStatus: "in-progress",
                    });

                    void this.startExport({ input, jsonExportFormat, inProgressExport });
                    return Promise.resolve(inProgressExport);
                }
            );
        } catch (error) {
            this.logger.error({
                id: LogId.exportCreationError,
                context: "Error when registering JSON export request",
                message: error instanceof Error ? error.message : String(error),
            });
            throw error;
        }
    }

    private async startExport({
        input,
        jsonExportFormat,
        inProgressExport,
    }: {
        input: FindCursor;
        jsonExportFormat: JSONExportFormat;
        inProgressExport: InProgressExport;
    }): Promise<void> {
        try {
            await this.withLock(
                {
                    exportName: inProgressExport.exportName,
                    mode: "write",
                    callbackName: "startExport",
                },
                async (): Promise<void> => {
                    let pipeSuccessful = false;
                    try {
                        await fs.mkdir(this.exportsDirectoryPath, { recursive: true });
                        const outputStream = createWriteStream(inProgressExport.exportPath);
                        await pipeline(
                            [
                                input.stream(),
                                this.docToEJSONStream(this.getEJSONOptionsForFormat(jsonExportFormat)),
                                outputStream,
                            ],
                            { signal: this.shutdownController.signal }
                        );
                        pipeSuccessful = true;
                    } catch (error) {
                        // If the pipeline errors out then we might end up with
                        // partial and incorrect export so we remove it entirely.
                        await this.silentlyRemoveExport(
                            inProgressExport.exportPath,
                            LogId.exportCreationCleanupError,
                            `Error when removing incomplete export ${inProgressExport.exportName}`
                        );
                        delete this.storedExports[inProgressExport.exportName];
                        throw error;
                    } finally {
                        if (pipeSuccessful) {
                            this.storedExports[inProgressExport.exportName] = {
                                ...inProgressExport,
                                exportCreatedAt: Date.now(),
                                exportStatus: "ready",
                            };
                            this.emit("export-available", inProgressExport.exportURI);
                        }
                        void input.close();
                    }
                }
            );
        } catch (error) {
            this.logger.error({
                id: LogId.exportCreationError,
                context: `Error when generating JSON export for ${inProgressExport.exportName}`,
                message: error instanceof Error ? error.message : String(error),
            });
        }
    }

    private getEJSONOptionsForFormat(format: JSONExportFormat): EJSONOptions | undefined {
        switch (format) {
            case "relaxed":
                return { relaxed: true };
            case "canonical":
                return { relaxed: false };
            default:
                return undefined;
        }
    }

    private docToEJSONStream(ejsonOptions: EJSONOptions | undefined): Transform {
        let docsTransformed = 0;
        return new Transform({
            objectMode: true,
            transform(chunk: unknown, encoding, callback): void {
                try {
                    const doc = EJSON.stringify(chunk, undefined, undefined, ejsonOptions);
                    if (docsTransformed === 0) {
                        this.push("[" + doc);
                    } else {
                        this.push(",\n" + doc);
                    }
                    docsTransformed++;
                    callback();
                } catch (err) {
                    callback(err as Error);
                }
            },
            flush(callback): void {
                if (docsTransformed === 0) {
                    this.push("[]");
                } else {
                    this.push("]");
                }
                callback();
            },
        });
    }

    private async cleanupExpiredExports(): Promise<void> {
        if (this.exportsCleanupInProgress) {
            return;
        }

        this.exportsCleanupInProgress = true;
        try {
            const exportsForCleanup = Object.values({ ...this.storedExports }).filter(
                (storedExport): storedExport is ReadyExport => storedExport.exportStatus === "ready"
            );

            await Promise.allSettled(
                exportsForCleanup.map(async ({ exportPath, exportCreatedAt, exportURI, exportName }) => {
                    if (isExportExpired(exportCreatedAt, this.config.exportTimeoutMs)) {
                        await this.withLock(
                            {
                                exportName,
                                mode: "write",
                                finalize: true,
                                callbackName: "cleanupExpiredExport",
                            },
                            async (): Promise<void> => {
                                delete this.storedExports[exportName];
                                await this.silentlyRemoveExport(
                                    exportPath,
                                    LogId.exportCleanupError,
                                    `Considerable error when removing export ${exportName}`
                                );
                                this.emit("export-expired", exportURI);
                            }
                        );
                    }
                })
            );
        } catch (error) {
            this.logger.error({
                id: LogId.exportCleanupError,
                context: "Error when cleaning up exports",
                message: error instanceof Error ? error.message : String(error),
            });
        } finally {
            this.exportsCleanupInProgress = false;
        }
    }

    private async silentlyRemoveExport(exportPath: string, logId: MongoLogId, logContext: string): Promise<void> {
        try {
            await fs.unlink(exportPath);
        } catch (error) {
            // If the file does not exist or the containing directory itself
            // does not exist then we can safely ignore that error anything else
            // we need to flag.
            if ((error as NodeJS.ErrnoException).code !== "ENOENT") {
                this.logger.error({
                    id: logId,
                    context: logContext,
                    message: error instanceof Error ? error.message : String(error),
                });
            }
        }
    }

    private assertIsNotShuttingDown(): void {
        if (this.shutdownController.signal.aborted) {
            throw new Error("ExportsManager is shutting down.");
        }
    }

    private async withLock<CallbackResult extends Promise<unknown>>(
        lockConfig: {
            exportName: string;
            mode: "read" | "write";
            finalize?: boolean;
            callbackName?: string;
        },
        callback: () => CallbackResult
    ): Promise<Awaited<CallbackResult>> {
        const { exportName, mode, finalize = false, callbackName } = lockConfig;
        const operationName = callbackName ? `${callbackName} - ${exportName}` : exportName;
        let lock = this.exportLocks.get(exportName);
        if (!lock) {
            lock = new RWLock();
            this.exportLocks.set(exportName, lock);
        }

        let lockAcquired: boolean = false;
        const acquireLock = async (): Promise<void> => {
            if (mode === "read") {
                await lock.readLock(this.readTimeoutMs);
            } else {
                await lock.writeLock(this.writeTimeoutMs);
            }
            lockAcquired = true;
        };

        try {
            await Promise.race([
                this.operationAbortedPromise(`Acquire ${mode} lock for ${operationName}`),
                acquireLock(),
            ]);
            return await Promise.race([this.operationAbortedPromise(operationName), callback()]);
        } finally {
            if (lockAcquired) {
                lock.unlock();
            }
            if (finalize) {
                this.exportLocks.delete(exportName);
            }
        }
    }

    private operationAbortedPromise(operationName?: string): Promise<never> {
        return new Promise((_, reject) => {
            const rejectIfAborted = (): void => {
                if (this.shutdownController.signal.aborted) {
                    // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
                    const abortReason = this.shutdownController.signal.reason;
                    const abortMessage =
                        typeof abortReason === "string"
                            ? abortReason
                            : `${operationName ?? "Operation"} aborted - ExportsManager shutting down!`;
                    reject(new OperationAbortedError(abortMessage));
                    this.shutdownController.signal.removeEventListener("abort", rejectIfAborted);
                }
            };

            rejectIfAborted();
            this.shutdownController.signal.addEventListener("abort", rejectIfAborted);
        });
    }

    static init(
        config: ExportsManagerConfig,
        logger: LoggerBase,
        sessionId = new ObjectId().toString()
    ): ExportsManager {
        const exportsDirectoryPath = path.join(config.exportsPath, sessionId);
        const exportsManager = new ExportsManager(exportsDirectoryPath, config, logger);
        exportsManager.init();
        return exportsManager;
    }
}

/**
 * Ensures the path ends with the provided extension */
export function ensureExtension(pathOrName: string, extension: string): string {
    const extWithDot = extension.startsWith(".") ? extension : `.${extension}`;
    if (pathOrName.endsWith(extWithDot)) {
        return pathOrName;
    }
    return `${pathOrName}${extWithDot}`;
}

/**
 * Small utility to decoding and validating provided export name for path
 * traversal or no extension */
export function validateExportName(nameWithExtension: string): string {
    const decodedName = decodeURIComponent(nameWithExtension);
    if (!path.extname(decodedName)) {
        throw new Error("Provided export name has no extension");
    }

    if (decodedName.includes("..") || decodedName.includes("/") || decodedName.includes("\\")) {
        throw new Error("Invalid export name: path traversal hinted");
    }

    return decodedName;
}

export function isExportExpired(createdAt: number, exportTimeoutMs: number): boolean {
    return Date.now() - createdAt > exportTimeoutMs;
}

1	import z from "zod";	2✔
2	import path from "path";	2✔
3	import fs from "fs/promises";	2✔
4	import EventEmitter from "events";	2✔
5	import { createWriteStream } from "fs";	2✔
6	import { FindCursor } from "mongodb";
7	import { EJSON, EJSONOptions, ObjectId } from "bson";	2✔
8	import { Transform } from "stream";	2✔
9	import { pipeline } from "stream/promises";	2✔
10	import { MongoLogId } from "mongodb-log-writer";
11	import { RWLock } from "async-rwlock";	2✔
12
13	import { UserConfig } from "./config.js";
14	import { LoggerBase, LogId } from "./logger.js";	2✔
15
16	export const jsonExportFormat = z.enum(["relaxed", "canonical"]);	2✔
17	export type JSONExportFormat = z.infer<typeof jsonExportFormat>;
18
19	interface CommonExportData {
20	exportName: string;
21	exportTitle: string;
22	exportURI: string;
23	exportPath: string;
24	}
25
26	interface ReadyExport extends CommonExportData {
27	exportStatus: "ready";
28	exportCreatedAt: number;
29	}
30
31	interface InProgressExport extends CommonExportData {
32	exportStatus: "in-progress";
33	}
34
35	type StoredExport = ReadyExport \| InProgressExport;
36
37	/**
38	* Ideally just exportName and exportURI should be made publicly available but
39	* we also make exportPath available because the export tool, also returns the
40	* exportPath in its response when the MCP server is running connected to stdio
41	* transport. The reasoning behind this is that a few clients, Cursor in
42	* particular, as of the date of this writing (7 August 2025) cannot refer to
43	* resource URIs which means they have no means to access the exported resource.
44	* As of this writing, majority of the usage of our MCP server is behind STDIO
45	* transport so we can assume that for most of the usages, if not all, the MCP
46	* server will be running on the same machine as of the MCP client and thus we
47	* can provide the local path to export so that these clients which do not still
48	* support parsing resource URIs, can still work with the exported data. We
49	* expect for clients to catch up and implement referencing resource URIs at
50	* which point it would be safe to remove the `exportPath` from the publicly
51	* exposed properties of an export.
52	*
53	* The editors that we would like to watch out for are Cursor and Windsurf as
54	* they don't yet support working with Resource URIs.
55	*
56	* Ref Cursor: https://forum.cursor.com/t/cursor-mcp-resource-feature-support/50987
57	* JIRA: https://jira.mongodb.org/browse/MCP-104 */
58	type AvailableExport = Pick<StoredExport, "exportName" \| "exportTitle" \| "exportURI" \| "exportPath">;
59
60	export type ExportsManagerConfig = Pick<UserConfig, "exportsPath" \| "exportTimeoutMs" \| "exportCleanupIntervalMs"> & {
61	// The maximum number of milliseconds to wait for in-flight operations to
62	// settle before shutting down ExportsManager.
63	activeOpsDrainTimeoutMs?: number;
64
65	// The maximum number of milliseconds to wait before timing out queued reads
66	readTimeout?: number;
67
68	// The maximum number of milliseconds to wait before timing out queued writes
69	writeTimeout?: number;
70	};
71
72	type ExportsManagerEvents = {
73	closed: [];
74	"export-expired": [string];
75	"export-available": [string];
76	};
77
78	class OperationAbortedError extends Error {}	2✔
79
80	export class ExportsManager extends EventEmitter<ExportsManagerEvents> {	2✔
81	private storedExports: Record<StoredExport["exportName"], StoredExport> = {};	2✔
82	private exportsCleanupInProgress: boolean = false;	2✔
83	private exportsCleanupInterval?: NodeJS.Timeout;
84	private readonly shutdownController: AbortController = new AbortController();	2✔
85	private readonly readTimeoutMs: number;
86	private readonly writeTimeoutMs: number;
87	private readonly exportLocks: Map<string, RWLock> = new Map();	2✔
88
89	private constructor(	2✔
90	private readonly exportsDirectoryPath: string,	128✔
91	private readonly config: ExportsManagerConfig,	128✔
92	private readonly logger: LoggerBase	128✔
93	) {	128✔
94	super();	128✔
95	this.readTimeoutMs = this.config.readTimeout ?? 30_0000; // 30 seconds is the default timeout for an MCP request	128✔
96	this.writeTimeoutMs = this.config.writeTimeout ?? 120_000; // considering that writes can take time	128✔
97	}	128✔
98
99	public get availableExports(): AvailableExport[] {	2✔
100	this.assertIsNotShuttingDown();	26✔
101	return Object.values(this.storedExports)	26✔
102	.filter((storedExport) => {	26✔
103	return (	24✔
104	storedExport.exportStatus === "ready" &&	24✔
105	!isExportExpired(storedExport.exportCreatedAt, this.config.exportTimeoutMs)	20✔
106	);
107	})	26✔
108	.map(({ exportName, exportTitle, exportURI, exportPath }) => ({	26✔
109	exportName,	20✔
110	exportTitle,	20✔
111	exportURI,	20✔
112	exportPath,	20✔
113	}));	26✔
114	}	26✔
115
116	protected init(): void {	2✔
117	if (!this.exportsCleanupInterval) {	128✔
118	this.exportsCleanupInterval = setInterval(	128✔
119	() => void this.cleanupExpiredExports(),	128✔
120	this.config.exportCleanupIntervalMs	128✔
121	);	128✔
122	}	128✔
123	}	128✔
124	public async close(): Promise<void> {	2✔
125	if (this.shutdownController.signal.aborted) {	119✔
126	return;	8✔
127	}	8✔
128	try {	111✔
129	clearInterval(this.exportsCleanupInterval);	111✔
130	this.shutdownController.abort();	111✔
131	await fs.rm(this.exportsDirectoryPath, { force: true, recursive: true });	111✔
132	this.emit("closed");	111✔
133	} catch (error) {	119!
NEW 134	this.logger.error({	×
NEW 135	id: LogId.exportCloseError,	×
NEW 136	context: "Error while closing ExportsManager",	×
NEW 137	message: error instanceof Error ? error.message : String(error),	×
NEW 138	});	×
NEW 139	}	×
140	}	119✔
141
142	public async readExport(exportName: string): Promise<string> {	2✔
143	try {	32✔
144	this.assertIsNotShuttingDown();	32✔
145	exportName = decodeURIComponent(exportName);	32✔
146	return await this.withLock(	32✔
147	{	32✔
148	exportName,	32✔
149	mode: "read",	32✔
150	callbackName: "readExport",	32✔
151	},	32✔
152	async (): Promise<string> => {	32✔
153	const exportHandle = this.storedExports[exportName];	30✔
154	if (!exportHandle) {	30✔
155	throw new Error("Requested export has either expired or does not exist!");	8✔
156	}	8✔
157
158	// This won't happen because of lock synchronization but
159	// keeping it here to make TS happy.
160	if (exportHandle.exportStatus === "in-progress") {	30!
NEW 161	throw new Error("Requested export is still being generated!");	×
NEW 162	}	✔
163
164	const { exportPath } = exportHandle;	22✔
165
166	return fs.readFile(exportPath, { encoding: "utf8", signal: this.shutdownController.signal });	22✔
167	}	30✔
168	);	32✔
169	} catch (error) {	32✔
170	this.logger.error({	10✔
171	id: LogId.exportReadError,	10✔
172	context: `Error when reading export - ${exportName}`,	10✔
173	message: error instanceof Error ? error.message : String(error),	10!
174	});	10✔
175	throw error;	10✔
176	}	10✔
177	}	32✔
178
179	public async createJSONExport({	2✔
180	input,	76✔
181	exportName,	76✔
182	exportTitle,	76✔
183	jsonExportFormat,	76✔
184	}: {	76✔
185	input: FindCursor;
186	exportName: string;
187	exportTitle: string;
188	jsonExportFormat: JSONExportFormat;
189	}): Promise<AvailableExport> {	76✔
190	try {	76✔
191	this.assertIsNotShuttingDown();	76✔
192	const exportNameWithExtension = validateExportName(ensureExtension(exportName, "json"));	76✔
193	return await this.withLock(	76✔
194	{	76✔
195	exportName: exportNameWithExtension,	76✔
196	mode: "write",	76✔
197	callbackName: "createJSONExport",	76✔
198	},	76✔
199	(): Promise<AvailableExport> => {	76✔
200	if (this.storedExports[exportNameWithExtension]) {	74✔
201	return Promise.reject(	2✔
202	new Error("Export with same name is either already available or being generated.")	2✔
203	);	2✔
204	}	2✔
205	const exportURI = `exported-data://${encodeURIComponent(exportNameWithExtension)}`;	72✔
206	const exportFilePath = path.join(this.exportsDirectoryPath, exportNameWithExtension);	72✔
207	const inProgressExport: InProgressExport = (this.storedExports[exportNameWithExtension] = {	72✔
208	exportName: exportNameWithExtension,	72✔
209	exportTitle,	72✔
210	exportPath: exportFilePath,	72✔
211	exportURI: exportURI,	72✔
212	exportStatus: "in-progress",	72✔
213	});	72✔
214
215	void this.startExport({ input, jsonExportFormat, inProgressExport });	72✔
216	return Promise.resolve(inProgressExport);	72✔
217	}	74✔
218	);	76✔
219	} catch (error) {	76✔
220	this.logger.error({	4✔
221	id: LogId.exportCreationError,	4✔
222	context: "Error when registering JSON export request",	4✔
223	message: error instanceof Error ? error.message : String(error),	4!
224	});	4✔
225	throw error;	4✔
226	}	4✔
227	}	76✔
228
229	private async startExport({	2✔
230	input,	72✔
231	jsonExportFormat,	72✔
232	inProgressExport,	72✔
233	}: {	72✔
234	input: FindCursor;
235	jsonExportFormat: JSONExportFormat;
236	inProgressExport: InProgressExport;
237	}): Promise<void> {	72✔
238	try {	72✔
239	await this.withLock(	72✔
240	{	72✔
241	exportName: inProgressExport.exportName,	72✔
242	mode: "write",	72✔
243	callbackName: "startExport",	72✔
244	},	72✔
245	async (): Promise<void> => {	72✔
246	let pipeSuccessful = false;	72✔
247	try {	72✔
248	await fs.mkdir(this.exportsDirectoryPath, { recursive: true });	72✔
249	const outputStream = createWriteStream(inProgressExport.exportPath);	72✔
250	await pipeline(	72✔
251	[	72✔
252	input.stream(),	72✔
253	this.docToEJSONStream(this.getEJSONOptionsForFormat(jsonExportFormat)),	72✔
254	outputStream,	72✔
255	],	72✔
256	{ signal: this.shutdownController.signal }	72✔
257	);	72✔
258	pipeSuccessful = true;	62✔
259	} catch (error) {	72✔
260	// If the pipeline errors out then we might end up with
261	// partial and incorrect export so we remove it entirely.
262	await this.silentlyRemoveExport(	8✔
263	inProgressExport.exportPath,	8✔
264	LogId.exportCreationCleanupError,	8✔
265	`Error when removing incomplete export ${inProgressExport.exportName}`	8✔
266	);	8✔
267	delete this.storedExports[inProgressExport.exportName];	8✔
268	throw error;	8✔
269	} finally {	72✔
270	if (pipeSuccessful) {	70✔
271	this.storedExports[inProgressExport.exportName] = {	62✔
272	...inProgressExport,	62✔
273	exportCreatedAt: Date.now(),	62✔
274	exportStatus: "ready",	62✔
275	};	62✔
276	this.emit("export-available", inProgressExport.exportURI);	62✔
277	}	62✔
278	void input.close();	70✔
279	}	70✔
280	}	72✔
281	);	72✔
282	} catch (error) {	72✔
283	this.logger.error({	8✔
284	id: LogId.exportCreationError,	8✔
285	context: `Error when generating JSON export for ${inProgressExport.exportName}`,	8✔
286	message: error instanceof Error ? error.message : String(error),	8!
287	});	8✔
288	}	8✔
289	}	72✔
290
291	private getEJSONOptionsForFormat(format: JSONExportFormat): EJSONOptions \| undefined {	2✔
292	switch (format) {	72✔
293	case "relaxed":	72✔
294	return { relaxed: true };	64✔
295	case "canonical":	72✔
296	return { relaxed: false };	8✔
297	default:	72!
NEW 298	return undefined;	×
299	}	72✔
300	}	72✔
301
302	private docToEJSONStream(ejsonOptions: EJSONOptions \| undefined): Transform {	2✔
303	let docsTransformed = 0;	70✔
304	return new Transform({	70✔
305	objectMode: true,	70✔
306	transform(chunk: unknown, encoding, callback): void {	70✔
307	try {	92✔
308	const doc = EJSON.stringify(chunk, undefined, undefined, ejsonOptions);	92✔
309	if (docsTransformed === 0) {	92✔
310	this.push("[" + doc);	52✔
311	} else {	92✔
312	this.push(",\n" + doc);	40✔
313	}	40✔
314	docsTransformed++;	92✔
315	callback();	92✔
316	} catch (err) {	92!
NEW 317	callback(err as Error);	×
NEW 318	}	×
319	},	92✔
320	flush(callback): void {	70✔
321	if (docsTransformed === 0) {	62✔
322	this.push("[]");	12✔
323	} else {	62✔
324	this.push("]");	50✔
325	}	50✔
326	callback();	62✔
327	},	62✔
328	});	70✔
329	}	70✔
330
331	private async cleanupExpiredExports(): Promise<void> {	2✔
332	if (this.exportsCleanupInProgress) {	30!
NEW 333	return;	×
NEW 334	}	×
335
336	this.exportsCleanupInProgress = true;	30✔
337	try {	30✔
338	const exportsForCleanup = Object.values({ ...this.storedExports }).filter(	30✔
339	(storedExport): storedExport is ReadyExport => storedExport.exportStatus === "ready"	30✔
340	);	30✔
341
342	await Promise.allSettled(	30✔
343	exportsForCleanup.map(async ({ exportPath, exportCreatedAt, exportURI, exportName }) => {	30✔
344	if (isExportExpired(exportCreatedAt, this.config.exportTimeoutMs)) {	8✔
345	await this.withLock(	4✔
346	{	4✔
347	exportName,	4✔
348	mode: "write",	4✔
349	finalize: true,	4✔
350	callbackName: "cleanupExpiredExport",	4✔
351	},	4✔
352	async (): Promise<void> => {	4✔
353	delete this.storedExports[exportName];	4✔
354	await this.silentlyRemoveExport(	4✔
355	exportPath,	4✔
356	LogId.exportCleanupError,	4✔
357	`Considerable error when removing export ${exportName}`	4✔
358	);	4✔
359	this.emit("export-expired", exportURI);	4✔
360	}	4✔
361	);	4✔
362	}	4✔
363	})	30✔
364	);	30✔
365	} catch (error) {	30!
NEW 366	this.logger.error({	×
NEW 367	id: LogId.exportCleanupError,	×
NEW 368	context: "Error when cleaning up exports",	×
NEW 369	message: error instanceof Error ? error.message : String(error),	×
NEW 370	});	×
371	} finally {	30✔
372	this.exportsCleanupInProgress = false;	30✔
373	}	30✔
374	}	30✔
375
376	private async silentlyRemoveExport(exportPath: string, logId: MongoLogId, logContext: string): Promise<void> {	2✔
377	try {	12✔
378	await fs.unlink(exportPath);	12✔
379	} catch (error) {	12!
380	// If the file does not exist or the containing directory itself
381	// does not exist then we can safely ignore that error anything else
382	// we need to flag.
NEW 383	if ((error as NodeJS.ErrnoException).code !== "ENOENT") {	×
NEW 384	this.logger.error({	×
NEW 385	id: logId,	×
NEW 386	context: logContext,	×
NEW 387	message: error instanceof Error ? error.message : String(error),	×
NEW 388	});	×
NEW 389	}	×
NEW 390	}	×
391	}	12✔
392
393	private assertIsNotShuttingDown(): void {	2✔
394	if (this.shutdownController.signal.aborted) {	134✔
395	throw new Error("ExportsManager is shutting down.");	6✔
396	}	6✔
397	}	134✔
398
399	private async withLock<CallbackResult extends Promise<unknown>>(	2✔
400	lockConfig: {	180✔
401	exportName: string;
402	mode: "read" \| "write";
403	finalize?: boolean;
404	callbackName?: string;
405	},
406	callback: () => CallbackResult	180✔
407	): Promise<Awaited<CallbackResult>> {	180✔
408	const { exportName, mode, finalize = false, callbackName } = lockConfig;	180✔
409	const operationName = callbackName ? `${callbackName} - ${exportName}` : exportName;	180!
410	let lock = this.exportLocks.get(exportName);	180✔
411	if (!lock) {	180✔
412	lock = new RWLock();	76✔
413	this.exportLocks.set(exportName, lock);	76✔
414	}	76✔
415
416	let lockAcquired: boolean = false;	180✔
417	const acquireLock = async (): Promise<void> => {	180✔
418	if (mode === "read") {	180✔
419	await lock.readLock(this.readTimeoutMs);	30✔
420	} else {	180✔
421	await lock.writeLock(this.writeTimeoutMs);	150✔
422	}	150✔
423	lockAcquired = true;	180✔
424	};	180✔
425
426	try {	180✔
427	await Promise.race([	180✔
428	this.operationAbortedPromise(`Acquire ${mode} lock for ${operationName}`),	180✔
429	acquireLock(),	180✔
430	]);	180✔
431	return await Promise.race([this.operationAbortedPromise(operationName), callback()]);	180✔
432	} finally {	180✔
433	if (lockAcquired) {	178✔
434	lock.unlock();	178✔
435	}	178✔
436	if (finalize) {	178✔
437	this.exportLocks.delete(exportName);	4✔
438	}	4✔
439	}	178✔
440	}	180✔
441
442	private operationAbortedPromise(operationName?: string): Promise<never> {	2✔
443	return new Promise((_, reject) => {	360✔
444	const rejectIfAborted = (): void => {	360✔
445	if (this.shutdownController.signal.aborted) {	700✔
446	// eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
447	const abortReason = this.shutdownController.signal.reason;	340✔
448	const abortMessage =	340✔
449	typeof abortReason === "string"	340!
NEW 450	? abortReason	×
451	: `${operationName ?? "Operation"} aborted - ExportsManager shutting down!`;	340!
452	reject(new OperationAbortedError(abortMessage));	340✔
453	this.shutdownController.signal.removeEventListener("abort", rejectIfAborted);	340✔
454	}	340✔
455	};	700✔
456
457	rejectIfAborted();	360✔
458	this.shutdownController.signal.addEventListener("abort", rejectIfAborted);	360✔
459	});	360✔
460	}	360✔
461
462	static init(	2✔
463	config: ExportsManagerConfig,	128✔
464	logger: LoggerBase,	128✔
465	sessionId = new ObjectId().toString()	128✔
466	): ExportsManager {	128✔
467	const exportsDirectoryPath = path.join(config.exportsPath, sessionId);	128✔
468	const exportsManager = new ExportsManager(exportsDirectoryPath, config, logger);	128✔
469	exportsManager.init();	128✔
470	return exportsManager;	128✔
471	}	128✔
472	}	2✔
473
474	/**
475	* Ensures the path ends with the provided extension */
476	export function ensureExtension(pathOrName: string, extension: string): string {	2✔
477	const extWithDot = extension.startsWith(".") ? extension : `.${extension}`;	86!
478	if (pathOrName.endsWith(extWithDot)) {	86✔
479	return pathOrName;	76✔
480	}	76✔
481	return `${pathOrName}${extWithDot}`;	10✔
482	}	10✔
483
484	/**
485	* Small utility to decoding and validating provided export name for path
486	* traversal or no extension */
487	export function validateExportName(nameWithExtension: string): string {	2✔
488	const decodedName = decodeURIComponent(nameWithExtension);	80✔
489	if (!path.extname(decodedName)) {	80✔
490	throw new Error("Provided export name has no extension");	2✔
491	}	2✔
492
493	if (decodedName.includes("..") \|\| decodedName.includes("/") \|\| decodedName.includes("\\")) {	80✔
494	throw new Error("Invalid export name: path traversal hinted");	2✔
495	}	2✔
496
497	return decodedName;	76✔
498	}	76✔
499
500	export function isExportExpired(createdAt: number, exportTimeoutMs: number): boolean {	2✔
501	return Date.now() - createdAt > exportTimeoutMs;	32✔
502	}	32✔

mongodb-js / mongodb-mcp-server / 16876454078

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous