1import * as kvs from "@distilled.cloud/aws/cloudfront-keyvaluestore";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import type { Input } from "../../Input.ts";
5import * as Provider from "../../Provider.ts";
6import { Resource } from "../../Resource.ts";
7import type { Providers } from "../Providers.ts";
8import {
9 extractValue,
10 getKvsEtag,
11 isKvsPreconditionFailed,
12 cappedKvsRetrySchedule,
13 retryForKvsReadiness,
14 withKvsRegionFn,
15} from "./common.ts";
16
17export interface KvEntriesProps {
18 /** ARN of the CloudFront KeyValueStore. */
19 store: string;
20 /** Namespace prefix for all keys. Keys are stored as `{namespace}:{key}`. */
21 namespace: string;
22 /** Map of key → value entries to manage. */
23 entries: Record<string, Input<string>>;
24 /** Whether to delete keys under this namespace that are not in `entries`. @default false */
25 purge?: boolean;
26}
27
28export interface KvEntries extends Resource<
29 "AWS.CloudFront.KvEntries",
30 KvEntriesProps,
31 {
32 /** ARN of the CloudFront KeyValueStore. */
33 store: string;
34 /** Namespace prefix used for keys. */
35 namespace: string;
36 /** Current entries managed under the namespace. */
37 entries: Record<string, string>;
38 },
39 never,
40 Providers
41> {}
42
43/**
44 * Manages namespaced key-value entries in a CloudFront KeyValueStore.
45 *
46 * Entries are stored with a `{namespace}:{key}` prefix to allow multiple
47 * logical groups within a single store. Updates use batched optimistic
48 * concurrency with automatic ETag retry.
49 * ### Managing Entries
50 * **Example:** Basic Entries
51 * ```typescript
52 * const entries = yield* KvEntries("Routes", {
53 * store: store.keyValueStoreArn,
54 * namespace: "routes",
55 * entries: {
56 * "/": "/index.html",
57 * "/about": "/about.html",
58 * },
59 * });
60 * ```
61 *
62 * **Example:** Purge Stale Keys
63 * ```typescript
64 * const entries = yield* KvEntries("Routes", {
65 * store: store.keyValueStoreArn,
66 * namespace: "routes",
67 * entries: { "/": "/index.html" },
68 * purge: true,
69 * });
70 * ```
71 *
72 * @resource
73 */
74export const KvEntries = Resource<KvEntries>("AWS.CloudFront.KvEntries");
75
76const BATCH_SIZE = 50;
77type ResolvedEntries = {
78 [key: string]: Input.Resolve<KvEntriesProps["entries"][string]>;
79};
80
81const resolveEntries = (entries: KvEntriesProps["entries"]): ResolvedEntries =>
82 Object.fromEntries(
83 Object.entries(entries).map(([key, value]) => [key, value]),
84 ) as ResolvedEntries;
85
86export const KvEntriesProvider = () =>
87 Provider.effect(
88 KvEntries,
89 // @ts-expect-error
90 Effect.gen(function* () {
91 const collectAllKeys = Effect.fn(function* (store: string) {
92 const keys: { Key: string; Value: string }[] = [];
93 let nextToken: string | undefined;
94 do {
95 const resp = yield* kvs.listKeys({
96 KvsARN: store,
97 NextToken: nextToken,
98 });
99 for (const item of resp.Items ?? []) {
100 keys.push({
101 Key: item.Key,
102 Value: extractValue(item.Value),
103 });
104 }
105 nextToken = resp.NextToken;
106 } while (nextToken);
107 return keys;
108 });
109
110 const sendBatch = Effect.fn(function* (
111 store: string,
112 etag: string,
113 puts: kvs.PutKeyRequestListItem[],
114 deletes: kvs.DeleteKeyRequestListItem[],
115 ) {
116 return yield* kvs.updateKeys({
117 KvsARN: store,
118 IfMatch: etag,
119 Puts: puts.length > 0 ? puts : undefined,
120 Deletes: deletes.length > 0 ? deletes : undefined,
121 });
122 });
123
124 const batchUpdateKeys = Effect.fn(function* (
125 store: string,
126 etag: string | undefined,
127 puts: kvs.PutKeyRequestListItem[],
128 deletes: kvs.DeleteKeyRequestListItem[],
129 ) {
130 let remainingPuts = puts;
131 let remainingDeletes = deletes;
132 let currentEtag: string | undefined = etag;
133
134 while (remainingPuts.length > 0 || remainingDeletes.length > 0) {
135 const batchPuts = remainingPuts.slice(0, BATCH_SIZE);
136 const batchDeletes = remainingDeletes.slice(
137 0,
138 BATCH_SIZE - batchPuts.length,
139 );
140
141 // A precondition failure means a concurrent writer (another
142 // KvEntries / KvRoutesUpdate on the same store) advanced the
143 // etag between our read and this batch — the retry MUST re-read
144 // the etag or it can never succeed. `stale` drops the cached
145 // etag on the failed attempt so the retried generator fetches a
146 // fresh one.
147 let stale = currentEtag;
148 const resp = yield* Effect.gen(function* () {
149 const attemptEtag = stale ?? (yield* getKvsEtag(store));
150 stale = undefined;
151 return yield* sendBatch(
152 store,
153 attemptEtag,
154 batchPuts,
155 batchDeletes,
156 );
157 }).pipe(
158 Effect.retry({
159 while: (error) =>
160 error._tag === "ValidationException" &&
161 isKvsPreconditionFailed(error),
162 schedule: cappedKvsRetrySchedule,
163 }),
164 );
165
166 currentEtag = resp.ETag;
167 remainingPuts = remainingPuts.slice(batchPuts.length);
168 remainingDeletes = remainingDeletes.slice(batchDeletes.length);
169 }
170 });
171
172 const upload = Effect.fn(function* (
173 store: string,
174 namespace: string,
175 entries: ResolvedEntries,
176 oldEntries: ResolvedEntries | undefined,
177 ) {
178 const puts: kvs.PutKeyRequestListItem[] = [];
179 for (const [key, value] of Object.entries(entries)) {
180 if (oldEntries === undefined || oldEntries[key] !== value) {
181 puts.push({ Key: `${namespace}:${key}`, Value: value as any });
182 }
183 }
184 if (puts.length > 0) {
185 yield* batchUpdateKeys(store, undefined, puts, []);
186 }
187 });
188
189 const purge = Effect.fn(function* (
190 store: string,
191 namespace: string,
192 keepEntries: ResolvedEntries | undefined,
193 ) {
194 const allKeys = yield* collectAllKeys(store);
195 const prefix = `${namespace}:`;
196 const deletes: kvs.DeleteKeyRequestListItem[] = [];
197 for (const item of allKeys) {
198 if (!item.Key.startsWith(prefix)) continue;
199 const unprefixed = item.Key.slice(prefix.length);
200 if (keepEntries && unprefixed in keepEntries) continue;
201 deletes.push({ Key: item.Key });
202 }
203 if (deletes.length > 0) {
204 yield* batchUpdateKeys(store, undefined, [], deletes);
205 }
206 });
207
208 return {
209 // Non-listable: a KvEntries resource is a logical group of key/value
210 // data keyed entirely by its parent store ARN + namespace (both chosen
211 // by the user). There is no account-wide API to enumerate these groups,
212 // and the entries are managed content rather than nuke-able infra.
213 list: () => Effect.succeed([]),
214 read: withKvsRegionFn(
215 Effect.fn(function* ({ output }) {
216 return output;
217 }),
218 ),
219 reconcile: withKvsRegionFn(
220 Effect.fn(function* ({ news, output }) {
221 return yield* retryForKvsReadiness(
222 Effect.gen(function* () {
223 const entries = resolveEntries(news.entries);
224
225 // Observe — diff against the prior persisted entries so
226 // we only `Put` keys whose values actually changed. When
227 // the store identity changes (or there's no prior
228 // output), treat every key as new.
229 const priorEntries =
230 output && output.store === news.store
231 ? resolveEntries(output.entries)
232 : undefined;
233
234 // Sync entries — push the changed keys.
235 yield* upload(
236 news.store,
237 news.namespace,
238 entries,
239 priorEntries,
240 );
241
242 // Sync stale keys — when `purge` is set, list every key
243 // under the namespace and delete anything not in the
244 // desired set. This converges adopted/drifted state.
245 if (news.purge) {
246 yield* purge(news.store, news.namespace, entries);
247 }
248
249 return {
250 store: news.store,
251 namespace: news.namespace,
252 entries,
253 };
254 }),
255 );
256 }),
257 ),
258 delete: withKvsRegionFn(
259 Effect.fn(function* ({ output }) {
260 if (!output.store) return;
261 yield* retryForKvsReadiness(
262 purge(output.store, output.namespace, undefined),
263 ).pipe(
264 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
265 );
266 }),
267 ),
268 };
269 }),
270 );