fix(server): preserve recent PR reads across server restarts

apps/server/src/pullRequest/PullRequestReadCache.test.ts

@@ -0,0 +1,78 @@
+import { assert, it } from "@effect/vitest";
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import { PullRequestOperationError } from "@t3tools/contracts";
+import * as Deferred from "effect/Deferred";
+import * as Effect from "effect/Effect";
+import * as Fiber from "effect/Fiber";
+import * as FileSystem from "effect/FileSystem";
+import * as Layer from "effect/Layer";
+import * as TestClock from "effect/testing/TestClock";
+import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore";
+import * as Persistence from "effect/unstable/persistence/Persistence";
+import * as PullRequestReadCache from "./PullRequestReadCache.ts";
+
+const cacheLayer = (directory: string) =>
+  PullRequestReadCache.make.pipe(
+    Effect.provide(
+      Persistence.layerKvs.pipe(Layer.provideMerge(KeyValueStore.layerFileSystem(directory))),
+    ),
+  );
+
+it.layer(NodeServices.layer)("PR filesystem cache", (it) => {
+  it.effect("reuses files after restart and respects the original expiry", () =>
+    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 first = yield* cacheLayer(directory);
+      const key = "long/repository/key".repeat(100);
+      assert.strictEqual(yield* first.get(key, lookup), "1");
+      yield* TestClock.adjust("59 seconds");
+      const restarted = yield* cacheLayer(directory);
+      assert.strictEqual(yield* restarted.get(key, lookup), "1");
+      yield* TestClock.adjust("1 second");
+      assert.strictEqual(yield* restarted.get(key, lookup), "2");
+      assert.strictEqual(reads, 2);
+      assert.strictEqual((yield* fs.readDirectory(directory)).length, 1);
+    }),
+  );
+
+  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"),
+          ),
+        )
+        .pipe(Effect.forkChild);
+      yield* Deferred.await(started);
+      const invalidate = yield* cache.invalidate.pipe(Effect.forkChild({ startImmediately: true }));
+      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")), "new");
+    }),
+  );
+
+  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");
+    }),
+  );
+});

apps/server/src/pullRequest/PullRequestReadCache.ts

@@ -0,0 +1,125 @@
+import * as Cache from "effect/Cache";
+import * as Clock from "effect/Clock";
+import * as Equal from "effect/Equal";
+import * as Hash from "effect/Hash";
+import { PullRequestOperationError, PullRequestUnavailableError } from "@t3tools/contracts";
+import * as Crypto from "effect/Crypto";
+import * as Encoding from "effect/Encoding";
+import * as Option from "effect/Option";
+import * as Context from "effect/Context";
+import * as Duration from "effect/Duration";
+import * as Effect from "effect/Effect";
+import * as Exit from "effect/Exit";
+import * as Layer from "effect/Layer";
+import * as Path from "effect/Path";
+import * as Schema from "effect/Schema";
+import * as Semaphore from "effect/Semaphore";
+import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore";
+import * as Persistable from "effect/unstable/persistence/Persistable";
+import * as PersistedCache from "effect/unstable/persistence/PersistedCache";
+import * as Persistence from "effect/unstable/persistence/Persistence";
+import { ServerConfig } from "../config.ts";
+
+const CONCURRENT_READS = 512;
+type ReadError = PullRequestOperationError | PullRequestUnavailableError;
+
+class Read extends Persistable.Class<{
+  payload: { key: string; lookup: Effect.Effect<string, ReadError> };
+}>()("PullRequestRead", {
+  primaryKey: ({ key }) => key,
+  success: Schema.Struct({ payload: Schema.String, expiresAt: Schema.Finite }),
+  error: Schema.Union([PullRequestOperationError, PullRequestUnavailableError]),
+}) {
+  [Equal.symbol](that: unknown): boolean {
+    return that instanceof Read && that.key === this.key;
+  }
+  [Hash.symbol](): number {
+    return Hash.string(this.key);
+  }
+}
+
+export class PullRequestReadCache extends Context.Service<
+  PullRequestReadCache,
+  {
+    readonly get: (
+      key: string,
+      lookup: Effect.Effect<string, ReadError>,
+    ) => Effect.Effect<string, ReadError>;
+    readonly invalidate: Effect.Effect<void>;
+  }
+>()("t3/pullRequest/PullRequestReadCache") {}
+
+export const make = Effect.gen(function* () {
+  const backing = yield* KeyValueStore.KeyValueStore;
+  const crypto = yield* Crypto.Crypto;
+  const clock = yield* Clock.Clock;
+  let enabled = true;
+  const lock = yield* Semaphore.make(CONCURRENT_READS);
+  const timeToLive: Persistable.TimeToLiveFn<Read> = (exit) =>
+    Exit.isSuccess(exit)
+      ? Duration.millis(Math.max(0, exit.value.expiresAt - clock.currentTimeMillisUnsafe()))
+      : Duration.zero;
+  const cache = yield* PersistedCache.make(
+    (request: Read) =>
+      request.lookup.pipe(
+        Effect.map((payload) => ({ payload, expiresAt: clock.currentTimeMillisUnsafe() + 60_000 })),
+      ),
+    {
+      storeId: "pr-v2",
+      timeToLive,
+      inMemoryTTL: timeToLive,
+      inMemoryCapacity: CONCURRENT_READS,
+    },
+  );
+  return PullRequestReadCache.of({
+    get: Effect.fn("PullRequestReadCache.get")(function* (key, lookup) {
+      if (!enabled) return yield* lookup;
+      const digest = yield* crypto
+        .digest("SHA-256", new TextEncoder().encode(key))
+        .pipe(Effect.option);
+      if (Option.isNone(digest)) return yield* lookup;
+      const read = yield* Effect.cached(lookup);
+      return yield* cache
+        .get(new Read({ key: Encoding.encodeHex(digest.value), lookup: read }))
+        .pipe(
+          Effect.map((result) => result.payload),
+          Effect.catchTags({
+            PersistenceError: () => read,
+            SchemaError: () => read,
+          }),
+          Effect.uninterruptible,
+          lock.withPermits(1),
+        );
+    }),
+    // Let existing reads finish before clearing, so they cannot repopulate stale entries.
+    invalidate: Cache.invalidateAll(cache.inMemory).pipe(
+      Effect.andThen(backing.clear),
+      Effect.catch(() => {
+        enabled = false;
+        return Effect.logWarning("PR cache disabled after clearing failed");
+      }),
+      lock.withPermits(CONCURRENT_READS),
+    ),
+  });
+});
+
+export const layer = Layer.unwrap(
+  Effect.gen(function* () {
+    const config = yield* ServerConfig;
+    const path = yield* Path.Path;
+    return Layer.effect(PullRequestReadCache, make).pipe(
+      Layer.provide(Persistence.layerKvs),
+      Layer.provide(
+        KeyValueStore.layerFileSystem(
+          path.join(config.providerStatusCacheDir, "pull-requests"),
+        ).pipe(
+          Layer.catch(() =>
+            Layer.effectDiscard(
+              Effect.logWarning("PR cache directory unavailable; using memory cache"),
+            ).pipe(Layer.provideMerge(KeyValueStore.layerMemory)),
+          ),
+        ),
+      ),
+    );
+  }),
+);

apps/server/src/pullRequest/PullRequestService.test.ts

@@ -1,3 +1,6 @@
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore";
+import * as Persistence from "effect/unstable/persistence/Persistence";
 import { assert, it } from "@effect/vitest";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
@@ -24,6 +27,7 @@ import {
 } from "./PullRequestProvider.ts";
 import { PullRequestProviderRegistry, fromProviders } from "./PullRequestProviderRegistry.ts";
 import * as PullRequestService from "./PullRequestService.ts";
+import * as PullRequestReadCache from "./PullRequestReadCache.ts";
 
 function project(input: {
   readonly id: string;
@@ -198,6 +202,11 @@ function makeService(input: {
             }),
         }),
         SourceControlRateLimit.layer,
+        Layer.effect(PullRequestReadCache.PullRequestReadCache, PullRequestReadCache.make).pipe(
+          Layer.provide(Persistence.layerKvs),
+          Layer.provide(KeyValueStore.layerMemory),
+          Layer.provide(NodeServices.layer),
+        ),
       ),
     ),
   );