1import type { Volume as FlyVolume } from "@distilled.cloud/fly-io/machines";
2import * as machines from "@distilled.cloud/fly-io/machines";
3import * as Data from "effect/Data";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Schedule from "effect/Schedule";
7import { listOwnedApps } from "./App.ts";
8import {
9 createFlyVolumeName,
10 matchesAlchemyPhysicalName,
11 sanitizeFlyVolumeName,
12} from "./Metadata.ts";
13import type { DiskSpec } from "./MountVolume.ts";
14
15export const DEFAULT_VOLUME_REGION = "iad";
16export const MIN_SIZE_GB = 1;
17
18export class VolumeNotCreated extends Data.TaggedError("Fly.VolumeNotCreated")<{
19 name: string;
20 appName: string;
21}> {}
22
23class VolumePending extends Data.TaggedError("Fly.VolumePending")<{
24 volumeId: string;
25 state: string;
26}> {}
27
28const backoff = Schedule.min([
29 Schedule.exponential(Duration.millis(500), 1.5),
30 Schedule.spaced(Duration.seconds(5)),
31]);
32
33export const destroying = (state: string | undefined) =>
34 state === "destroyed" ||
35 state === "pending_destroy" ||
36 state === "scheduled_for_destruction";
37
38const transientState = (state: string | undefined) =>
39 state === "creating" || state === "pending" || state === "extending";
40
41export const getVolumeById = (appName: string, volumeId: string) =>
42 machines.getVolumeById({ app_name: appName, volume_id: volumeId }).pipe(
43 Effect.map((volume) => (destroying(volume.state) ? undefined : volume)),
44 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
45 );
46
47export const listVolumesByApp = (appName: string) =>
48 machines.listVolumes({ app_name: appName }).pipe(
49 Effect.map((volumes) =>
50 volumes.filter((volume) => !destroying(volume.state)),
51 ),
52 Effect.catchTag(["NotFound", "Forbidden"], () => Effect.succeed([])),
53 );
54
55const newestFirst = (left: FlyVolume, right: FlyVolume) => {
56 const byCreated =
57 Date.parse(left.created_at ?? "") - Date.parse(right.created_at ?? "");
58 if (byCreated !== 0) return byCreated;
59 return (left.id ?? "").localeCompare(right.id ?? "");
60};
61
62export const listVolumeGroup = (
63 appName: string,
64 name: string,
65 region: string,
66) =>
67 listVolumesByApp(appName).pipe(
68 Effect.map((volumes) =>
69 volumes
70 .filter((volume) => volume.name === name && volume.region === region)
71 .sort(newestFirst),
72 ),
73 );
74
75export const waitUntilVolumeReady = (appName: string, volumeId: string) =>
76 getVolumeById(appName, volumeId).pipe(
77 Effect.flatMap((volume) => {
78 if (volume === undefined) return Effect.succeed(undefined);
79 if (transientState(volume.state)) {
80 return Effect.fail(
81 new VolumePending({
82 volumeId,
83 state: volume.state ?? "creating",
84 }),
85 );
86 }
87 return Effect.succeed(volume);
88 }),
89 Effect.retry({
90 while: (e) => e._tag === "Fly.VolumePending",
91 times: 8,
92 schedule: backoff,
93 }),
94 Effect.catchTag("Fly.VolumePending", () =>
95 getVolumeById(appName, volumeId),
96 ),
97 );
98
99export const waitUntilVolumeGone = (appName: string, volumeId: string) =>
100 getVolumeById(appName, volumeId).pipe(
101 Effect.map((volume) => volume === undefined),
102 Effect.repeat({
103 schedule: Schedule.spaced("2 seconds"),
104 until: (gone) => gone,
105 times: 10,
106 }),
107 );
108
109export const pathKey = (path: string): string => {
110 const key = path
111 .toLowerCase()
112 .replace(/[^a-z0-9]+/g, "_")
113 .replace(/^_+|_+$/g, "");
114 return key.length === 0 ? "data" : key;
115};
116
117export const volumeGroupName = (
118 id: string,
119 disk: Pick<DiskSpec, "path" | "name">,
120) =>
121 Effect.gen(function* () {
122 if (disk.name !== undefined) return sanitizeFlyVolumeName(disk.name);
123 return yield* createFlyVolumeName(`${id}/${pathKey(disk.path)}`);
124 });
125
126export const createVolume = Effect.fn(function* (input: {
127 appName: string;
128 name: string;
129 region: string;
130 disk: DiskSpec;
131}) {
132 const sizeGb = Math.max(input.disk.sizeGb, MIN_SIZE_GB);
133 const created = yield* machines
134 .createVolume({
135 app_name: input.appName,
136 name: input.name,
137 region: input.region,
138 size_gb: sizeGb,
139 encrypted: input.disk.encrypted,
140 fstype: input.disk.fstype,
141 auto_backup_enabled: input.disk.autoBackupEnabled,
142 snapshot_retention: input.disk.snapshotRetention,
143 snapshot_id: input.disk.snapshotId,
144 source_volume_id: input.disk.sourceVolumeId,
145 require_unique_zone: input.disk.requireUniqueZone,
146 })
147 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
148 if (created?.id !== undefined && created.id.length > 0) {
149 return (yield* waitUntilVolumeReady(input.appName, created.id)) ?? created;
150 }
151 const group = yield* listVolumeGroup(input.appName, input.name, input.region);
152 const hit = group.at(-1);
153 if (hit === undefined || hit.id === undefined) {
154 return yield* new VolumeNotCreated({
155 name: input.name,
156 appName: input.appName,
157 });
158 }
159 return (yield* waitUntilVolumeReady(input.appName, hit.id)) ?? hit;
160});
161
162export const syncVolume = Effect.fn(function* (
163 appName: string,
164 volume: FlyVolume,
165 disk: DiskSpec,
166) {
167 const volumeId = volume.id;
168 if (volumeId === undefined || volumeId.length === 0) return volume;
169 let current = volume;
170 const sizeGb = Math.max(disk.sizeGb, MIN_SIZE_GB);
171 const observedSize = current.size_gb ?? 0;
172 if (sizeGb > observedSize) {
173 const extended = yield* machines.extendVolume({
174 app_name: appName,
175 volume_id: volumeId,
176 size_gb: sizeGb,
177 });
178 current =
179 extended.volume ??
180 (yield* waitUntilVolumeReady(appName, volumeId)) ??
181 current;
182 }
183
184 const backupChanged =
185 disk.autoBackupEnabled !== undefined &&
186 disk.autoBackupEnabled !== current.auto_backup_enabled;
187 const retentionChanged =
188 disk.snapshotRetention !== undefined &&
189 disk.snapshotRetention !== current.snapshot_retention;
190 if (backupChanged || retentionChanged) {
191 current = yield* machines.updateVolume({
192 app_name: appName,
193 volume_id: volumeId,
194 auto_backup_enabled: disk.autoBackupEnabled,
195 snapshot_retention: disk.snapshotRetention,
196 });
197 }
198 return current;
199});
200
201/**
202 * Observe-ensure-sync a Fly volume group (`name` shared, `count`
203 * independent volumes). Extras are left in place for the caller to
204 * delete after the Machines that mount them are gone.
205 */
206export const ensureVolumeGroup = Effect.fn(function* (input: {
207 appName: string;
208 name: string;
209 region: string;
210 count: number;
211 disk: DiskSpec;
212 preferIds?: readonly string[];
213}) {
214 const prefer = new Set((input.preferIds ?? []).filter((id) => id.length > 0));
215 let group = yield* listVolumeGroup(input.appName, input.name, input.region);
216 const preferred = group.filter(
217 (volume) => volume.id !== undefined && prefer.has(volume.id),
218 );
219 const rest = group.filter(
220 (volume) => volume.id === undefined || !prefer.has(volume.id),
221 );
222 group = [...preferred, ...rest];
223
224 let attempts = 0;
225 while (group.length < input.count && attempts < input.count + 3) {
226 attempts += 1;
227 const created = yield* createVolume({
228 appName: input.appName,
229 name: input.name,
230 region: input.region,
231 disk: input.disk,
232 });
233 if (
234 created.id !== undefined &&
235 !group.some((volume) => volume.id === created.id)
236 ) {
237 group.push(created);
238 } else {
239 group = yield* listVolumeGroup(input.appName, input.name, input.region);
240 }
241 }
242 if (group.length < input.count) {
243 return yield* new VolumeNotCreated({
244 name: input.name,
245 appName: input.appName,
246 });
247 }
248
249 const kept = group.slice(0, input.count);
250 const extras = group.slice(input.count);
251 const synced: FlyVolume[] = [];
252 for (const volume of kept) {
253 synced.push(yield* syncVolume(input.appName, volume, input.disk));
254 }
255 return { volumes: synced, extras };
256});
257
258export const deleteVolume = Effect.fn(function* (
259 appName: string,
260 volumeId: string,
261) {
262 if (appName.length === 0 || volumeId.length === 0) return;
263 yield* machines
264 .deleteVolume({
265 app_name: appName,
266 volume_id: volumeId,
267 })
268 .pipe(
269 Effect.catchTag("NotFound", () => Effect.void),
270 Effect.retry({
271 while: (e) => e._tag === "Conflict",
272 times: 8,
273 schedule: backoff,
274 }),
275 );
276 yield* waitUntilVolumeGone(appName, volumeId);
277});
278
279export const listOwnedVolumes = Effect.fn(function* () {
280 const apps = yield* listOwnedApps();
281 const groups = yield* Effect.forEach(
282 apps,
283 (app) =>
284 listVolumesByApp(app.appName).pipe(
285 Effect.map((volumes) =>
286 volumes.flatMap((volume) => {
287 if (!matchesAlchemyPhysicalName(volume.name)) return [];
288 const volumeId = volume.id;
289 if (volumeId === undefined || volumeId.length === 0) return [];
290 return [{ appName: app.appName, volumeId, volume }];
291 }),
292 ),
293 ),
294 { concurrency: 8 },
295 );
296 return groups.flat();
297});