1import * as amp from "@distilled.cloud/aws/amp";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
11import { toWireDays } from "../../Util/Duration.ts";
12import type { Providers } from "../Providers.ts";
13import { syncAmpTags, toTagRecord } from "./internal.ts";
14
15export interface WorkspaceLabelSetLimit {
16 /**
17 * The label set the limit applies to, as exact label name/value pairs. An
18 * empty record (`{}`) is the default bucket — it applies to all time
19 * series that match no other label set entry.
20 */
21 labelSet: Record<string, string>;
22 /**
23 * Maximum number of active time series that can carry this label set.
24 * Omit to track the label set without enforcing a limit.
25 */
26 maxSeries?: number;
27}
28
29export interface WorkspaceProps {
30 /**
31 * A human-readable alias for the workspace. Aliases are not unique — many
32 * workspaces can share one. Updating the alias is an in-place update.
33 */
34 alias?: string;
35 /**
36 * ARN of a customer-managed KMS key used to encrypt data at rest. If
37 * omitted, an AWS-owned key is used. Changing the key replaces the
38 * workspace (encryption configuration is immutable).
39 */
40 kmsKeyArn?: string;
41 /**
42 * How long the workspace retains ingested metric data. Accepts any
43 * `Duration.Input` (e.g. `"30 days"`, `Duration.days(30)`; a bare number
44 * is milliseconds); the wire unit is whole days
45 * (`retentionPeriodInDays`). If omitted, the workspace keeps the service
46 * default retention (150 days) and any retention configured out-of-band
47 * is left untouched.
48 */
49 retentionPeriod?: Duration.Input;
50 /**
51 * Per-label-set ingestion limits (maximum active series per label set).
52 * If omitted, existing label-set limits are left untouched.
53 */
54 limitsPerLabelSet?: WorkspaceLabelSetLimit[];
55 /**
56 * User-defined tags for the workspace.
57 */
58 tags?: Record<string, string>;
59}
60
61export interface Workspace extends Resource<
62 "AWS.AMP.Workspace",
63 WorkspaceProps,
64 {
65 workspaceId: string;
66 workspaceArn: string;
67 prometheusEndpoint: string | undefined;
68 alias: string | undefined;
69 status: string;
70 },
71 never,
72 Providers
73> {}
74
75/**
76 * An Amazon Managed Service for Prometheus (AMP) workspace — a logical,
77 * fully-managed Prometheus-compatible metrics store. Metrics are ingested
78 * via remote-write and queried through the workspace's Prometheus-compatible
79 * endpoint.
80 *
81 * ### Creating a Workspace
82 * **Example:** Basic Workspace
83 * ```typescript
84 * const workspace = yield* AMP.Workspace("Metrics", {
85 * alias: "production-metrics",
86 * });
87 * ```
88 *
89 * **Example:** Workspace with Customer-Managed Encryption
90 * ```typescript
91 * const workspace = yield* AMP.Workspace("Metrics", {
92 * alias: "production-metrics",
93 * kmsKeyArn: key.keyArn,
94 * tags: { team: "observability" },
95 * });
96 * ```
97 *
98 * **Example:** Workspace with Custom Retention and Series Limits
99 * ```typescript
100 * const workspace = yield* AMP.Workspace("Metrics", {
101 * alias: "production-metrics",
102 * retentionPeriod: "30 days",
103 * limitsPerLabelSet: [
104 * { labelSet: { team: "billing" }, maxSeries: 100_000 },
105 * { labelSet: {}, maxSeries: 1_000_000 }, // default bucket
106 * ],
107 * });
108 * ```
109 *
110 * ### Using the Endpoint
111 * **Example:** Read the Remote-Write URL
112 * ```typescript
113 * // prometheusEndpoint ends in a trailing slash; append `api/v1/remote_write`
114 * const remoteWrite = `${workspace.prometheusEndpoint}api/v1/remote_write`;
115 * ```
116 *
117 * ### Runtime Bindings
118 * **Example:** Write and Query Metrics from a Function
119 * ```typescript
120 * // inside a Lambda Function's effect (provide the *Http layers):
121 * const remoteWrite = yield* AMP.RemoteWrite(workspace);
122 * const metrics = yield* AMP.QueryMetrics(workspace);
123 *
124 * yield* remoteWrite({
125 * timeseries: [{ name: "jobs_done_total", samples: [{ value: 1 }] }],
126 * });
127 * const result = yield* metrics.query({ query: "jobs_done_total" });
128 * ```
129 *
130 * @resource
131 */
132export const Workspace = Resource<Workspace>("AWS.AMP.Workspace");
133
134export const WorkspaceProvider = () =>
135 Provider.effect(
136 Workspace,
137 Effect.gen(function* () {
138 const toAttrs = (workspace: amp.WorkspaceDescription) => ({
139 workspaceId: workspace.workspaceId,
140 workspaceArn: workspace.arn,
141 prometheusEndpoint: workspace.prometheusEndpoint,
142 alias: workspace.alias,
143 status: workspace.status.statusCode,
144 });
145
146 /** Describe a workspace by id; typed not-found → undefined. */
147 const describe = Effect.fn(function* (workspaceId: string) {
148 const response = yield* amp
149 .describeWorkspace({ workspaceId })
150 .pipe(
151 Effect.catchTag("ResourceNotFoundException", () =>
152 Effect.succeed(undefined),
153 ),
154 );
155 return response?.workspace;
156 });
157
158 /**
159 * Find the workspace carrying this resource's alchemy ownership tags.
160 * Workspace ids are server-assigned, so when the persisted output is
161 * lost (e.g. a run crashed after create but before state persistence)
162 * this tag search is the only way to reclaim the existing workspace
163 * instead of creating an orphan-producing duplicate.
164 */
165 const findByInternalTags = Effect.fn(function* (
166 internalTags: Record<string, string>,
167 ) {
168 const pages = yield* amp.listWorkspaces
169 .pages({})
170 .pipe(Stream.runCollect);
171 const summary = Array.from(pages)
172 .flatMap((page) => page.workspaces)
173 .find(
174 (w) =>
175 w.status.statusCode !== "DELETING" &&
176 Object.entries(internalTags).every(
177 ([key, value]) => toTagRecord(w.tags)[key] === value,
178 ),
179 );
180 return summary === undefined
181 ? undefined
182 : yield* describe(summary.workspaceId);
183 });
184
185 /**
186 * Canonical form of a label-set-limits list for observed-vs-desired
187 * comparison: entries keyed and sorted by their sorted label set.
188 */
189 const canonicalLimits = (
190 limits: {
191 labelSet: Record<string, string | undefined>;
192 maxSeries: number | undefined;
193 }[],
194 ) =>
195 JSON.stringify(
196 limits
197 .map((entry) => ({
198 labelSet: Object.fromEntries(
199 Object.entries(entry.labelSet)
200 .filter(
201 (kv): kv is [string, string] => typeof kv[1] === "string",
202 )
203 .sort(([a], [b]) => a.localeCompare(b)),
204 ),
205 maxSeries: entry.maxSeries ?? null,
206 }))
207 .sort((a, b) =>
208 JSON.stringify(a.labelSet).localeCompare(
209 JSON.stringify(b.labelSet),
210 ),
211 ),
212 );
213
214 /**
215 * Sync the workspace configuration (retention period + label-set
216 * limits) — diff OBSERVED configuration against desired and apply
217 * only the delta. Skipped entirely when neither prop is set, so
218 * out-of-band configuration is never clobbered.
219 */
220 const syncWorkspaceConfiguration = Effect.fn(function* (
221 workspaceId: string,
222 news: WorkspaceProps,
223 ) {
224 if (
225 news.retentionPeriod === undefined &&
226 news.limitsPerLabelSet === undefined
227 ) {
228 return;
229 }
230 const observed = (yield* amp.describeWorkspaceConfiguration({
231 workspaceId,
232 })).workspaceConfiguration;
233
234 const desiredDays = toWireDays(news.retentionPeriod);
235 const retentionDrifts =
236 desiredDays !== undefined &&
237 desiredDays !== observed.retentionPeriodInDays;
238
239 const desiredLimits = news.limitsPerLabelSet?.map((entry) => ({
240 labelSet: entry.labelSet,
241 limits: { maxSeries: entry.maxSeries },
242 }));
243 const limitsDrift =
244 desiredLimits !== undefined &&
245 canonicalLimits(
246 desiredLimits.map((l) => ({
247 labelSet: l.labelSet,
248 maxSeries: l.limits.maxSeries,
249 })),
250 ) !==
251 canonicalLimits(
252 (observed.limitsPerLabelSet ?? []).map((l) => ({
253 labelSet: l.labelSet,
254 maxSeries: l.limits.maxSeries,
255 })),
256 );
257
258 if (!retentionDrifts && !limitsDrift) return;
259
260 // A previous configuration update may still be applying
261 // (`UPDATING` rejects new updates with a conflict) — retry briefly.
262 yield* amp
263 .updateWorkspaceConfiguration({
264 workspaceId,
265 retentionPeriodInDays: desiredDays,
266 limitsPerLabelSet: desiredLimits,
267 })
268 .pipe(
269 Effect.retry({
270 while: (e) => e._tag === "ConflictException",
271 schedule: Schedule.max([
272 Schedule.fixed("6 seconds"),
273 Schedule.recurs(15),
274 ]),
275 }),
276 );
277
278 // Bounded best-effort wait for the update to apply (the status
279 // returns to ACTIVE once the new configuration is in effect). A
280 // slower apply converges on a later reconcile.
281 yield* amp.describeWorkspaceConfiguration({ workspaceId }).pipe(
282 Effect.map((r) => r.workspaceConfiguration),
283 Effect.repeat({
284 schedule: Schedule.max([
285 Schedule.fixed("3 seconds"),
286 Schedule.recurs(30),
287 ]),
288 until: (c): boolean =>
289 c.status.statusCode === "ACTIVE" &&
290 (desiredDays === undefined ||
291 c.retentionPeriodInDays === desiredDays),
292 }),
293 );
294 });
295
296 /**
297 * Poll until the workspace leaves CREATING/UPDATING and reaches
298 * ACTIVE. Fails fast on a *_FAILED terminal status.
299 */
300 const waitActive = Effect.fn(function* (workspaceId: string) {
301 const workspace = yield* amp.describeWorkspace({ workspaceId }).pipe(
302 Effect.map((r) => r.workspace),
303 Effect.repeat({
304 schedule: Schedule.max([
305 Schedule.fixed("2 seconds"),
306 Schedule.recurs(30),
307 ]),
308 until: (w) => w.status.statusCode === "ACTIVE",
309 }),
310 );
311 if (workspace.status.statusCode !== "ACTIVE") {
312 return yield* Effect.fail(
313 new Error(
314 `AMP workspace ${workspaceId} did not become ACTIVE (status: ${workspace.status.statusCode})`,
315 ),
316 );
317 }
318 return workspace;
319 });
320
321 return {
322 stables: ["workspaceId", "workspaceArn", "prometheusEndpoint"],
323
324 diff: Effect.fn(function* ({ olds, news }) {
325 if (!isResolved(news)) return undefined;
326 // Encryption configuration is immutable — a change replaces.
327 if (
328 (olds?.kmsKeyArn ?? undefined) !== (news?.kmsKeyArn ?? undefined)
329 ) {
330 return { action: "replace" } as const;
331 }
332 }),
333
334 read: Effect.fn(function* ({ id, output }) {
335 // Workspace ids are server-assigned; without an output cache the
336 // only recoverable identity is the alchemy ownership tags.
337 if (!output?.workspaceId) {
338 const internalTags = yield* createInternalTags(id);
339 const found = yield* findByInternalTags(internalTags);
340 // A tag match is by definition owned by this resource.
341 return found === undefined ? undefined : toAttrs(found);
342 }
343 const workspace = yield* describe(output.workspaceId);
344 if (workspace === undefined) return undefined;
345 const attrs = toAttrs(workspace);
346 const tags = toTagRecord(workspace.tags);
347 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
348 }),
349
350 reconcile: Effect.fn(function* ({ id, news = {}, output, session }) {
351 const internalTags = yield* createInternalTags(id);
352 const desiredTags = { ...internalTags, ...news.tags };
353
354 // 1. Observe — cloud state is authoritative; output is an id cache.
355 // With no cached id (greenfield OR lost state after a crash), fall
356 // back to searching by ownership tags: the id is server-assigned,
357 // so tags are the only recoverable identity.
358 let workspace =
359 output?.workspaceId !== undefined
360 ? yield* describe(output.workspaceId)
361 : yield* findByInternalTags(internalTags);
362
363 // 2. Ensure — create if missing, then wait for ACTIVE.
364 if (workspace === undefined) {
365 const created = yield* amp.createWorkspace({
366 alias: news.alias,
367 kmsKeyArn: news.kmsKeyArn,
368 tags: desiredTags,
369 });
370 workspace = yield* waitActive(created.workspaceId);
371 }
372
373 const workspaceId = workspace.workspaceId;
374
375 // 3. Sync alias — in-place update when it drifts.
376 if ((news.alias ?? undefined) !== (workspace.alias ?? undefined)) {
377 yield* amp.updateWorkspaceAlias({
378 workspaceId,
379 alias: news.alias,
380 });
381 }
382
383 // 3b. Sync tags — diff against OBSERVED cloud tags.
384 yield* syncAmpTags(workspace.arn, desiredTags);
385
386 // 3c. Sync workspace configuration (retention + label-set limits).
387 yield* syncWorkspaceConfiguration(workspaceId, news);
388
389 // 4. Re-read for fresh attributes (alias/status may have changed).
390 const fresh = (yield* describe(workspaceId)) ?? workspace;
391 yield* session.note(workspaceId);
392 return toAttrs(fresh);
393 }),
394
395 delete: Effect.fn(function* ({ output }) {
396 yield* amp.deleteWorkspace({ workspaceId: output.workspaceId }).pipe(
397 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
398 // A workspace mid-transition rejects deletion; retry briefly.
399 Effect.retry({
400 while: (e) => e._tag === "ConflictException",
401 schedule: Schedule.max([
402 Schedule.fixed("3 seconds"),
403 Schedule.recurs(20),
404 ]),
405 }),
406 );
407 }),
408
409 list: () =>
410 amp.listWorkspaces.pages({}).pipe(
411 Stream.runCollect,
412 Effect.map((chunk) =>
413 Array.from(chunk).flatMap((page) => page.workspaces),
414 ),
415 Effect.flatMap(
416 Effect.forEach(
417 (summary) =>
418 describe(summary.workspaceId).pipe(
419 Effect.map((w) => (w ? toAttrs(w) : undefined)),
420 ),
421 { concurrency: 4 },
422 ),
423 ),
424 Effect.map((items) => items.filter((item) => item !== undefined)),
425 ),
426 };
427 }),
428 );