1import * as Cloudflare from "alchemy/Cloudflare";
2import * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Schema from "effect/Schema";
5import { Cache } from "./Bindings.ts";
6
7/** KV's floor for both `expirationTtl` and `cacheTtl`. */
8const MIN_TTL = Duration.minutes(1);
9
10const seconds = (duration: Duration.Duration) =>
11 Math.floor(Duration.toSeconds(duration));
12
13/**
14 * A lookup-through cache on the registry's KV namespace, shared by every
15 * isolate of the Worker.
16 *
17 * Concurrent misses each run the lookup and each write; the last write wins.
18 * That is deliberate: a cache that shares an in-flight lookup, like Effect's
19 * `Cache`, has later callers join the first caller's fiber, and workerd drops
20 * a fiber resumed from another request's context once that request
21 * completes, which surfaces as the joining request hanging.
22 *
23 * Reads set `cacheTtl`, so a value can still be served from the colo's edge
24 * cache for up to a minute after it expired; a lookup's `ttl` must leave
25 * that margin. Values with less than a minute left are not stored. KV
26 * failures are logged and fall through to the lookup.
27 */
28export const make = <K, A, I, E, R>(
29 key: (input: K) => string,
30 value: Schema.Codec<A, I>,
31 lookup: (
32 input: K,
33 ) => Effect.Effect<{ value: A; ttl: Duration.Duration }, E, R>,
34) => {
35 const json = Schema.fromJsonString(value);
36 const decode = Schema.decodeUnknownEffect(json);
37 const encode = Schema.encodeEffect(json);
38 return (input: K) =>
39 Effect.gen(function* () {
40 const kv = yield* Cloudflare.KV.ReadWriteNamespace(Cache);
41 const id = key(input);
42 const hit = yield* kv.get(id, { cacheTtl: seconds(MIN_TTL) }).pipe(
43 Effect.flatMap((text) =>
44 text === null ? Effect.succeed(undefined) : decode(text),
45 ),
46 Effect.catch((e) =>
47 Effect.logWarning(`cache read of ${id} failed: ${e}`).pipe(
48 Effect.as(undefined),
49 ),
50 ),
51 );
52 if (hit !== undefined) {
53 return hit;
54 }
55 const fresh = yield* lookup(input);
56 if (Duration.isGreaterThanOrEqualTo(fresh.ttl, MIN_TTL)) {
57 yield* encode(fresh.value).pipe(
58 Effect.flatMap((text) =>
59 kv.put(id, text, { expirationTtl: seconds(fresh.ttl) }),
60 ),
61 Effect.catch((e) =>
62 Effect.logWarning(`cache write of ${id} failed: ${e}`),
63 ),
64 );
65 }
66 return fresh.value;
67 });
68};