IPFS Upload Tweaks (#5)

* ipfs fixes:
- allow reconfiguring auto-js-ipfs with new daemonURL, if provided
- enable /ipfs/daemonURL standalone endpoint
- use smaller car sizes for uploads
- for embed mode, only set page if provided
- compute approx. total size of all chunks to be streamed
- send size and total size after each chunk is uploaded
- update ipfs dependencies, including auto-js-ipfs

bump to 0.3.0

Author: ikreymer

package.json

@@ -1,13 +1,13 @@
 {
   "name": "@webrecorder/awp-sw",
   "browser": "dist/sw.js",
+  "version": "0.3.0",
   "main": "index.js",
   "type": "module",
   "repository": {
     "type": "git",
     "url": "git+https://github.com/webrecorder/awp-sw.git"
   },
-  "version": "0.2.3",
   "license": "AGPL-3.0-or-later",
   "exports": {
     ".": "./src/index.js"
@@ -16,14 +16,14 @@
     "src/*"
   ],
   "dependencies": {
-    "@ipld/car": "^5.0.3",
-    "@ipld/unixfs": "^2.0.0",
+    "@ipld/car": "^5.1.1",
+    "@ipld/unixfs": "^2.1.1",
     "@webrecorder/wabac": "^2.16.1",
-    "auto-js-ipfs": "^2.1.1",
+    "auto-js-ipfs": "^2.2.0",
     "client-zip": "^2.3.0",
     "hash-wasm": "^4.9.0",
     "idb": "^7.1.1",
-    "p-queue": "^7.3.0",
+    "p-queue": "^7.3.4",
     "uuid": "^9.0.0",
     "warcio": "^2.1.0"
   },

src/api.js

@@ -25,7 +25,7 @@ class ExtAPI extends API
       "pageTitle": ["c/:coll/pageTitle", "POST"],
       "ipfsAdd": ["c/:coll/ipfs", "POST"],
       "ipfsRemove": ["c/:coll/ipfs", "DELETE"],
-      "ipfsDaemonUrl": ["ipfsDaemonUrl", "POST"],
+      "ipfsDaemonUrl": ["ipfs/daemonUrl", "POST"],
       "publicKey": "publicKey",
     };
   }
@@ -58,6 +58,9 @@ class ExtAPI extends API
     case "ipfsRemove":
       return await this.ipfsRemove(request, params.coll);
 
+    case "ipfsDaemonUrl":
+      return await this.setIPFSDaemonUrlFromBody(request);
+
     default:
       return await super.handleApi(request, params);
     }
@@ -100,6 +103,12 @@ class ExtAPI extends API
       return {error: "collection_not_found"};
     }
 
+    const body = await this.setIPFSDaemonUrlFromBody(request);
+
+    return {coll, body};
+  }
+
+  async setIPFSDaemonUrlFromBody(request) {
     let body;
 
     try {
@@ -111,7 +120,7 @@ class ExtAPI extends API
       body = {};
     }
 
-    return {coll, body};
+    return body;
   }
 
   async startIpfsAdd(event, request, collId) {
@@ -188,17 +197,19 @@ class ExtAPI extends API
 // ===========================================================================
 async function runIPFSAdd(collId, coll, client, opts, collections, replayOpts) {
   let size = 0;
+  let totalSize = 0;
 
   const sendMessage = (type, result = null) => {
     if (client) {
       client.postMessage({
-        type, collId, size, result
+        type, collId, size, result, totalSize
       });
     }
   };
 
-  const {url, cid} = await ipfsAdd(coll, opts, replayOpts, (incSize) => {
+  const {url, cid} = await ipfsAdd(coll, opts, replayOpts, (incSize, _totalSize) => {
     size += incSize;
+    totalSize = _totalSize;
     sendMessage("ipfsProgress");
   });
 

src/ipfsutils.js

@@ -1,21 +1,29 @@
-import { initAutoIPFS } from "@webrecorder/wabac/src/ipfs.js";
 import { Downloader } from "./downloader.js";
 
+import { create as createAutoIPFS } from "auto-js-ipfs";
+
 import * as UnixFS from "@ipld/unixfs";
 import { CarWriter } from "@ipld/car";
 import Queue from "p-queue";
 
 // eslint-disable-next-line no-undef
 const autoipfsOpts = {web3StorageToken: __WEB3_STORAGE_TOKEN__};
 
+let autoipfs = null;
+
 export async function setAutoIPFSUrl(url) {
+  if (autoipfsOpts.daemonURL !== url) {
+    autoipfs = null;
+  }
   autoipfsOpts.daemonURL = url;
 }
 
 export async function ipfsAdd(coll, downloaderOpts = {}, replayOpts = {}, progress = null) {
-  const autoipfs = await initAutoIPFS(autoipfsOpts);
+  if (!autoipfs) {
+    autoipfs = await createAutoIPFS(autoipfsOpts);
+  }
 
-  const filename = "webarchive.wacz";
+  const filename = replayOpts.filename || "webarchive.wacz";
 
   if (replayOpts.customSplits) {
     const ZIP = new Uint8Array([]);
@@ -27,7 +35,7 @@ export async function ipfsAdd(coll, downloaderOpts = {}, replayOpts = {}, progre
   const gzip = replayOpts.gzip !== undefined ? replayOpts.gzip : true;
 
   const dl = new Downloader({...downloaderOpts, coll, filename, gzip});
-  const dlResponse = await dl.download(progress);
+  const dlResponse = await dl.download();
 
   if (!coll.config.metadata.ipfsPins) {
     coll.config.metadata.ipfsPins = [];
@@ -44,9 +52,10 @@ export async function ipfsAdd(coll, downloaderOpts = {}, replayOpts = {}, progre
     capacity = 1048576 * 200;
   } else {
     concur = 3;
-    shardSize = 1024 * 1024 * 10;
+    shardSize = 1024 * 1024 * 5;
     // use default capacity
-    capacity = undefined;
+    // capacity = undefined;
+    capacity = 1048576 * 200;
   }
 
   const { readable, writable } = new TransformStream(
@@ -65,6 +74,17 @@ export async function ipfsAdd(coll, downloaderOpts = {}, replayOpts = {}, progre
     console.warn("Couldn't load favicon");
   }
 
+  const htmlContent = getReplayHtml(dlResponse.filename, replayOpts);
+
+  let totalSize = 0;
+
+  if (coll.config && coll.config.metadata && coll.config.metadata.size) {
+    totalSize = coll.config.metadata.size +
+    swContent.length + uiContent.length + favicon.length + htmlContent.length;
+  }
+
+  progress(0, totalSize);
+
   let url, cid;
 
   const p = readable
@@ -73,16 +93,21 @@ export async function ipfsAdd(coll, downloaderOpts = {}, replayOpts = {}, progre
     .pipeTo(
       new WritableStream({
         write: (res) => {
-          url = res.url;
-          cid = res.cid;
+          if (res.url && res.cid) {
+            url = res.url;
+            cid = res.cid;
+          }
+          if (res.size) {
+            progress(res.size, totalSize);
+          }
         },
       })
     );
 
   ipfsGenerateCar( 
     writable, 
     dlResponse.filename, dlResponse.body,
-    swContent, uiContent, replayOpts,
+    swContent, uiContent, htmlContent, replayOpts,
     downloaderOpts.markers, favicon,
   );
 
@@ -98,7 +123,9 @@ export async function ipfsAdd(coll, downloaderOpts = {}, replayOpts = {}, progre
 }
 
 export async function ipfsRemove(coll) {
-  const autoipfs = await initAutoIPFS(autoipfsOpts);
+  if (!autoipfs) {
+    autoipfs = await createAutoIPFS(autoipfsOpts);
+  }
 
   if (coll.config.metadata.ipfsPins) {
 
@@ -138,16 +165,14 @@ async function ipfsWriteBuff(writer, name, content, dir) {
 
 // ===========================================================================
 export async function ipfsGenerateCar(writable, waczPath, 
-  waczContent, swContent, uiContent, replayOpts, markers, favicon) {
+  waczContent, swContent, uiContent, htmlContent, replayOpts, markers, favicon) {
 
   const writer = UnixFS.createWriter({ writable });
 
   const rootDir = UnixFS.createDirectoryWriter(writer);
 
   const encoder = new TextEncoder();
 
-  const htmlContent = getReplayHtml(waczPath, replayOpts);
-
   await ipfsWriteBuff(writer, "ui.js", uiContent, rootDir);
 
   if (replayOpts.showEmbed) {
@@ -388,7 +413,7 @@ function getReplayHtml(waczPath, replayOpts = {}) {
     </style>
   </head>
   <body>${showEmbed ? `
-    <replay-web-page ${deepLink ? "deepLink=\"true\" " : ""}url="${pageUrl}" loading="${loading || ""}" embed="replay-with-info" src="${waczPath}"></replay-web-page>` : `
+    <replay-web-page ${deepLink ? "deepLink=\"true\" " : ""} ${pageUrl ? `url="${pageUrl}"` : ""} loading="${loading || ""}" embed="replay-with-info" src="${waczPath}"></replay-web-page>` : `
     <replay-app-main source="${waczPath}"></replay-app-main>`
 }
   </body>
@@ -399,9 +424,6 @@ function getReplayHtml(waczPath, replayOpts = {}) {
 
 // Copied from https://github.com/web3-storage/w3protocol/blob/main/packages/upload-client/src/sharding.js
 
-const SHARD_SIZE = 1024 * 1024 * 10;
-const CONCURRENT_UPLOADS = 3;
-
 /**
  * Shard a set of blocks into a set of CAR files. The last block is assumed to
  * be the DAG root and becomes the CAR root CID for the last CAR output.
@@ -412,36 +434,45 @@ export class ShardingStream extends TransformStream {
   /**
    * @param {import('./types').ShardingOptions} [options]
    */
-  constructor(shardSize = SHARD_SIZE) {
+  constructor(shardSize) {
     /** @type {import('@ipld/unixfs').Block[]} */
     let shard = [];
     /** @type {import('@ipld/unixfs').Block[] | null} */
     let readyShard = null;
-    let size = 0;
+    let readySize = 0;
+
+    let currSize = 0;
 
     super({
       async transform(block, controller) {
         if (readyShard != null) {
-          controller.enqueue(await encodeBlocks(readyShard));
+          const blocks = await encodeBlocks(readyShard);
+          const size = readySize;
+          controller.enqueue({blocks, size});
           readyShard = null;
         }
-        if (shard.length && size + block.bytes.length > shardSize) {
+        if (shard.length && currSize + block.bytes.length > shardSize) {
           readyShard = shard;
+          readySize = currSize;
           shard = [];
-          size = 0;
+          currSize = 0;
         }
         shard.push(block);
-        size += block.bytes.length;
+        currSize += block.bytes.length;
       },
 
       async flush(controller) {
         if (readyShard != null) {
-          controller.enqueue(await encodeBlocks(readyShard));
+          const blocks = await encodeBlocks(readyShard);
+          const size = readySize;
+          controller.enqueue({blocks, size});
         }
 
         const rootBlock = shard.at(-1);
         if (rootBlock != null) {
-          controller.enqueue(await encodeBlocks(shard, rootBlock.cid));
+          const blocks = await encodeBlocks(shard, rootBlock.cid);
+          const size = currSize;
+          controller.enqueue({blocks, size});
         }
       },
     });
@@ -460,19 +491,20 @@ export class ShardingStream extends TransformStream {
  * @extends {TransformStream<import('./types').CARFile, import('./types').CARMetadata>}
  */
 export class ShardStoringStream extends TransformStream {
-  constructor(autoipfs, concurrency = CONCURRENT_UPLOADS) {
+  constructor(autoipfs, concurrency) {
     const queue = new Queue({ concurrency });
     const abortController = new AbortController();
     super({
-      async transform(car, controller) {
+      async transform({blocks, size}, controller) {
         void queue.add(
           async () => {
             try {
-              //const opts = { ...options, signal: abortController.signal };
-              //const cid = await add(conf, car, opts)
-              const resUrls = await autoipfs.uploadCAR(car);
+              const cid = blocks.roots[0];
+
+              const resUrls = await autoipfs.uploadCAR(blocks);
+              const url = resUrls[0];
 
-              controller.enqueue({cid: car.roots[0], url: resUrls[0]});
+              controller.enqueue({cid, url, size});
 
               //const { version, roots, size } = car
               //controller.enqueue({ version, roots, cid, size })