Data loader support (#89)

* First cut and data loader support

* Respect async on close()

* Switch back to watch instead of watchFile

* Fix type error

* Update tests

* Add build support for data loaders

* Move data loader cache to .observablehq/cache

* Add .sh extension

Author: wiltsecarpenter

.gitignore

@@ -1,4 +1,5 @@
 .DS_Store
+.observablehq
 dist/
 node_modules/
 test/output/*-changed.*

src/build.ts

@@ -1,12 +1,13 @@
-import {access, constants, copyFile, mkdir, readFile, writeFile} from "node:fs/promises";
+import {access, constants, copyFile, readFile, writeFile} from "node:fs/promises";
 import {basename, dirname, join, normalize, relative} from "node:path";
 import {cwd} from "node:process";
 import {fileURLToPath} from "node:url";
 import {parseArgs} from "node:util";
-import {visitFiles, visitMarkdownFiles} from "./files.js";
+import {getStats, prepareOutput, visitFiles, visitMarkdownFiles} from "./files.js";
 import {readPages} from "./navigation.js";
 import {renderServerless} from "./render.js";
 import {makeCLIResolver} from "./resolver.js";
+import {findLoader, runCommand} from "./dataloader.js";
 
 const EXTRA_FILES = new Map([["node_modules/@observablehq/runtime/dist/runtime.js", "_observablehq/runtime.js"]]);
 
@@ -52,6 +53,17 @@ async function build(context: CommandContext) {
   for (const file of files) {
     const sourcePath = join(sourceRoot, file);
     const outputPath = join(outputRoot, "_file", file);
+    const stats = await getStats(sourcePath);
+    if (!stats) {
+      const {path} = await findLoader("", sourcePath);
+      if (!path) {
+        console.error("missing referenced file", sourcePath);
+        continue;
+      }
+      console.log("generate", path, "→", outputPath);
+      await runCommand(path, outputPath);
+      continue;
+    }
     console.log("copy", sourcePath, "→", outputPath);
     await prepareOutput(outputPath);
     await copyFile(sourcePath, outputPath);
@@ -67,12 +79,6 @@ async function build(context: CommandContext) {
   }
 }
 
-async function prepareOutput(outputPath: string): Promise<void> {
-  const outputDir = dirname(outputPath);
-  if (outputDir === ".") return;
-  await mkdir(outputDir, {recursive: true});
-}
-
 const USAGE = `Usage: observable build [--root dir] [--output dir]`;
 
 interface CommandContext {

src/dataloader.ts

@@ -0,0 +1,57 @@
+import {open} from "node:fs/promises";
+import {spawn} from "node:child_process";
+import {join} from "node:path";
+import {getStats, prepareOutput} from "./files.js";
+import {renameSync, unlinkSync} from "node:fs";
+
+const runningCommands = new Map<string, Promise<void>>();
+
+export async function runCommand(commandPath: string, outputPath: string) {
+  if (runningCommands.has(commandPath)) return runningCommands.get(commandPath);
+  const command = new Promise<void>((resolve, reject) => {
+    const outputTempPath = outputPath + ".tmp";
+    prepareOutput(outputTempPath).then(() =>
+      open(outputTempPath, "w").then((cacheFd) => {
+        const cacheFileStream = cacheFd.createWriteStream({highWaterMark: 1024 * 1024});
+        try {
+          const subprocess = spawn(commandPath, [], {
+            argv0: commandPath,
+            //cwd: dirname(commandPath), // TODO: Need to change commandPath to be relative this?
+            windowsHide: true,
+            stdio: ["ignore", "pipe", "inherit"]
+            // timeout: // time in ms
+            // signal: // abort signal
+          });
+          subprocess.stdout.on("data", (data) => cacheFileStream.write(data));
+          subprocess.on("error", (error) => console.error(`${commandPath}: ${error.message}`));
+          subprocess.on("close", (code) => {
+            cacheFd.close().then(() => {
+              if (code === 0) {
+                renameSync(outputTempPath, outputPath);
+              } else {
+                unlinkSync(outputTempPath);
+              }
+              resolve();
+            }, reject);
+          });
+        } catch (error) {
+          reject(error);
+        } finally {
+          runningCommands.delete(commandPath);
+        }
+      })
+    );
+  });
+  runningCommands.set(commandPath, command);
+  return command;
+}
+
+export async function findLoader(root: string, name: string) {
+  // TODO: It may be more efficient use fs.readdir
+  for (const ext of [".js", ".ts", ".sh"]) {
+    const path = join(root, name) + ext;
+    const stats = await getStats(path);
+    if (stats) return {path, stats};
+  }
+  return {};
+}

src/files.ts

@@ -1,6 +1,7 @@
+import type {Stats} from "node:fs";
 import {accessSync, constants, statSync} from "node:fs";
-import {readdir, stat} from "node:fs/promises";
-import {extname, join, normalize, relative} from "node:path";
+import {mkdir, readdir, stat} from "node:fs/promises";
+import {dirname, extname, join, normalize, relative} from "node:path";
 import {isNodeError} from "./error.js";
 
 // A file is local if it exists in the root folder or a subfolder.
@@ -50,3 +51,18 @@ export async function* visitFiles(root: string): AsyncGenerator<string> {
     }
   }
 }
+
+export async function getStats(path: string): Promise<Stats | undefined> {
+  try {
+    return await stat(path);
+  } catch (error) {
+    if (!isNodeError(error) || error.code !== "ENOENT") throw error;
+  }
+  return;
+}
+
+export async function prepareOutput(outputPath: string): Promise<void> {
+  const outputDir = dirname(outputPath);
+  if (outputDir === ".") return;
+  await mkdir(outputDir, {recursive: true});
+}

src/javascript.ts

@@ -1,6 +1,5 @@
 import {Parser, tokTypes, type Options} from "acorn";
 import mime from "mime";
-import {isLocalFile} from "./files.js";
 import {findAwaits} from "./javascript/awaits.js";
 import {findDeclarations} from "./javascript/declarations.js";
 import {findFeatures} from "./javascript/features.js";
@@ -43,13 +42,12 @@ export interface ParseOptions {
 }
 
 export function transpileJavaScript(input: string, options: ParseOptions): Transpile {
-  const {root, id} = options;
+  const {id} = options;
   try {
     const node = parseJavaScript(input, options);
     const databases = node.features.filter((f) => f.type === "DatabaseClient").map((f) => ({name: f.name}));
     const files = node.features
       .filter((f) => f.type === "FileAttachment")
-      .filter((f) => isLocalFile(f.name, root))
       .map((f) => ({name: f.name, mimeType: mime.getType(f.name)}));
     const inputs = Array.from(new Set<string>(node.references.map((r) => r.name)));
     const output = new Sourcemap(input);

src/preview.ts

@@ -14,21 +14,26 @@ import {readPages} from "./navigation.js";
 import {renderPreview} from "./render.js";
 import type {CellResolver} from "./resolver.js";
 import {makeCLIResolver} from "./resolver.js";
+import {findLoader, runCommand} from "./dataloader.js";
+import {getStats} from "./files.js";
 
 const publicRoot = join(dirname(fileURLToPath(import.meta.url)), "..", "public");
+const cacheRoot = join(dirname(fileURLToPath(import.meta.url)), "..", ".observablehq", "cache");
 
 class Server {
   private _server: ReturnType<typeof createServer>;
   private _socketServer: WebSocketServer;
   readonly port: number;
   readonly hostname: string;
   readonly root: string;
+  readonly cacheRoot: string;
   private _resolver: CellResolver | undefined;
 
-  constructor({port, hostname, root}: CommandContext) {
+  constructor({port, hostname, root, cacheRoot}: CommandContext) {
     this.port = port;
     this.hostname = hostname;
     this.root = root;
+    this.cacheRoot = cacheRoot;
     this._server = createServer();
     this._server.on("request", this._handleRequest);
     this._socketServer = new WebSocketServer({server: this._server});
@@ -52,7 +57,34 @@ class Server {
       } else if (pathname.startsWith("/_observablehq/")) {
         send(req, pathname.slice("/_observablehq".length), {root: publicRoot}).pipe(res);
       } else if (pathname.startsWith("/_file/")) {
-        send(req, pathname.slice("/_file".length), {root: this.root}).pipe(res);
+        const path = pathname.slice("/_file".length);
+        const filepath = join(this.root, path);
+        try {
+          await access(filepath, constants.R_OK);
+          send(req, pathname.slice("/_file".length), {root: this.root}).pipe(res);
+        } catch (error) {
+          if (isNodeError(error) && error.code !== "ENOENT") {
+            throw error;
+          }
+        }
+
+        // Look for a data loader for this file.
+        const {path: loaderPath, stats: loaderStat} = await findLoader(this.root, path);
+        if (loaderStat) {
+          const cachePath = join(this.cacheRoot, filepath);
+          const cacheStat = await getStats(cachePath);
+          if (cacheStat && cacheStat.mtimeMs > loaderStat.mtimeMs) {
+            send(req, filepath, {root: this.cacheRoot}).pipe(res);
+            return;
+          }
+          if (!(loaderStat.mode & constants.S_IXUSR)) {
+            throw new HttpError("Data loader is not executable", 404);
+          }
+          await runCommand(loaderPath, cachePath);
+          send(req, filepath, {root: this.cacheRoot}).pipe(res);
+          return;
+        }
+        throw new HttpError("Not found", 404);
       } else {
         if (normalize(pathname).startsWith("..")) throw new Error("Invalid path: " + pathname);
         let path = join(this.root, pathname);
@@ -122,11 +154,37 @@ class Server {
 }
 
 class FileWatchers {
-  watchers: FSWatcher[];
+  watchers: FSWatcher[] = [];
+
+  constructor(
+    readonly root: string,
+    readonly files: {name: string}[],
+    readonly cb: (name: string) => void
+  ) {}
+
+  async watchAll() {
+    const fileset = [...new Set(this.files.map(({name}) => name))];
+    for (const name of fileset) {
+      const watchPath = await FileWatchers.getWatchPath(this.root, name);
+      let prevState = await getStats(watchPath);
+      this.watchers.push(
+        watch(watchPath, async () => {
+          const newState = await getStats(watchPath);
+          // Ignore if the file was truncated or not modified.
+          if (prevState?.mtimeMs === newState?.mtimeMs || newState?.size === 0) return;
+          prevState = newState;
+          this.cb(name);
+        })
+      );
+    }
+  }
 
-  constructor(root: string, files: {name: string}[], cb: (name: string) => void) {
-    const fileset = [...new Set(files.map(({name}) => name))];
-    this.watchers = fileset.map((name) => watch(join(root, name), async () => cb(name)));
+  static async getWatchPath(root: string, name: string) {
+    const path = join(root, name);
+    const stats = await getStats(path);
+    if (stats?.isFile()) return path;
+    const {path: loaderPath, stats: loaderStat} = await findLoader(root, name);
+    return loaderStat?.isFile() ? loaderPath : path;
   }
 
   close() {
@@ -165,6 +223,7 @@ function handleWatch(socket: WebSocket, options: {root: string; resolver: CellRe
   async function refreshMarkdown(path: string): Promise<WatchListener<string>> {
     let current = await readMarkdown(path, root);
     attachmentWatcher = new FileWatchers(root, current.parse.files, refreshAttachment(current.parse));
+    await attachmentWatcher.watchAll();
     return async function watcher(event) {
       switch (event) {
         case "rename": {
@@ -247,6 +306,7 @@ interface CommandContext {
   root: string;
   hostname: string;
   port: number;
+  cacheRoot: string;
 }
 
 function makeCommandContext(): CommandContext {
@@ -274,7 +334,8 @@ function makeCommandContext(): CommandContext {
   return {
     root: normalize(values.root).replace(/\/$/, ""),
     hostname: values.hostname ?? process.env.HOSTNAME ?? "127.0.0.1",
-    port: values.port ? +values.port : process.env.PORT ? +process.env.PORT : 3000
+    port: values.port ? +values.port : process.env.PORT ? +process.env.PORT : 3000,
+    cacheRoot
   };
 }
 

test/output/dynamic-import.js

@@ -1,4 +1,4 @@
-define({id: "0", outputs: ["foo"], body: async () => {
+define({id: "0", outputs: ["foo"], files: [{"name":"./bar.js","mimeType":"application/javascript"}], body: async () => {
 const foo = await import("/_file/bar.js");
 return {foo};
 }});

test/output/static-import.js

@@ -1,4 +1,4 @@
-define({id: "0", inputs: ["display"], outputs: ["foo"], body: async (display) => {
+define({id: "0", inputs: ["display"], outputs: ["foo"], files: [{"name":"./bar.js","mimeType":"application/javascript"}], body: async (display) => {
 const {foo} = await import("/_file/bar.js");
 
 display(foo);