PullRequestReadCache.test.ts

import * as NodeV8 from "node:v8";
import * as NodeVM from "node:vm";
import { assert, it } from "@effect/vitest";
import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
import * as NodeServices from "@effect/platform-node/NodeServices";
import * as Layer from "effect/Layer";
import { PullRequestOperationError } from "@t3tools/contracts";
import * as Deferred from "effect/Deferred";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as Fiber from "effect/Fiber";
import * as FileSystem from "effect/FileSystem";
import * as Option from "effect/Option";
import * as Path from "effect/Path";
import * as KeyValueStore from "effect/persistence/KeyValueStore";
import * as TestClock from "effect/testing/TestClock";
import * as PullRequestReadCache from "./PullRequestReadCache.ts";

NodeV8.setFlagsFromString("--expose-gc");
const collectGarbage = NodeVM.runInNewContext("gc") as () => void;

const cacheLayer = (directory: string) =>
  PullRequestReadCache.make.pipe(Effect.provide(KeyValueStore.layerFileSystem(directory)));

it.layer(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))("PR filesystem cache", (it) => {
  it.effect("clears in-flight reads before a new service can reuse them", () =>
    Effect.gen(function* () {
      const fs = yield* FileSystem.FileSystem;
      const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
      const started = yield* Deferred.make<void>();
      const release = yield* Deferred.make<void>();
      const cache = yield* cacheLayer(directory);
      const read = yield* cache
        .get(
          "summary",
          Deferred.succeed(started, undefined).pipe(
            Effect.andThen(Deferred.await(release)),
            Effect.as("old"),
          ),
          ["pr"],
        )
        .pipe(Effect.forkChild);
      yield* Deferred.await(started);
      const invalidate = yield* cache
        .invalidate("pr")
        .pipe(Effect.forkChild({ startImmediately: true }));
      // Another PR's invalidation does not wait on this read; this one's does.
      yield* cache.invalidate("other-pr");
      assert.isUndefined(invalidate.pollUnsafe());
      yield* Deferred.succeed(release, undefined);
      yield* Fiber.join(read);
      yield* Fiber.join(invalidate);
      const restarted = yield* cacheLayer(directory);
      assert.strictEqual(yield* restarted.get("summary", Effect.succeed("new"), ["pr"]), "new");
    }),
  );

  it.effect("prunes expired entry files the store wrote", () =>
    Effect.gen(function* () {
      const fs = yield* FileSystem.FileSystem;
      const path = yield* Path.Path;
      const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
      const cache = yield* cacheLayer(directory);
      yield* cache.get("summary", Effect.succeed("value"));
      const [entry] = yield* fs.readDirectory(directory);
      assert.isDefined(entry);
      const info = yield* fs.stat(path.join(directory, entry));
      yield* TestClock.setTime(
        Option.getOrThrow(info.mtime).getTime() + Duration.toMillis(Duration.days(2)),
      );
      yield* PullRequestReadCache.pruneExpiredEntryFiles(directory);
      assert.deepStrictEqual(yield* fs.readDirectory(directory), []);
    }),
  );

  it.effect("invalidates only the changed scope across restarts and coalesces its next reads", () =>
    Effect.gen(function* () {
      const fs = yield* FileSystem.FileSystem;
      const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
      let reads = 0;
      const lookup = Effect.sync(() => String(++reads));
      const cache = yield* cacheLayer(directory);
      yield* cache.get("first", lookup, ["project", "pr-1"]);
      yield* cache.get("second", lookup, ["project", "pr-2"]);
      yield* cache.get("third", lookup, ["other-project", "pr-3"]);
      yield* cache.invalidate("pr-1");
      const restarted = yield* cacheLayer(directory);
      const answers = yield* Effect.all(
        Array.from({ length: 10 }, () => restarted.get("first", lookup, ["project", "pr-1"])),
        { concurrency: 10 },
      );
      assert.deepStrictEqual(answers, Array(10).fill("4"));
      assert.strictEqual(yield* restarted.get("second", lookup, ["project", "pr-2"]), "2");
      yield* restarted.invalidate("project");
      const again = yield* cacheLayer(directory);
      assert.strictEqual(yield* again.get("second", lookup, ["project", "pr-2"]), "5");
      assert.strictEqual(yield* again.get("third", lookup, ["other-project", "pr-3"]), "3");
      assert.strictEqual(reads, 5);
      const files = (yield* fs.readDirectory(directory)).length;
      for (let index = 0; index < 3; index++) {
        yield* again.invalidate("pr-1");
        yield* again.get("first", lookup, ["project", "pr-1"]);
      }
      assert.strictEqual((yield* fs.readDirectory(directory)).length, files);
    }),
  );

  it.effect("shares a pending refresh without blocking an unrelated cached PR", () =>
    Effect.gen(function* () {
      const cache = yield* PullRequestReadCache.make;
      yield* cache.get("first", Effect.succeed("old"), ["pr-1"]);
      yield* cache.get("second", Effect.succeed("warm"), ["pr-2"]);
      yield* cache.invalidate("pr-1");
      const started = yield* Deferred.make<void>();
      const release = yield* Deferred.make<void>();
      let reads = 0;
      const refresh = cache.get(
        "first",
        Effect.gen(function* () {
          reads++;
          yield* Deferred.succeed(started, undefined);
          yield* Deferred.await(release);
          return "fresh";
        }),
        ["pr-1"],
      );
      const pending = yield* Effect.all(
        Array.from({ length: 10 }, () => refresh),
        {
          concurrency: 10,
        },
      ).pipe(Effect.forkChild);
      yield* Deferred.await(started);
      assert.strictEqual(yield* cache.get("second", Effect.die("cache miss"), ["pr-2"]), "warm");
      yield* Deferred.succeed(release, undefined);
      assert.deepStrictEqual(yield* Fiber.join(pending), Array(10).fill("fresh"));
      assert.strictEqual(reads, 1);
    }).pipe(Effect.provide(KeyValueStore.layerMemory)),
  );

  it.effect("does not persist failed GitHub reads", () =>
    Effect.gen(function* () {
      const fs = yield* FileSystem.FileSystem;
      const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
      const cache = yield* cacheLayer(directory);
      const error = new PullRequestOperationError({ operation: "summary", detail: "unavailable" });
      yield* cache.get("summary", Effect.fail(error)).pipe(Effect.flip);
      const restarted = yield* cacheLayer(directory);
      assert.strictEqual(yield* restarted.get("summary", Effect.succeed("recovered")), "recovered");
    }),
  );

  it.effect.each(["", "{", '{"pr":{"revision":42}}'])(
    "recovers from malformed scope revisions without reusing stale entries: %j",
    (corrupt) =>
      Effect.gen(function* () {
        const fs = yield* FileSystem.FileSystem;
        const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
        const original = yield* cacheLayer(directory);
        yield* original.get("summary", Effect.succeed("stale"), ["pr"]);
        yield* fs.writeFileString(directory + "/revisions", corrupt);
        const recovered = yield* cacheLayer(directory);
        let reads = 0;
        const lookup = Effect.sync(() => String(++reads));
        assert.strictEqual(yield* recovered.get("summary", lookup, ["pr"]), "1");
        assert.strictEqual(yield* recovered.get("summary", lookup, ["pr"]), "1");
        yield* recovered.invalidate("pr");
        assert.strictEqual(yield* recovered.get("summary", lookup, ["pr"]), "2");
        const restarted = yield* cacheLayer(directory);
        assert.strictEqual(yield* restarted.get("summary", lookup, ["pr"]), "2");
        assert.strictEqual(reads, 2);
      }),
  );
  it.effect("resumes caching after a failed scope read", () =>
    Effect.gen(function* () {
      const backing = yield* KeyValueStore.KeyValueStore;
      let fail = true;
      let reads = 0;
      const cache = yield* PullRequestReadCache.make.pipe(
        Effect.provideService(KeyValueStore.KeyValueStore, {
          ...backing,
          get: (key) =>
            Effect.suspend(() => {
              if (!fail) return backing.get(key);
              fail = false;
              return Effect.fail(
                new KeyValueStore.KeyValueStoreError({ method: "get", message: "unavailable" }),
              );
            }),
        }),
      );
      const read = cache.get(
        "summary",
        Effect.sync(() => String(++reads)),
        ["pr"],
      );
      assert.strictEqual(yield* read, "1");
      assert.strictEqual(yield* read, "2");
      assert.strictEqual(yield* read, "2");
    }).pipe(Effect.provide(KeyValueStore.layerMemory)),
  );

  it.effect("does not keep a caller's data alive after its reads and invalidations", () =>
    Effect.gen(function* () {
      const cache = yield* PullRequestReadCache.make;
      const sweeps: Array<WeakRef<object>> = [];
      // Like a sync sweep: an Effect.fn made per run that closes over that run's data.
      const sweep = Effect.fn("sweep")(function* (index: number) {
        const groups = new Map<string, number>(
          Array.from({ length: 1_000 }, (_, id) => [`group-${id}`, id]),
        );
        sweeps.push(new WeakRef(groups));
        const read = Effect.fn("read")(function* (pr: string) {
          const size = groups.size;
          yield* cache.invalidate(pr);
          const answer = yield* cache.get(`${pr}-${index}`, Effect.succeed("answer"), [pr]);
          return size + answer.length;
        });
        yield* Effect.forEach(["pr-1", "pr-2"], read, { discard: true });
      });
      for (let index = 0; index < 5; index++) yield* sweep(index);
      // WeakRef targets stay alive until the current job ends, so yield before collecting.
      yield* Effect.yieldNow;
      collectGarbage();
      assert.strictEqual(sweeps.filter((sweepData) => sweepData.deref() !== undefined).length, 0);
    }).pipe(Effect.provide(KeyValueStore.layerMemory)),
  );

  it.effect("cancels abandoned reads without blocking invalidation", () =>
    Effect.gen(function* () {
      const cache = yield* PullRequestReadCache.make;
      const started = yield* Deferred.make<void>();
      const read = yield* cache
        .get("summary", Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)), [
          "pr",
        ])
        .pipe(Effect.forkChild);
      yield* Deferred.await(started);
      yield* Fiber.interrupt(read);
      yield* cache.invalidate("pr");
      assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh");
    }).pipe(Effect.provide(KeyValueStore.layerMemory)),
  );

  it.effect(
    "finishes the in-memory revision update when invalidation is canceled after writing",
    () =>
      Effect.gen(function* () {
        const backing = yield* KeyValueStore.KeyValueStore;
        const written = yield* Deferred.make<void>();
        const release = yield* Deferred.make<void>();
        const cache = yield* PullRequestReadCache.make.pipe(
          Effect.provideService(KeyValueStore.KeyValueStore, {
            ...backing,
            set: (key, value) =>
              backing
                .set(key, value)
                .pipe(
                  Effect.andThen(
                    key === "revisions"
                      ? Deferred.succeed(written, undefined).pipe(
                          Effect.andThen(Deferred.await(release)),
                        )
                      : Effect.void,
                  ),
                ),
          }),
        );
        yield* cache.get("summary", Effect.succeed("old"), ["pr"]);
        const invalidation = yield* cache.invalidate("pr").pipe(Effect.forkChild);
        yield* Deferred.await(written);
        const interrupt = yield* Fiber.interrupt(invalidation).pipe(
          Effect.forkChild({ startImmediately: true }),
        );
        yield* Deferred.succeed(release, undefined);
        yield* Fiber.join(interrupt);
        assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh");
      }).pipe(Effect.provide(KeyValueStore.layerMemory)),
  );
});