winget-source: enhance syncing with V2 index

README.md

@@ -443,10 +443,11 @@ An alternative HTTP(S) syncing tool, replacing `rclone` and `lftp` in some cases
 
 A handy tool to sync pre-indexed [Windows Package Manager](https://github.com/microsoft/winget-cli) (aka. WinGet) sources.
 
-| Parameter             | Description                                                      |
-| --------------------- | ---------------------------------------------------------------- |
+| Parameter             | Description                                                       |
+| --------------------- | ----------------------------------------------------------------- |
+| `WINGET_FORCE_SYNC`   | Force syncs everything against the upstream. Defaults to `false`. |
 | `WINGET_REPO_URL`     | Sets the URL of upstream. Defaults to [`https://cdn.winget.microsoft.com/cache`](https://cdn.winget.microsoft.com/cache) |
-| `WINGET_REPO_JOBS`    | Parallel jobs. Defaults to 8.                                    |
+| `WINGET_REPO_JOBS`    | Parallel jobs. Defaults to 8.                                     |
 
 ### yukina
 

winget-source/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@ustcmirror/winget-source",
-  "version": "1.2.0",
+  "version": "1.2.1",
   "description": "Sync with pre-indexed WinGet source repository.",
   "main": "sync-repo.js",
   "author": "YR Chen <[email protected]>",

winget-source/sync-repo.js

@@ -19,7 +19,7 @@ import {
 } from './utilities.js'
 
 
-const { parallelLimit, remote, sqlite3, winston } = setupEnvironment();
+const { forceSync, parallelLimit, remote, sqlite3, winston } = setupEnvironment();
 
 /**
  * Sync with the official WinGet repository index.
@@ -35,11 +35,11 @@ async function syncIndex(version, handler) {
     try {
         // download index package to buffer
         const [indexBuffer, modifiedDate, updated] = await syncFile(sourceFilename, true, false);
-        if (!updated) {
+        if (!updated && !forceSync) {
             winston.info(`skip syncing version ${version} from ${remote}`);
             return;
         }
-        assert(indexBuffer !== null, "Failed to get the source index buffer!");
+        assert(Buffer.isBuffer(indexBuffer), 'Failed to get the source index buffer!');
 
         // unpack, extract and load index database
         try {
@@ -58,7 +58,9 @@ async function syncIndex(version, handler) {
         }
 
         // update index package
-        await cacheFileWithURI(sourceFilename, indexBuffer, modifiedDate);
+        if (updated) {
+            await cacheFileWithURI(sourceFilename, indexBuffer, modifiedDate);
+        }
     } catch (error) {
         try {
             await rm(tempDirectory, { recursive: true });
@@ -76,17 +78,17 @@ await syncIndex(2, async (db) => {
     try {
         const packageURIs = buildPackageMetadataURIs(await db.all('SELECT id, hash FROM packages'));
         try {
-            // sync latest package metadata in parallel
-            const manifestURIs = await async.concatLimit(packageURIs, parallelLimit, async (uri) => {
-                const [metadataBuffer] = await syncFile(uri, false);
-                try {
-                    return metadataBuffer ? await buildManifestURIsFromPackageMetadata(metadataBuffer) : [];
-                } catch (error) {
-                    exitWithCode(EX_SOFTWARE, error);
+            // sync latest package metadata and manifests in parallel
+            await async.eachLimit(packageURIs, parallelLimit, async (uri) => {
+                const [metadataBuffer, modifiedDate, updated] = await syncFile(uri, forceSync, false);
+                if (metadataBuffer) {
+                    const manifestURIs = await buildManifestURIsFromPackageMetadata(metadataBuffer);
+                    await async.eachSeries(manifestURIs, async (uri) => await syncFile(uri, forceSync));
+                    if (updated) {
+                        await cacheFileWithURI(uri, metadataBuffer, modifiedDate);
+                    }
                 }
             });
-            // sync latest manifests in parallel
-            await async.eachLimit(manifestURIs, parallelLimit, async (uri) => await syncFile(uri, false));
         } catch (error) {
             exitWithCode(EX_TEMPFAIL, error);
         }
@@ -101,7 +103,7 @@ await syncIndex(1, async (db) => {
         const uris = buildManifestURIs(await db.all('SELECT pathpart FROM manifest ORDER BY rowid DESC'), pathparts);
         // sync latest manifests in parallel
         try {
-            await async.eachLimit(uris, parallelLimit, async (uri) => await syncFile(uri, false));
+            await async.eachLimit(uris, parallelLimit, async (uri) => await syncFile(uri, forceSync));
         } catch (error) {
             exitWithCode(EX_TEMPFAIL, error);
         }

winget-source/sync.sh

@@ -6,6 +6,7 @@
 ## SET IN ENVIRONMENT VARIABLES
 #BIND_ADDRESS=
 #DEBUG=
+#WINGET_FORCE_SYNC=
 #WINGET_REPO_URL=
 #WINGET_REPO_JOBS=
 

winget-source/utilities.js

@@ -49,6 +49,9 @@ const parallelLimit = parseInt(process.env.WINGET_REPO_JOBS ?? 8);
 /** Whether the debug mode is enabled. */
 const debugMode = process.env.DEBUG === 'true';
 
+/** Whether to perform a forced sync. */
+const forceSync = process.env.WINGET_FORCE_SYNC === 'true';
+
 /** Local IP address to be bound to HTTPS requests. */
 const localAddress = process.env.BIND_ADDRESS;
 
@@ -87,10 +90,20 @@ function getRemoteURL(uri) {
  * @returns {Buffer} The decompressed buffer.
  */
 async function decompressMSZIP(buffer) {
-    if (buffer.toString('ascii', 28, 30) != 'CK') {
+    const magicHeader = Buffer.from([0, 0, 0x43, 0x4b]);
+    if (!buffer.subarray(26, 30).equals(magicHeader)) {
         throw new Error('Invalid MSZIP format');
     }
-    return await inflateRaw(buffer.subarray(30));
+    var chunkIndex = 26;
+    var decompressed = Buffer.alloc(0);
+    while ((chunkIndex = buffer.indexOf(magicHeader, chunkIndex)) > -1) {
+        chunkIndex += magicHeader.byteLength;
+        const decompressedChunk = await inflateRaw(buffer.subarray(chunkIndex), {
+            dictionary: decompressed.subarray(-32768)
+        });
+        decompressed = Buffer.concat([decompressed, decompressedChunk]);
+    }
+    return decompressed;
 }
 
 /**
@@ -295,6 +308,7 @@ export function setupEnvironment() {
     }
     return {
         debugMode,
+        forceSync,
         local,
         parallelLimit,
         remote,
@@ -345,7 +359,7 @@ export async function syncFile(uri, update = true, save = true) {
             const localFile = await stat(localPath);
             if (localFile.mtime.getTime() == lastModified.getTime() && localFile.size == contentLength) {
                 winston.debug(`skipped ${uri} because it's up to date`);
-                return [null, lastModified, false];
+                return [await readFile(localPath), lastModified, false];
             }
         }
     }