Make blossom requests in parallel

.changeset/selfish-turtles-deny.md

@@ -0,0 +1,5 @@
+---
+"nsite-gateway": minor
+---
+
+Make blossom requests in parallel

src/blossom.ts

@@ -1,3 +1,4 @@
+import { IncomingMessage } from "node:http";
 import { getServersFromServerListEvent, USER_BLOSSOM_SERVER_LIST_KIND } from "blossom-client-sdk";
 
 import { BLOSSOM_SERVERS, MAX_FILE_SIZE } from "./env.js";
@@ -10,27 +11,51 @@ export async function getUserBlossomServers(pubkey: string, relays: string[]) {
   return blossomServersEvent ? getServersFromServerListEvent(blossomServersEvent).map((u) => u.toString()) : undefined;
 }
 
-// TODO: download the file to /tmp and verify it
+/**
-export async function downloadFile(sha256: string, servers = BLOSSOM_SERVERS) {
+ * Downloads a file from multiple servers
-  for (const server of servers) {
+ * @todo download the file to /tmp and verify it
-    try {
+ */
-      const { response } = await makeRequestWithAbort(new URL(sha256, server));
+export function downloadFile(sha256: string, servers = BLOSSOM_SERVERS): Promise<IncomingMessage> {
+  return new Promise((resolve, reject) => {
+    const controllers = new Map<string, AbortController>();
+
+    // make all requests in parallel
+    servers.forEach(async (server) => {
+      const url = new URL(sha256, server);
+      const controller = new AbortController();
+      let res: IncomingMessage | undefined = undefined;
+      controllers.set(server, controller);
 
       try {
+        const response = await makeRequestWithAbort(url, controller);
+        res = response;
+
         if (!response.statusCode) throw new Error("Missing headers or status code");
 
         const size = response.headers["content-length"];
         if (size && parseInt(size) > MAX_FILE_SIZE) throw new Error("File too large");
 
         if (response.statusCode >= 200 && response.statusCode < 300) {
-          return response;
+          // cancel the other requests
-        } else throw new Error("Request failed");
+          for (const [other, abort] of controllers) {
+            if (other !== server) abort.abort();
+          }
+
+          controllers.delete(server);
+          return resolve(response);
+        }
       } catch (error) {
-        // Consume response data to free up memory
+        controllers.delete(server);
-        response.resume();
+        if (res) res.resume();
       }
-    } catch (error) {
+
-      // ignore error, try next server
+      // reject if last
-    }
+      if (controllers.size === 0) reject(new Error("Failed to find blob on servers"));
-  }
+    });
+
+    // reject if all servers don't respond in 30s
+    setTimeout(() => {
+      reject(new Error("Timeout"));
+    }, 30_000);
+  });
 }

src/helpers/http.ts

@@ -4,17 +4,18 @@ const { http, https } = followRedirects;
 
 import agent from "../proxy.js";
 
-export function makeRequestWithAbort(url: URL) {
+export function makeRequestWithAbort(url: URL, controller: AbortController) {
-  return new Promise<{ response: IncomingMessage; controller: AbortController }>((res, rej) => {
+  return new Promise<IncomingMessage>((res, rej) => {
-    const cancelController = new AbortController();
+    controller.signal.addEventListener("abort", () => rej(new Error("Aborted")));
+
     const request = (url.protocol === "https:" ? https : http).get(
       url,
       {
-        signal: cancelController.signal,
+        signal: controller.signal,
         agent,
       },
       (response) => {
-        res({ response, controller: cancelController });
+        res(response);
       },
     );
     request.on("error", (err) => rej(err));