1import * as analytics from "@distilled.cloud/aws/kinesis-analytics-v2";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import type { Providers } from "../Providers.ts";
10
11export type SnapshotStatus = analytics.SnapshotStatus;
12
13export interface ApplicationSnapshotProps {
14 /**
15 * Name of the Managed Service for Apache Flink application to snapshot.
16 * The application must be `RUNNING` with snapshots enabled. Changing the
17 * application replaces the snapshot.
18 */
19 applicationName: string;
20 /**
21 * Name of the snapshot. Changing the name replaces the snapshot.
22 * @default ${app}-${id}-${stage}-${instanceId}
23 */
24 snapshotName?: string;
25}
26
27export interface ApplicationSnapshot extends Resource<
28 "AWS.KinesisAnalyticsV2.ApplicationSnapshot",
29 ApplicationSnapshotProps,
30 {
31 /**
32 * Name of the application the snapshot belongs to.
33 */
34 applicationName: string;
35 /**
36 * Physical name of the snapshot.
37 */
38 snapshotName: string;
39 /**
40 * Current status of the snapshot.
41 */
42 snapshotStatus: SnapshotStatus;
43 /**
44 * Application version the snapshot was taken from.
45 */
46 applicationVersionId: number;
47 },
48 never,
49 Providers
50> {}
51
52/**
53 * A snapshot (Flink savepoint) of a running Managed Service for Apache
54 * Flink application's state.
55 *
56 * Snapshots are immutable — every prop change replaces the snapshot. The
57 * source application must be `RUNNING` with `snapshotsEnabled: true` when
58 * the snapshot is created.
59 * ### Creating Snapshots
60 * **Example:** Snapshot a running application
61 * ```typescript
62 * import * as AWS from "alchemy/AWS";
63 *
64 * const app = yield* AWS.KinesisAnalyticsV2.Application("Enrichment", {
65 * runtimeEnvironment: "FLINK-1_20",
66 * code: { bucketArn: bucket.bucketArn, fileKey: "jobs/enrichment-1.0.jar" },
67 * snapshotsEnabled: true,
68 * start: true,
69 * });
70 * const snapshot = yield* AWS.KinesisAnalyticsV2.ApplicationSnapshot(
71 * "Checkpoint",
72 * { applicationName: app.applicationName },
73 * );
74 * ```
75 *
76 * @resource
77 */
78export const ApplicationSnapshot = Resource<ApplicationSnapshot>(
79 "AWS.KinesisAnalyticsV2.ApplicationSnapshot",
80);
81
82/**
83 * The snapshot entered `FAILED` (or vanished) while waiting for `READY`.
84 */
85export class SnapshotFailed extends Data.TaggedError("SnapshotFailed")<{
86 readonly applicationName: string;
87 readonly snapshotName: string;
88 readonly status: string;
89}> {}
90
91class SnapshotPending extends Data.TaggedError("SnapshotPending")<{
92 readonly status: string;
93}> {}
94
95class SnapshotStillExists extends Data.TaggedError("SnapshotStillExists") {}
96
97const createSnapshotName = (
98 id: string,
99 props: { snapshotName?: string | undefined },
100) =>
101 Effect.gen(function* () {
102 if (props.snapshotName) {
103 return props.snapshotName;
104 }
105 return yield* createPhysicalName({ id, maxLength: 256 });
106 });
107
108/**
109 * Retries `ResourceInUseException` (the application is transitioning, e.g.
110 * an update settling before the snapshot can start) on a bounded schedule.
111 * Explicit return annotation for the declaration-emit reason documented in
112 * `Application.ts`.
113 */
114const retryWhileInUse = <A, E extends { _tag: string }, R>(
115 self: Effect.Effect<A, E, R>,
116): Effect.Effect<A, E, R> =>
117 Effect.retry(self, {
118 while: (e) => e._tag === "ResourceInUseException",
119 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(30)]),
120 });
121
122const describeSnapshot = Effect.fn(function* ({
123 applicationName,
124 snapshotName,
125}: {
126 applicationName: string;
127 snapshotName: string;
128}) {
129 const response = yield* analytics
130 .describeApplicationSnapshot({
131 ApplicationName: applicationName,
132 SnapshotName: snapshotName,
133 })
134 .pipe(
135 Effect.catchTag("ResourceNotFoundException", () =>
136 Effect.succeed(undefined),
137 ),
138 );
139 return response?.SnapshotDetails;
140});
141
142// Snapshot creation is proportional to state size; small test jobs finish
143// in seconds. 5s × 36 bounds the wait at 180s.
144const waitForSnapshotReady = ({
145 applicationName,
146 snapshotName,
147}: {
148 applicationName: string;
149 snapshotName: string;
150}) =>
151 Effect.gen(function* () {
152 const details = yield* describeSnapshot({ applicationName, snapshotName });
153 const status = details?.SnapshotStatus ?? "MISSING";
154 if (status === "READY") {
155 return details!;
156 }
157 if (status === "CREATING") {
158 return yield* Effect.fail(new SnapshotPending({ status }));
159 }
160 return yield* Effect.fail(
161 new SnapshotFailed({ applicationName, snapshotName, status }),
162 );
163 }).pipe(
164 Effect.retry({
165 while: (e: { _tag: string }) => e._tag === "SnapshotPending",
166 schedule: Schedule.max([
167 Schedule.fixed("5 seconds"),
168 Schedule.recurs(36),
169 ]),
170 }),
171 Effect.catchTag("SnapshotPending", (e) =>
172 Effect.fail(
173 new SnapshotFailed({ applicationName, snapshotName, status: e.status }),
174 ),
175 ),
176 );
177
178const waitForSnapshotDeleted = ({
179 applicationName,
180 snapshotName,
181}: {
182 applicationName: string;
183 snapshotName: string;
184}) =>
185 Effect.gen(function* () {
186 const details = yield* describeSnapshot({ applicationName, snapshotName });
187 if (details !== undefined) {
188 return yield* Effect.fail(new SnapshotStillExists());
189 }
190 }).pipe(
191 Effect.retry({
192 while: (e: { _tag: string }) => e._tag === "SnapshotStillExists",
193 schedule: Schedule.max([
194 Schedule.fixed("3 seconds"),
195 Schedule.recurs(30),
196 ]),
197 }),
198 );
199
200export const ApplicationSnapshotProvider = () =>
201 Provider.effect(
202 ApplicationSnapshot,
203 Effect.gen(function* () {
204 return ApplicationSnapshot.Provider.of({
205 stables: ["applicationName", "snapshotName", "applicationVersionId"],
206
207 // Sub-resource keyed by its parent application — snapshots are not
208 // enumerable account-wide, so `list` is empty and refresh flows
209 // through `read` on known instances.
210 list: () => Effect.succeed([]),
211
212 read: Effect.fn(function* ({ id, olds, output }) {
213 const applicationName =
214 output?.applicationName ?? olds?.applicationName;
215 if (!applicationName) return undefined;
216 const snapshotName =
217 output?.snapshotName ?? (yield* createSnapshotName(id, olds ?? {}));
218 const details = yield* describeSnapshot({
219 applicationName,
220 snapshotName,
221 });
222 if (!details) return undefined;
223 return {
224 applicationName,
225 snapshotName,
226 snapshotStatus: details.SnapshotStatus,
227 applicationVersionId: details.ApplicationVersionId,
228 };
229 }),
230
231 diff: Effect.fn(function* ({ id, news, olds }) {
232 if (!isResolved(news)) return undefined;
233 // Snapshots are immutable — any identity change replaces.
234 const oldName = yield* createSnapshotName(id, olds ?? {});
235 const newName = yield* createSnapshotName(id, news ?? {});
236 if (
237 olds?.applicationName !== news?.applicationName ||
238 oldName !== newName
239 ) {
240 return { action: "replace" } as const;
241 }
242 }),
243
244 // Existence-only resource: observe → if missing, create → wait for
245 // READY. Nothing about a snapshot is mutable.
246 reconcile: Effect.fn(function* ({ id, news, output, session }) {
247 const applicationName = news.applicationName;
248 const snapshotName =
249 output?.snapshotName ?? (yield* createSnapshotName(id, news));
250
251 const observed = yield* describeSnapshot({
252 applicationName,
253 snapshotName,
254 });
255 if (observed === undefined) {
256 yield* retryWhileInUse(
257 analytics.createApplicationSnapshot({
258 ApplicationName: applicationName,
259 SnapshotName: snapshotName,
260 }),
261 );
262 yield* session.note(`Creating snapshot ${snapshotName}...`);
263 }
264 const ready = yield* waitForSnapshotReady({
265 applicationName,
266 snapshotName,
267 });
268
269 yield* session.note(snapshotName);
270 return {
271 applicationName,
272 snapshotName,
273 snapshotStatus: ready.SnapshotStatus,
274 applicationVersionId: ready.ApplicationVersionId,
275 };
276 }),
277
278 delete: Effect.fn(function* ({ output }) {
279 // DeleteApplicationSnapshot requires the snapshot's creation
280 // timestamp — observe it fresh. A missing snapshot (or a missing
281 // application) means the delete already happened.
282 const details = yield* describeSnapshot({
283 applicationName: output.applicationName,
284 snapshotName: output.snapshotName,
285 });
286 if (!details?.SnapshotCreationTimestamp) {
287 return;
288 }
289 yield* retryWhileInUse(
290 analytics
291 .deleteApplicationSnapshot({
292 ApplicationName: output.applicationName,
293 SnapshotName: output.snapshotName,
294 SnapshotCreationTimestamp: details.SnapshotCreationTimestamp,
295 })
296 .pipe(
297 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
298 ),
299 );
300 yield* waitForSnapshotDeleted({
301 applicationName: output.applicationName,
302 snapshotName: output.snapshotName,
303 });
304 }),
305 });
306 }),
307 );