PullRequestReadCache.ts

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 Digest from "effect/crypto/Digest";
import * as SecureRandom from "effect/crypto/SecureRandom";
import * as Hex from "effect/encoding/Hex";
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 FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Path from "effect/Path";
import * as Schedule from "effect/Schedule";
import * as Schema from "effect/Schema";
import * as Semaphore from "effect/Semaphore";
import * as KeyValueStore from "effect/persistence/KeyValueStore";
import * as Persistable from "effect/persistence/Persistable";
import * as PersistedCache from "effect/persistence/PersistedCache";
import * as Persistence from "effect/persistence/Persistence";
import * as ServerConfig from "../config.ts";
import { detachStackFrame } from "../git/detachStackFrame.ts";
import { forkParked } from "../serverActivation.ts";

const CONCURRENT_READS = 512;
/** Invalidation excludes only reads whose scopes share its stripe, not every PR read. */
const SCOPE_STRIPES = 64;
// Entry filenames are the URI-encoded `<storeId>:<key>`; the scope revisions file has
// no store id. Effect before 4.1 wrote `<storeId><key>`, still pruned here.
const STORE_ID = "pr-v2";
const ENTRY_FILE_NAME = new RegExp(`^${STORE_ID}(?:%3A)?[0-9a-f]{64}$`);
/**
 * A file untouched for a day is long expired; the day only leaves slack for
 * clock changes.
 */
const ENTRY_FILE_MAX_AGE = Duration.days(1);
type ReadError = PullRequestOperationError | PullRequestUnavailableError;
const revisionCodec = Schema.fromJsonString(
  Schema.Record(
    Schema.String,
    Schema.Struct({ revision: Schema.String, expiresAt: Schema.Finite }),
  ),
);
class Read extends Persistable.Class<{
  payload: { key: string; revision: string; lookup: Effect.Effect<string, ReadError> };
}>()("PullRequestRead", {
  primaryKey: ({ key }) => key,
  success: Schema.Struct({
    payload: Schema.String,
    expiresAt: Schema.Finite,
    revision: Schema.optionalKey(Schema.String),
  }),
  error: Schema.Union([PullRequestOperationError, PullRequestUnavailableError]),
}) {
  matchesRevision(revision: string | undefined): boolean {
    const stored = revision?.split(":") ?? [];
    return this.revision
      .split(":")
      .every((value, index) => value === "" || value === stored[index]);
  }

  [Equal.symbol](that: unknown): boolean {
    return that instanceof Read && that.key === this.key && that.revision === this.revision;
  }
  [Hash.symbol](): number {
    return Hash.string(`${this.key}:${this.revision}`);
  }
}

export class PullRequestReadCache extends Context.Service<
  PullRequestReadCache,
  {
    readonly get: (
      key: string,
      lookup: Effect.Effect<string, ReadError>,
      scopes?: ReadonlyArray<string>,
    ) => Effect.Effect<string, ReadError>;
    readonly invalidate: (scope: string) => Effect.Effect<void>;
  }
>()("t3/pullRequest/PullRequestReadCache") {}

export const make = Effect.gen(function* () {
  const backing = yield* KeyValueStore.KeyValueStore;
  const random = yield* SecureRandom.SecureRandom;
  const digester = yield* Digest.Digest;
  const clock = yield* Clock.Clock;
  let enabled = true;
  const lock = yield* Semaphore.make(CONCURRENT_READS);
  const stripes = yield* Effect.forEach(Array.from({ length: SCOPE_STRIPES }), () =>
    Semaphore.make(CONCURRENT_READS),
  );
  const stripeOf = (scope: string) => (Hash.string(scope) >>> 0) % SCOPE_STRIPES;
  const revisionWrites = yield* Semaphore.make(1);
  const digest = (key: string) =>
    digester.digest(Digest.sha256, new TextEncoder().encode(key)).pipe(Effect.map(Hex.encode));
  const revisions = yield* Cache.makeWith(
    () =>
      backing.get("revisions").pipe(
        Effect.flatMap((raw) => Schema.decodeEffect(revisionCodec)(raw ?? "{}")),
        Effect.catchTags({
          SchemaError: Effect.fn("PullRequestReadCache.recoverRevisions")(function* () {
            // Without scope revisions, persisted PR entries cannot be trusted.
            yield* backing.clear;
            yield* Effect.logWarning("PR cache reset after invalid scope revisions");
            return yield* Schema.decodeEffect(revisionCodec)("{}");
          }),
        }),
      ),
    {
      capacity: 1,
      timeToLive: (exit) => (Exit.isSuccess(exit) ? Duration.infinity : Duration.zero),
    },
  );
  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,
          revision: request.revision,
        })),
      ),
    {
      storeId: STORE_ID,
      timeToLive,
      inMemoryTTL: timeToLive,
      inMemoryCapacity: CONCURRENT_READS,
    },
  ).pipe(Effect.provide(Persistence.layerKvs));
  const refreshes = yield* Cache.makeWith(
    Effect.fn("PullRequestReadCache.refresh")(function* (request: Read) {
      const stored = yield* cache.get(request);
      if (request.matchesRevision(stored.revision)) return stored;
      yield* cache.invalidate(request);
      return yield* cache.get(request);
    }),
    { capacity: CONCURRENT_READS, timeToLive: () => Duration.zero },
  );
  // A cache entry keeps the fiber that filled it, whose lazy stack frames would keep the caller's
  // data (e.g. a whole sync sweep) alive until eviction.
  return PullRequestReadCache.of({
    get: Effect.fn("PullRequestReadCache.get")(function* (key, lookup, scopes = []) {
      if (!enabled) return yield* lookup;
      const read = yield* Effect.cached(lookup);
      return yield* Effect.gen(function* () {
        const current = yield* Cache.get(revisions, undefined);
        const now = clock.currentTimeMillisUnsafe();
        const revision = scopes
          .map((scope) => {
            const value = current[scope];
            return value !== undefined && value.expiresAt > now ? value.revision : "";
          })
          .join(":");
        const request = new Read({ key: yield* digest(key), revision, lookup: read });
        const stored = yield* cache.get(request);
        return (
          request.matchesRevision(stored.revision) ? stored : yield* Cache.get(refreshes, request)
        ).payload;
      }).pipe(
        Effect.catchTags({
          KeyValueStoreError: () => read,
          PersistenceError: () => read,
          SchemaError: () => read,
        }),
        // A read stores its result under the revisions it started with, so an invalidation of
        // one of its scopes must wait for it; reads of other scopes run on.
        (effect) =>
          [...new Set(scopes.map(stripeOf))]
            .toSorted((left, right) => left - right)
            .reduceRight((inner, stripe) => stripes[stripe]!.withPermits(1)(inner), effect),
        lock.withPermits(1),
      );
    }, detachStackFrame),
    invalidate: Effect.fn("PullRequestReadCache.invalidate")(
      function* (scope) {
        const now = clock.currentTimeMillisUnsafe();
        const current = yield* Cache.get(revisions, undefined);
        const next = Object.fromEntries(
          Object.entries(current).filter(([, value]) => value.expiresAt > now),
        );
        next[scope] = { revision: yield* random.uuidV4(), expiresAt: now + 60_000 };
        const encoded = yield* Schema.encodeEffect(revisionCodec)(next);
        yield* backing.set("revisions", encoded);
        yield* Cache.set(revisions, undefined, next);
      },
      Effect.catch(() => {
        enabled = false;
        return Effect.logWarning("PR cache disabled after clearing failed");
      }),
      Effect.uninterruptible,
      revisionWrites.withPermits(1),
      (effect, scope) => stripes[stripeOf(scope)]!.withPermits(CONCURRENT_READS)(effect),
      detachStackFrame,
    ),
  });
});

/**
 * The persisted cache drops an expired entry only when it is read again, so
 * files for PRs nobody reopens would otherwise stay forever.
 */
export const pruneExpiredEntryFiles = Effect.fn("PullRequestReadCache.pruneExpiredEntryFiles")(
  function* (directory: string) {
    const fileSystem = yield* FileSystem.FileSystem;
    const path = yield* Path.Path;
    const cutoff = (yield* Clock.currentTimeMillis) - Duration.toMillis(ENTRY_FILE_MAX_AGE);
    const entries = (yield* fileSystem.readDirectory(directory)).filter((name) =>
      ENTRY_FILE_NAME.test(name),
    );
    const [, failures] = yield* Effect.partition(entries, (name) => {
      const entryPath = path.join(directory, name);
      return fileSystem.stat(entryPath).pipe(
        Effect.flatMap((info) =>
          Option.exists(info.mtime, (mtime) => mtime.getTime() < cutoff)
            ? fileSystem.remove(entryPath)
            : Effect.void,
        ),
        Effect.catchReason("PlatformError", "NotFound", () => Effect.void),
      );
    });
    if (failures.length > 0) {
      yield* Effect.logWarning("Failed to prune some PR cache files", {
        failed: failures.length,
        cause: failures[0],
      });
    }
  },
);

export const layer = Layer.unwrap(
  Effect.gen(function* () {
    const config = yield* ServerConfig.ServerConfig;
    const path = yield* Path.Path;
    const directory = path.join(config.providerStatusCacheDir, "pull-requests");
    return Layer.effect(PullRequestReadCache, make).pipe(
      Layer.provide(
        // Memoize cleanup acquisition with the store; Layer.tap runs per consumer.
        Layer.effectDiscard(
          forkParked(
            pruneExpiredEntryFiles(directory).pipe(
              Effect.catch((cause) =>
                Effect.logWarning("Failed to prune PR cache files", { cause }),
              ),
              Effect.repeat(Schedule.spaced(Duration.hours(1))),
            ),
          ),
        ).pipe(
          Layer.provideMerge(KeyValueStore.layerFileSystem(directory)),
          Layer.catch(() =>
            Layer.effectDiscard(
              Effect.logWarning("PR cache directory unavailable; using memory cache"),
            ).pipe(Layer.provideMerge(KeyValueStore.layerMemory)),
          ),
        ),
      ),
    );
  }),
);