feat(privateNpmRegistry): add endpoints to fetch with config send via body and queue tar retrieval

hulutter · Thomasr · commit 98defe4172d1 · 2024-08-23T12:37:36.000-04:00
diff --git a/server/node-service/src/controllers/npm.ts b/server/node-service/src/controllers/npm.ts
@@ -1,8 +1,8 @@
 import "../common/logger";
 import fs from "fs/promises";
 import { spawn } from "child_process";
-import { Request as ServerRequest, Response as ServerResponse } from "express";
-import { NpmRegistryService, NpmRegistryConfigEntry } from "../services/npmRegistry";
+import { response, Request as ServerRequest, Response as ServerResponse } from "express";
+import { NpmRegistryService, NpmRegistryConfigEntry, NpmRegistryConfig } from "../services/npmRegistry";
 
 
 type PackagesVersionInfo = {
@@ -19,21 +19,76 @@ type PackagesVersionInfo = {
 };
 
 
+class PackageProcessingQueue {
+    public static readonly promiseRegistry: {[packageId: string]: Promise<void>} = {};
+    public static readonly resolveRegistry: {[packageId: string]:() => void} = {};
+
+    public static add(packageId: string) {
+        PackageProcessingQueue.promiseRegistry[packageId] = new Promise<void>((resolve) => {
+            PackageProcessingQueue.resolveRegistry[packageId] = resolve;
+        });
+    }
+
+    public static has(packageId: string) {
+        return !!PackageProcessingQueue.promiseRegistry[packageId];
+    }
+
+    public static wait(packageId: string) {
+        if (!PackageProcessingQueue.has(packageId)) {
+            return Promise.resolve();
+        }   
+        return PackageProcessingQueue.promiseRegistry[packageId];
+    }
+
+    public static resolve(packageId: string) {
+        if (!PackageProcessingQueue.has(packageId)) {
+            return;
+        }
+        PackageProcessingQueue.resolveRegistry[packageId]();
+        delete PackageProcessingQueue.promiseRegistry[packageId];
+        delete PackageProcessingQueue.resolveRegistry[packageId];
+    }
+}
+
+
 /**
  * Initializes npm registry cache directory
  */
 const CACHE_DIR = process.env.NPM_CACHE_DIR || "/tmp/npm-package-cache";
 try {
     fs.mkdir(CACHE_DIR, { recursive: true });
 } catch (error) {
-    console.error("Error creating cache directory", error);
+    logger.error("Error creating cache directory", error);
 }
 
 
 /**
  * Fetches package info from npm registry
  */
+
 const fetchRegistryBasePath = "/npm/registry";
+
+export async function fetchRegistryWithConfig(request: ServerRequest, response: ServerResponse) {
+    try {
+        const path = request.path.replace(fetchRegistryBasePath, "");
+        logger.info(`Fetch registry info for path: ${path}`);
+
+        const pathPackageInfo = parsePackageInfoFromPath(path);
+        if (!pathPackageInfo) {
+            return response.status(400).send(`Invalid package path: ${path}`);
+        }
+
+        const registryConfig: NpmRegistryConfig = request.body;
+        const config = NpmRegistryService.getRegistryEntryForPackageWithConfig(pathPackageInfo.packageId, registryConfig);
+
+        const registryResponse = await fetchFromRegistry(path, config);
+        response.json(await registryResponse.json());
+    } catch (error) {
+        logger.error("Error fetching registry", error);
+        response.status(500).send("Internal server error");
+    }
+}
+
 export async function fetchRegistry(request: ServerRequest, response: ServerResponse) {
     try {
         const path = request.path.replace(fetchRegistryBasePath, "");
@@ -43,10 +98,9 @@ export async function fetchRegistry(request: ServerRequest, response: ServerResp
         if (!pathPackageInfo) {
             return response.status(400).send(`Invalid package path: ${path}`);
         }
-        const {organization, name} = pathPackageInfo;
-        const packageName = organization ? `@${organization}/${name}` : name;
 
-        const registryResponse = await fetchFromRegistry(packageName, path);
+        const config = NpmRegistryService.getInstance().getRegistryEntryForPackage(pathPackageInfo.packageId);
+        const registryResponse = await fetchFromRegistry(path, config);
         response.json(await registryResponse.json());
     } catch (error) {
         logger.error("Error fetching registry", error);
@@ -58,53 +112,100 @@ export async function fetchRegistry(request: ServerRequest, response: ServerResp
 /**
  * Fetches package files from npm registry if not yet cached
  */
+
 const fetchPackageFileBasePath = "/npm/package";
+
+export async function fetchPackageFileWithConfig(request: ServerRequest, response: ServerResponse) {
+    const path = request.path.replace(fetchPackageFileBasePath, "");
+    logger.info(`Fetch file for path with config: ${path}`);
+    
+    const pathPackageInfo = parsePackageInfoFromPath(path);
+    if (!pathPackageInfo) {
+        return response.status(400).send(`Invalid package path: ${path}`);
+    }
+    
+    const registryConfig: NpmRegistryConfig = request.body;
+    const config = NpmRegistryService.getRegistryEntryForPackageWithConfig(pathPackageInfo.packageId, registryConfig);
+
+    fetchPackageFileInner(request, response, config);
+}
+
 export async function fetchPackageFile(request: ServerRequest, response: ServerResponse) {
+    const path = request.path.replace(fetchPackageFileBasePath, "");
+    logger.info(`Fetch file for path: ${path}`);
+
+    const pathPackageInfo = parsePackageInfoFromPath(path);
+    if (!pathPackageInfo) {
+        return response.status(400).send(`Invalid package path: ${path}`);
+    }
+
+    const config = NpmRegistryService.getInstance().getRegistryEntryForPackage(pathPackageInfo.packageId);
+    fetchPackageFileInner(request, response, config);
+}
+
+async function fetchPackageFileInner(request: ServerRequest, response: ServerResponse, config: NpmRegistryConfigEntry) {
     try {
-        const path = request.path.replace(fetchPackageFileBasePath, "");
-        logger.info(`Fetch file for path: ${path}`);
-    
+        const path = request.path.replace(fetchPackageFileBasePath, "");    
         const pathPackageInfo = parsePackageInfoFromPath(path);
         if (!pathPackageInfo) {
             return response.status(400).send(`Invalid package path: ${path}`);
         }
      
-        logger.info(`Fetch file for package: ${JSON.stringify(pathPackageInfo)}`);
-        const {organization, name, version, file} = pathPackageInfo;
-        const packageName = organization ? `@${organization}/${name}` : name;
+        logger.debug(`Fetch file for package: ${JSON.stringify(pathPackageInfo)}`);
+        const {packageId, version, file} = pathPackageInfo;
         let packageVersion = version;
     
         let packageInfo: PackagesVersionInfo | null = null;
         if (version === "latest") {
-            const packageInfo: PackagesVersionInfo = await fetchPackageInfo(packageName);
+            const packageInfo: PackagesVersionInfo|null = await fetchPackageInfo(packageId, config);
+            if (packageInfo === null) {
+                return response.status(404).send("Not found");
+            }
             packageVersion = packageInfo["dist-tags"].latest;
         }
+
+        // Wait for package to be processed if it's already being processed
+        if (PackageProcessingQueue.has(packageId)) {
+            logger.info("Waiting for package to be processed", packageId);
+            await PackageProcessingQueue.wait(packageId);
+        }
     
-        const packageBaseDir = `${CACHE_DIR}/${packageName}/${packageVersion}/package`;
+        const packageBaseDir = `${CACHE_DIR}/${packageId}/${packageVersion}/package`;
         const packageExists = await fileExists(`${packageBaseDir}/package.json`)
         if (!packageExists) {
-            if (!packageInfo) {
-                packageInfo = await fetchPackageInfo(packageName);
-            }
-    
-            if (!packageInfo || !packageInfo.versions || !packageInfo.versions[packageVersion]) {
-                return response.status(404).send("Not found");
+            try {
+                logger.info(`Package does not exist, fetch from registy: ${packageId}@${packageVersion}`);
+                PackageProcessingQueue.add(packageId);
+                if (!packageInfo) {
+                    packageInfo = await fetchPackageInfo(packageId, config);
+                }
+                
+                if (!packageInfo || !packageInfo.versions || !packageInfo.versions[packageVersion]) {
+                    return response.status(404).send("Not found");
+                }
+                
+                const tarball = packageInfo.versions[packageVersion].dist.tarball;
+                logger.info(`Fetching tarball: ${tarball}`);
+                await fetchAndUnpackTarball(tarball, packageId, packageVersion, config);
+            } catch (error) {
+                logger.error("Error fetching package tarball", error);
+                return response.status(500).send("Internal server error");
+            } finally {
+                PackageProcessingQueue.resolve(packageId);
             }
-            
-            const tarball = packageInfo.versions[packageVersion].dist.tarball;
-            logger.info("Fetching tarball...", tarball);
-            await fetchAndUnpackTarball(tarball, packageName, packageVersion);
+        } else {
+            logger.info(`Package already exists, serve from cache: ${packageBaseDir}/${file}`)
         }
   
         // Fallback to index.mjs if index.js is not present
         if (file === "index.js" && !await fileExists(`${packageBaseDir}/${file}`)) {
-            logger.info("Fallback to index.mjs");
+            logger.debug("Fallback to index.mjs");
             return response.sendFile(`${packageBaseDir}/index.mjs`);
         }
     
         return response.sendFile(`${packageBaseDir}/${file}`);
     } catch (error) {
-        logger.error("Error fetching package file", error);
+        logger.error(`Error fetching package file: ${error} ${(error as {stack: string})?.stack?.toString()}`);
         response.status(500).send("Internal server error");
     }
 };
@@ -114,26 +215,22 @@ export async function fetchPackageFile(request: ServerRequest, response: ServerR
  * Helpers
  */
 
-function parsePackageInfoFromPath(path: string): {organization: string, name: string, version: string, file: string} | undefined { 
-    logger.info(`Parse package info from path: ${path}`);
+function parsePackageInfoFromPath(path: string): {packageId: string, organization: string, name: string, version: string, file: string} | undefined { 
     //@ts-ignore - regex groups
-    const packageInfoRegex = /^\/?(?<fullName>(?:@(?<organization>[a-z0-9-~][a-z0-9-._~]*)\/)?(?<name>[a-z0-9-~][a-z0-9-._~]*))(?:@(?<version>[-a-z0-9><=_.^~]+))?\/(?<file>[^\r\n]*)?$/;
+    const packageInfoRegex = /^\/?(?<packageId>(?:@(?<organization>[a-z0-9-~][a-z0-9-._~]*)\/)?(?<name>[a-z0-9-~][a-z0-9-._~]*))(?:@(?<version>[-a-z0-9><=_.^~]+))?\/(?<file>[^\r\n]*)?$/;
     const matches = path.match(packageInfoRegex);
-    logger.info(`Parse package matches: ${JSON.stringify(matches)}`);
     if (!matches?.groups) {
         return;
     }
 
-    let {organization, name, version, file} = matches.groups;
+    let {packageId, organization, name, version, file} = matches.groups;
     version = /^\d+\.\d+\.\d+(-[\w\d]+)?/.test(version) ? version : "latest";
     
-    return {organization, name, version, file};
+    return {packageId, organization, name, version, file};
 }
 
-function fetchFromRegistry(packageName: string, urlOrPath: string): Promise<Response> {
-    const config: NpmRegistryConfigEntry = NpmRegistryService.getInstance().getRegistryEntryForPackage(packageName);
+function fetchFromRegistry(urlOrPath: string, config: NpmRegistryConfigEntry): Promise<Response> {
     const registryUrl = config?.registry.url;
-
     const headers: {[key: string]: string} = {}; 
     switch (config?.registry.auth.type) {
         case "none":
@@ -154,31 +251,35 @@ function fetchFromRegistry(packageName: string, urlOrPath: string): Promise<Resp
         url = `${registryUrl}${separator}${urlOrPath}`;
     }
 
-    logger.debug(`Fetch from registry: ${url}`);
+    logger.debug(`Fetch from registry: ${url}, ${JSON.stringify(headers)}`);
     return fetch(url, {headers});
 }
 
-function fetchPackageInfo(packageName: string): Promise<PackagesVersionInfo> {
-    return fetchFromRegistry(packageName, packageName).then(res => res.json());
+function fetchPackageInfo(packageName: string, config: NpmRegistryConfigEntry): Promise<PackagesVersionInfo|null> {
+    return fetchFromRegistry(`/${packageName}`, config).then(res => {
+        if (!res.ok) {
+            logger.error(`Failed to fetch package info for package ${packageName}: ${res.statusText}`);
+            return null;
+        }
+        return res.json();
+    });
 }
 
-async function fetchAndUnpackTarball(url: string, packageName: string, packageVersion: string) {
-    const response: Response = await fetchFromRegistry(packageName, url);
+async function fetchAndUnpackTarball(url: string, packageId: string, packageVersion: string, config: NpmRegistryConfigEntry) {
+    const response: Response = await fetchFromRegistry(url, config);
     const arrayBuffer = await response.arrayBuffer();
     const buffer = Buffer.from(arrayBuffer);
     const path = `${CACHE_DIR}/${url.split("/").pop()}`;
     await fs.writeFile(path, buffer);
-    await unpackTarball(path, packageName, packageVersion);
+    await unpackTarball(path, packageId, packageVersion);
     await fs.unlink(path);
 }
 
-async function unpackTarball(path: string, packageName: string, packageVersion: string) {
-    const destinationPath = `${CACHE_DIR}/${packageName}/${packageVersion}`;
+async function unpackTarball(path: string, packageId: string, packageVersion: string) {
+    const destinationPath = `${CACHE_DIR}/${packageId}/${packageVersion}`;
     await fs.mkdir(destinationPath, { recursive: true });
     await new Promise<void> ((resolve, reject) => {
         const tar = spawn("tar", ["-xvf", path, "-C", destinationPath]);
-        tar.stdout.on("data", (data) => logger.info(data));
-        tar.stderr.on("data", (data) => console.error(data));
         tar.on("close", (code) => {
             code === 0 ? resolve() : reject();
         });
diff --git a/server/node-service/src/routes/apiRouter.ts b/server/node-service/src/routes/apiRouter.ts
@@ -16,4 +16,7 @@ apiRouter.post("/validatePluginDataSourceConfig", pluginControllers.validatePlug
 apiRouter.get("/npm/registry/*", npmControllers.fetchRegistry);
 apiRouter.get("/npm/package/*", npmControllers.fetchPackageFile);
 
+apiRouter.post("/npm/registry/*", npmControllers.fetchRegistryWithConfig);
+apiRouter.post("/npm/package/*", npmControllers.fetchPackageFileWithConfig);
+
 export default apiRouter;
diff --git a/server/node-service/src/services/npmRegistry.ts b/server/node-service/src/services/npmRegistry.ts
@@ -108,4 +108,26 @@ export class NpmRegistryService {
 
         return config;
     }
+
+    public static getRegistryEntryForPackageWithConfig(packageName: string, registryConfig: NpmRegistryConfig): NpmRegistryConfigEntry {
+        registryConfig = NpmRegistryService.sortRegistryConfig(registryConfig);
+        const config: NpmRegistryConfigEntry | undefined = registryConfig.find(entry => {
+            if (entry.scope.type === "organization") {
+                return packageName.startsWith(entry.scope.pattern);
+            } else if (entry.scope.type === "package") {
+                return packageName === entry.scope.pattern;
+            } else {
+                return true;
+            }
+        });
+
+        if (!config) {
+            logger.info(`No registry entry found for package: ${packageName}`);
+            return NpmRegistryService.DEFAULT_REGISTRY;
+        } else {
+            logger.info(`Found registry entry for package: ${packageName} -> ${config.registry.url}`);
+        }
+
+        return config;
+    }
 }
