tombl · tombl · Feb 1, 2025 · Feb 1, 2025
diff --git a/compression.ts b/compression.ts
@@ -1,4 +1,6 @@
-import { toText } from "@std/streams";
+import { spawn } from "node:child_process";
+import { once } from "node:events";
+import { text as toText } from "node:stream/consumers";
 
 const ALGORITHMS = [
   "none",
@@ -15,65 +17,55 @@ export function isCompressionAlgorithm(
   return (ALGORITHMS as readonly string[]).includes(value);
 }
 
-function spawn(command: string): TransformStream<Uint8Array, Uint8Array> {
+function spawnTransformer(
+  command: string,
+): TransformStream<Uint8Array, Uint8Array> {
   // warning! this function is horrible. i've tried like 10 different ways to
   // make it work, and this is the only one that works. i'm sorry.
   // also it leaks ops, but not if you pass --trace-leaks
 
-  const proc = new Deno.Command(command, {
-    stdin: "piped",
-    stdout: "piped",
-    stderr: "piped",
-    args: ["-vv"],
-  }).spawn();
-
-  const stdin = proc.stdin.getWriter();
-  let writer: Promise<void>;
+  const proc = spawn(command);
 
   return new TransformStream({
     start(controller) {
-      writer = proc.stdout.pipeTo(
-        new WritableStream({
-          write(chunk) {
-            try {
-              controller.enqueue(chunk);
-            } catch {
-              // intentionally ignore because we're already terminating
-            }
-          },
-          close() {
-            controller.terminate();
-          },
-          abort(reason) {
-            controller.error(reason);
-          },
-        }),
-      );
+      proc.stdout.on("data", (chunk) => {
+        controller.enqueue(chunk);
+      });
+      proc.stdout.on("close", () => {
+        controller.terminate();
+      });
+      proc.stdout.on("error", (err) => {
+        controller.error(err);
+      });
 
-      proc.status.then(
-        async (status) => {
-          if (status.success) {
-            await proc.stderr.cancel();
-          } else {
-            const text = await toText(proc.stderr);
-            controller.error(new Error(text));
-          }
-        },
-        (err) => {
-          controller.error(err);
-        },
-      );
+      proc.on("exit", async (status) => {
+        if (status === 0) {
+          proc.stderr.destroy();
+        } else {
+          const text = await toText(proc.stderr);
+          controller.error(new Error(text));
+        }
+      });
+
+      proc.on("error", (err) => {
+        controller.error(err);
+      });
     },
     async transform(chunk) {
-      await stdin.write(chunk);
+      await new Promise<void>((resolve, reject) =>
+        proc.stdin.write(chunk, (err) => {
+          if (err) reject(err);
+          else resolve();
+        })
+      );
     },
     async flush() {
-      await stdin.close();
-      await writer;
+      await new Promise<void>((resolve) => proc.stdin.end(resolve));
+      await once(proc.stdin, "close");
     },
     async cancel(reason) {
-      await stdin.abort(reason);
-      proc.kill();
+      proc.stdin.destroy(reason);
+      await once(proc.stdin, "close");
     },
   });
 }
@@ -87,11 +79,11 @@ export function createDecompressionStream(
     case "gzip":
       return new DecompressionStream("gzip");
     case "bzip2":
-      return spawn("bzcat");
+      return spawnTransformer("bzcat");
     case "zstd":
-      return spawn("zstdcat");
+      return spawnTransformer("zstdcat");
     case "xz":
-      return spawn("xzcat");
+      return spawnTransformer("xzcat");
     default:
       throw new Error(
         `Unsupported decompression algorithm: ${algorithm satisfies never}`,
@@ -108,3 +100,9 @@ export function getCompressionAlgorithmFromExtension(
   if (name.endsWith("xz")) return "xz";
   return "none";
 }
+
+if (import.meta.main) {
+  await Deno.stdin.readable
+    .pipeThrough(createDecompressionStream("zstd"))
+    .pipeTo(Deno.stderr.writable);
+}
diff --git a/deno.json b/deno.json
@@ -14,7 +14,6 @@
     "exclude": [".*", "__snapshots__", "*_test.ts"]
   },
   "imports": {
-    "@db/sqlite": "jsr:@db/sqlite@^0.12.0",
     "@std/assert": "jsr:@std/assert@^1.0.11",
     "@std/bytes": "jsr:@std/bytes@^1.0.4",
     "@std/cli": "jsr:@std/cli@^1.0.11",
@@ -23,6 +22,7 @@
     "@std/path": "jsr:@std/path@^1.0.8",
     "@std/streams": "jsr:@std/streams@^1.0.8",
     "@std/testing": "jsr:@std/testing@^1.0.9",
+    "libsql": "npm:libsql@^0.5.0-pre.6",
     "p-queue": "npm:p-queue@^8.1.0"
   }
 }
diff --git a/deno.lock b/deno.lock
diff --git a/main.ts b/main.ts
@@ -1,8 +1,8 @@
 #!/usr/bin/env -S deno run -A
-import { Database } from "@db/sqlite";
 import { parseArgs } from "@std/cli";
 import { join } from "@std/path";
 import { SuperConsole } from "https://raw.githubusercontent.com/tombl/superconsole/9bac929/mod.ts";
+import Database from "libsql";
 import Queue from "p-queue";
 import { createDecompressionStream } from "./compression.ts";
 import { Keychain, NIXOS_KEY } from "./keychain.ts";
@@ -22,7 +22,7 @@ if (!(await Deno.stat(DB_PATH).then((f) => f.isFile, () => false))) {
   );
 }
 
-const db = new Database(DB_PATH, { readonly: true, create: false });
+const db = new Database(DB_PATH, { readonly: true, fileMustExist: true });
 
 const keychain = new Keychain();
 await keychain.trust(NIXOS_KEY);
@@ -56,7 +56,8 @@ if (args.help) {
 }
 
 const requestedPackages = db
-  .sql`select name, hash, full_name from packages where name in (${args._})`;
+  .prepare("select name, hash, full_name from packages where name = ?")
+  .all(args._) as Array<{ name: string; hash: string; full_name: string }>;
 
 if (requestedPackages.length !== args._.length) {
   const missing = args._.filter((name) =>