Support CloudFlare workers (#217)

* Update integration tests to exclude package-lock.json

* Update CloudFlare worker example to test using streaming LLM

* Run the CloudFlare worker integration test in CI

* Add support for CloudFlare workers

Author: aron

.github/workflows/ci.yml

@@ -34,7 +34,7 @@ jobs:
     strategy:
       matrix:
         node-version: [18.x, 20.x]
-        suite: [commonjs, esm, typescript]
+        suite: [commonjs, esm, typescript, cloudflare-worker]
         # See supported Node.js release schedule at https://nodejs.org/en/about/previous-releases
 
     env:
@@ -49,6 +49,8 @@ jobs:
           cache: "npm"
       # Build a production tarball and run the integration tests against it.
       - run: |
+          test "${{ matrix.suite }}" = "cloudflare-worker" && echo "REPLICATE_API_TOKEN=${{ secrets.REPLICATE_API_TOKEN }}" > integration/${{ matrix.suite }}/.dev.vars
           PKG_TARBALL=$(npm --loglevel error pack)
+          npm --prefix integration/${{ matrix.suite }} install
           npm --prefix integration/${{ matrix.suite }} install "file:/./$PKG_TARBALL"
           npm --prefix integration/${{ matrix.suite }} test

index.js

@@ -47,7 +47,9 @@ class Replicate {
    * @param {Function} [options.fetch] - Fetch function to use. Defaults to `globalThis.fetch`
    */
   constructor(options = {}) {
-    this.auth = options.auth || process.env.REPLICATE_API_TOKEN;
+    this.auth =
+      options.auth ||
+      (typeof process !== "undefined" ? process.env.REPLICATE_API_TOKEN : null);
     this.userAgent =
       options.userAgent || `replicate-javascript/${packageJSON.version}`;
     this.baseUrl = options.baseUrl || "https://api.replicate.com/v1";

integration/cloudflare-worker/.gitignore

@@ -1,2 +1,4 @@
 .wrangler
 node_modules
+.dev.vars
+

integration/cloudflare-worker/.npmrc

@@ -0,0 +1,2 @@
+package-lock=false
+

integration/cloudflare-worker/index.js

@@ -1,25 +1,31 @@
 import Replicate from "replicate";
 
-const replicate = new Replicate();
-
 export default {
-  async fetch(_request, _env, _ctx) {
-    const stream = new ReadableStream({
-      async start(controller) {
-        for await (const event of replicate.stream(
-          "replicate/hello-world:5c7d5dc6dd8bf75c1acaa8565735e7986bc5b66206b55cca93cb72c9bf15ccaa",
-          {
-            input: {
-              text: "Colin CloudFlare",
-            },
-          }
-        )) {
-          controller.enqueue(`${event}`);
+  async fetch(_request, env, _ctx) {
+    const replicate = new Replicate({ auth: env.REPLICATE_API_TOKEN });
+
+    try {
+      const output = replicate.stream(
+        "replicate/canary:30e22229542eb3f79d4f945dacb58d32001b02cc313ae6f54eef27904edf3272",
+        {
+          input: {
+            text: "Colin CloudFlare",
+          },
         }
-        controller.close();
-      },
-    });
+      );
+      const stream = new ReadableStream({
+        async start(controller) {
+          for await (const event of output) {
+            controller.enqueue(new TextEncoder().encode(`${event}`));
+          }
+          controller.enqueue(new TextEncoder().encode("\n"));
+          controller.close();
+        },
+      });
 
-    return new Response(stream);
+      return new Response(stream);
+    } catch (err) {
+      return Response("", { status: 500 });
+    }
   },
 };

integration/cloudflare-worker/index.test.js

@@ -1,28 +1,36 @@
 // https://developers.cloudflare.com/workers/wrangler/api/#unstable_dev
 import { unstable_dev as dev } from "wrangler";
-import { test, after, before } from "node:test";
+import { test, after, before, describe } from "node:test";
 import assert from "node:assert";
 
-let worker;
+/** @type {import("wrangler").UnstableDevWorker} */
+describe("CloudFlare Worker", () => {
+  let worker;
 
-before(async () => {
-  worker = await dev("index.js", {
-    experimental: { disableExperimentalWarning: true },
+  before(async () => {
+    worker = await dev("./index.js", {
+      port: 3000,
+      experimental: { disableExperimentalWarning: true },
+    });
   });
-});
 
-after(async () => {
-  if (!worker) {
-    // If no worker the before hook failed to run and the process will hang.
-    process.exit(1);
-  }
-  await worker.stop();
-});
+  after(async () => {
+    if (!worker) {
+      // If no worker the before hook failed to run and the process will hang.
+      process.exit(1);
+    }
+    await worker.stop();
+  });
 
-test("worker streams back a response", { timeout: 1000 }, async (t) => {
-  const resp = await worker.fetch("/", { signal: t.signal });
-  const text = await resp.text();
+  test("worker streams back a response", { timeout: 1000 }, async () => {
+    const resp = await worker.fetch();
+    const text = await resp.text();
 
-  assert.ok(resp.ok, "status is 2xx");
-  assert(text.length > 0, "body.length is greater than 0");
+    assert.ok(resp.ok, "status is 2xx");
+    assert(text.length > 0, "body.length is greater than 0");
+    assert(
+      text.includes("Colin CloudFlare"),
+      "body includes stream characters"
+    );
+  });
 });