1/**
2 * Internal helpers shared by the Kubernetes workload platforms
3 * (`Kubernetes.Deployment`, `Kubernetes.Job`): image-source dispatch
4 * through the cluster adapter, binding-env collection, connection
5 * identity, and the platform-neutral generated container entries.
6 */
7import * as Effect from "effect/Effect";
8import * as FileSystem from "effect/FileSystem";
9import * as Path from "effect/Path";
10import { hashDirectory } from "../../Command/Memo.ts";
11import { isInlineDockerfile } from "../../Docker/Dockerfile.ts";
12import type { ResourceBinding } from "../../Resource.ts";
13import { Self } from "../../Self.ts";
14import { sha256Object } from "../../Util/sha256.ts";
15import type {
16 ClusterAdapterService,
17 IdentityState,
18 RegistryState,
19 WorkloadBindingContract,
20 WorkloadImageSource,
21} from "../ClusterAdapter.ts";
22import type { ClusterLike, Connection, ConnectionAuth } from "../Connection.ts";
23import type { ConnectionRegistry } from "./registry.ts";
24
25/**
26 * Structural deep merge: objects merge recursively; arrays and primitives
27 * from `override` replace the base value wholesale. Powers escape hatches
28 * like `Kubernetes.Deployment.podTemplate`.
29 */
30export const deepMerge = <T>(base: T, override: unknown): T => {
31 if (override === undefined) return base;
32 if (
33 base === null ||
34 override === null ||
35 typeof base !== "object" ||
36 typeof override !== "object" ||
37 Array.isArray(base) ||
38 Array.isArray(override)
39 ) {
40 return override as T;
41 }
42 const out: Record<string, unknown> = {
43 ...(base as Record<string, unknown>),
44 };
45 for (const [key, value] of Object.entries(override)) {
46 out[key] =
47 key in (base as Record<string, unknown>)
48 ? deepMerge((base as Record<string, unknown>)[key], value)
49 : value;
50 }
51 return out as T;
52};
53
54/**
55 * The image build platform of a workload: its own `architecture`, else the
56 * connection's node architecture, else `linux/amd64`.
57 */
58export const imagePlatformOf = (
59 architecture: "amd64" | "arm64" | undefined,
60 connection?: Connection | undefined,
61): string =>
62 (architecture ?? connection?.architecture) === "arm64"
63 ? "linux/arm64"
64 : "linux/amd64";
65
66/**
67 * Best-effort {@link Connection} of a `cluster` prop value — `undefined`
68 * instead of throwing, for plan-time diffs where the referenced resource
69 * may resolve to stables-only (or `{}`).
70 */
71export const tryConnectionOf = (
72 cluster: ClusterLike | undefined,
73): Connection | undefined => {
74 if (cluster === undefined) return undefined;
75 if ("auth" in cluster && cluster.auth !== undefined) return cluster;
76 if ("connection" in cluster && cluster.connection !== undefined) {
77 return cluster.connection;
78 }
79 return undefined;
80};
81
82const sortKeysDeep = (value: unknown): unknown => {
83 if (Array.isArray(value)) return value.map(sortKeysDeep);
84 if (value === null || typeof value !== "object") return value;
85 return Object.fromEntries(
86 Object.entries(value as Record<string, unknown>)
87 .filter(([, entry]) => entry !== undefined)
88 .sort(([a], [b]) => a.localeCompare(b))
89 .map(([key, entry]) => [key, sortKeysDeep(entry)]),
90 );
91};
92
93/**
94 * A stable identity string for a connection's *target cluster* — the auth
95 * descriptor (which names the cluster for managed clouds and the
96 * kubeconfig context otherwise). Changing it means the workload moves
97 * clusters, which is a replacement. Deliberately excludes `endpoint` /
98 * CA: managed clusters can rotate those in place.
99 */
100export const connectionIdentity = (
101 connection: Connection | undefined,
102): string | undefined =>
103 connection === undefined
104 ? undefined
105 : JSON.stringify(sortKeysDeep(connection.auth));
106
107/**
108 * The persisted connection of a workload's attributes, tolerating legacy
109 * pre-rename rows (`AWS.EKS.*` attributes carried a flat `clusterName`
110 * instead of a `connection`) by synthesizing an `aws-eks` connection —
111 * the only platform those legacy types could target.
112 */
113export const connectionOfOutput = (
114 output: Record<string, unknown>,
115): Connection | undefined => {
116 const connection = output.connection as Connection | undefined;
117 if (connection?.auth !== undefined) return connection;
118 if (typeof output.clusterName === "string") {
119 return {
120 auth: {
121 kind: "aws-eks",
122 clusterName: output.clusterName,
123 } as unknown as ConnectionAuth,
124 };
125 }
126 return undefined;
127};
128
129/**
130 * Collect environment variables from a host's active bindings and report
131 * which cloud-grant channels (any binding-data key besides `env`, e.g.
132 * AWS `policyStatements`) are in play — the caller fails when grants
133 * exist but the cluster has no identity adapter to materialize them.
134 */
135export const collectBindingEnv = (
136 bindings: ResourceBinding<WorkloadBindingContract>[],
137): { env: Record<string, any>; grantKeys: string[] } => {
138 const activeBindings = bindings.filter(
139 (
140 binding: ResourceBinding<WorkloadBindingContract> & {
141 action?: string;
142 },
143 ) => binding.action !== "delete",
144 );
145
146 const env = activeBindings
147 .map((binding) => binding?.data?.env)
148 .reduce<Record<string, any>>((acc, value) => ({ ...acc, ...value }), {});
149
150 const grantKeys = [
151 ...new Set(
152 activeBindings.flatMap((binding) =>
153 Object.keys(binding?.data ?? {}).filter(
154 (key) =>
155 key !== "env" &&
156 (binding.data as Record<string, unknown>)[key] !== undefined,
157 ),
158 ),
159 ),
160 ];
161
162 return { env, grantKeys };
163};
164
165export type ImageSourceKind = "main" | "context" | "image";
166
167/** Which image source a props bag declares (`main` always wins). */
168export const imageSourceKind = (
169 source: WorkloadImageSource,
170): ImageSourceKind | undefined =>
171 source.main !== undefined
172 ? "main"
173 : source.image !== undefined
174 ? "image"
175 : source.context !== undefined || source.dockerfile !== undefined
176 ? "context"
177 : undefined;
178
179/**
180 * Content hash for image sources whose identity is computable without a
181 * bundler: `image` (the ref + platform) and `context` (the build-context
182 * directory + Dockerfile content + platform). `main` returns `undefined`
183 * — its hash comes from the bundle output inside the registry adapter.
184 */
185export const computeStaticWorkloadImageHash = Effect.fn(function* (
186 source: WorkloadImageSource,
187 platform: string,
188) {
189 const kind = imageSourceKind(source);
190 if (kind === "image") {
191 return (yield* sha256Object({ image: source.image!, platform })).slice(
192 0,
193 16,
194 );
195 }
196 if (kind === "context") {
197 if (
198 source.dockerfile !== undefined &&
199 isInlineDockerfile(source.dockerfile)
200 ) {
201 if (typeof source.dockerfile.content !== "string") return undefined;
202 return (yield* sha256Object({
203 dockerfile: source.dockerfile.content,
204 platform,
205 })).slice(0, 16);
206 }
207 const fs = yield* FileSystem.FileSystem;
208 const path = yield* Path.Path;
209 const context = path.resolve(source.context ?? ".");
210 const dockerfile =
211 typeof source.dockerfile === "string"
212 ? path.resolve(source.dockerfile)
213 : path.join(context, "Dockerfile");
214 if (!(yield* fs.exists(context)) || !(yield* fs.exists(dockerfile))) {
215 return undefined;
216 }
217 const contextHash = yield* hashDirectory({ cwd: context });
218 const dockerfileContent = yield* fs.readFileString(dockerfile);
219 return (yield* sha256Object({
220 contextHash,
221 dockerfile: dockerfileContent,
222 platform,
223 })).slice(0, 16);
224 }
225 return undefined;
226});
227
228export interface ResolveWorkloadImageOptions {
229 adapter: ClusterAdapterService;
230 /** The target connection — its `registry`, if any, receives builds. */
231 connection: Connection;
232 /** Publisher for connection-level registries. */
233 connectionRegistry: ConnectionRegistry;
234 id: string;
235 source: WorkloadImageSource;
236 platform: string;
237 port?: number | undefined;
238 isExternal?: boolean | undefined;
239 bootstrap: (importPath: string) => string;
240 tags: Record<string, string>;
241 /** Persisted registry state hints (legacy-shape tolerant). */
242 state: Record<string, unknown> | undefined;
243 session: { note: (message: string) => Effect.Effect<void> };
244}
245
246/**
247 * Resolve the container image for a workload: through the cluster
248 * adapter's managed registry when it has one (build/mirror + push). Other
249 * clusters run a pre-built `image` reference verbatim and build `main` /
250 * `context` sources into the connection's `registry`, which those sources
251 * require.
252 */
253export const resolveWorkloadImage = Effect.fn(function* (
254 options: ResolveWorkloadImageOptions,
255) {
256 const { adapter, source } = options;
257 if (adapter.registry !== undefined) {
258 return yield* adapter.registry.resolve({
259 id: options.id,
260 source,
261 platform: options.platform,
262 port: options.port,
263 isExternal: options.isExternal,
264 bootstrap: options.bootstrap,
265 tags: options.tags,
266 state: options.state,
267 session: options.session,
268 });
269 }
270
271 const kind = imageSourceKind(source);
272 if (kind === "image") {
273 const codeHash = (yield* computeStaticWorkloadImageHash(
274 source,
275 options.platform,
276 ))!;
277 return {
278 imageUri: source.image!,
279 codeHash,
280 state: undefined,
281 };
282 }
283
284 const registry = options.connection.registry;
285 if (registry !== undefined) {
286 return yield* options.connectionRegistry.resolve({
287 id: options.id,
288 registry,
289 source,
290 platform: options.platform,
291 port: options.port,
292 isExternal: options.isExternal,
293 bootstrap: options.bootstrap,
294 state: options.state,
295 session: options.session,
296 });
297 }
298
299 return yield* Effect.die(
300 new Error(
301 `'${options.id}': 'main' and 'context' image sources are built and ` +
302 "pushed to a container registry, and this cluster connection has " +
303 "none. Add a `registry` to the connection (e.g. " +
304 '`Kubernetes.KubeConfig({ context, registry: { server: "ghcr.io/acme" } })`), ' +
305 "use `Kubernetes.LocalCluster`, which includes one, or run a " +
306 "pre-built `image` reference the cluster can pull.",
307 ),
308 );
309});
310
311/** Plan-time content hash for `diff` — adapter-aware. */
312export const workloadImageHash = Effect.fn(function* (options: {
313 adapter: ClusterAdapterService;
314 connection: Connection;
315 connectionRegistry: ConnectionRegistry;
316 source: WorkloadImageSource;
317 platform: string;
318 port?: number | undefined;
319 isExternal?: boolean | undefined;
320 bootstrap: (importPath: string) => string;
321}) {
322 if (options.adapter.registry !== undefined) {
323 return yield* options.adapter.registry.hash({
324 source: options.source,
325 platform: options.platform,
326 port: options.port,
327 isExternal: options.isExternal,
328 bootstrap: options.bootstrap,
329 });
330 }
331 if (
332 options.connection.registry !== undefined &&
333 imageSourceKind(options.source) === "main"
334 ) {
335 return yield* options.connectionRegistry.hash({
336 source: options.source,
337 platform: options.platform,
338 port: options.port,
339 isExternal: options.isExternal,
340 bootstrap: options.bootstrap,
341 });
342 }
343 return yield* computeStaticWorkloadImageHash(
344 options.source,
345 options.platform,
346 );
347});
348
349/** The identity-adapter state persisted on workload attributes. */
350export type PersistedIdentityState = IdentityState | undefined;
351
352/**
353 * Platform-neutral generated container entry for an Effect-native server
354 * workload: resolves the program's registered runners and serves the
355 * returned `{ fetch }` handler on `PORT`. Cloud adapters override this
356 * through {@link ClusterAdapterService.bootstrap} to wire their runtime
357 * credential chains (e.g. EKS Pod Identity).
358 */
359export const makeServerBootstrap =
360 (handler: string) =>
361 (importPath: string): string =>
362 `
363import { BunServices } from "@effect/platform-bun";
364import { BunHttpServer } from "alchemy/Http";
365import { Stack } from "alchemy/Stack";
366import { Stage } from "alchemy/Stage";
367import { makeEntrypointLayer, reifyBoundConfigProvider } from "alchemy/Runtime";
368import { provideProcessTelemetry } from "alchemy/Telemetry";
369import * as Context from "effect/Context";
370import * as Config from "effect/Config";
371import * as ConfigProvider from "effect/ConfigProvider";
372import * as Effect from "effect/Effect";
373import * as FetchHttpClient from "effect/http/FetchHttpClient";
374import * as Layer from "effect/Layer";
375import * as Logger from "effect/Logger";
376
377import { ${handler} as entrypoint } from ${JSON.stringify(importPath)};
378
379// Normalize the entrypoint export: an inline-effect class default export is
380// an Effect resolving the platform instance, while the tagged form
381// (X.make(props, impl)) exports a Layer providing the Self tag. Both fold
382// into a Layer via makeEntrypointLayer (same pattern as the ECS/Lambda/
383// Cloudflare Container bridges).
384const tag = Context.Service("${Self.key}");
385const layer = makeEntrypointLayer(tag, entrypoint);
386
387const platform = Layer.mergeAll(
388 BunServices.layer,
389 FetchHttpClient.layer,
390 Logger.layer([Logger.consolePretty()]),
391);
392
393// Resolve the bundled program (the runners registered via host.run / serve)
394// and run it with a Bun HTTP server bound to PORT, so a returned { fetch }
395// handler is served and host.run loops stay alive.
396const program = tag.pipe(
397 // Process-lifetime telemetry: built once into the root scope; exporters
398 // batch on their intervals and flush when the scope closes on graceful
399 // shutdown.
400 Effect.flatMap((host) =>
401 host.RuntimeContext.exports.pipe(
402 Effect.flatMap((exports) => exports.program),
403 provideProcessTelemetry(host.RuntimeContext),
404 ),
405 ),
406 Effect.provide(
407 layer.pipe(Layer.provideMerge(Layer.mergeAll(
408 Layer.effect(
409 Stack,
410 Effect.all([
411 Config.String("ALCHEMY_STACK_NAME"),
412 Config.String("ALCHEMY_STAGE")
413 ]).pipe(
414 Effect.map(([name, stage]) => ({
415 name,
416 stage,
417 bindings: {},
418 resources: {}
419 }))
420 )
421 ),
422 // Module-scope declarations shared with the Stack may read the stage.
423 Layer.effect(Stage, Config.String("ALCHEMY_STAGE")),
424 )),
425 Layer.provideMerge(BunHttpServer()),
426 Layer.provideMerge(platform),
427 Layer.provideMerge(
428 Layer.succeed(
429 ConfigProvider.ConfigProvider,
430 reifyBoundConfigProvider(ConfigProvider.fromEnv(), process.env)
431 )
432 ),
433 )
434 ),
435 Effect.scoped
436);
437
438console.log(\`Kubernetes Deployment bootstrap starting on port \${process.env.PORT ?? 3000}...\`);
439await Effect.runPromise(program).catch((err) => {
440 console.error("Kubernetes Deployment bootstrap failed:", err);
441 process.exit(1);
442});
443`;
444
445/**
446 * Platform-neutral generated container entry for an Effect-native one-shot
447 * Job: resolves the program's `run` effect, executes it to completion, and
448 * exits.
449 */
450export const makeJobBootstrap =
451 (handler: string) =>
452 (importPath: string): string =>
453 `
454import { BunServices } from "@effect/platform-bun";
455import { Stack } from "alchemy/Stack";
456import { Stage } from "alchemy/Stage";
457import { makeEntrypointLayer, reifyBoundConfigProvider } from "alchemy/Runtime";
458import { provideProcessTelemetry } from "alchemy/Telemetry";
459import * as Context from "effect/Context";
460import * as Config from "effect/Config";
461import * as ConfigProvider from "effect/ConfigProvider";
462import * as Effect from "effect/Effect";
463import * as FetchHttpClient from "effect/http/FetchHttpClient";
464import * as Layer from "effect/Layer";
465import * as Logger from "effect/Logger";
466
467import { ${handler} as entrypoint } from ${JSON.stringify(importPath)};
468
469// Normalize the entrypoint export (see the server bootstrap).
470const tag = Context.Service("${Self.key}");
471const layer = makeEntrypointLayer(tag, entrypoint);
472
473const platform = Layer.mergeAll(
474 BunServices.layer,
475 FetchHttpClient.layer,
476 Logger.layer([Logger.consolePretty()]),
477);
478
479// Resolve the bundled program's registered one-shot runners (the shape's
480// \`run\` effect and any host.run work) and execute them to completion.
481const program = tag.pipe(
482 // Process-lifetime telemetry: built once into the root scope; exporters
483 // batch on their intervals and flush when the scope closes as the
484 // one-shot program completes.
485 Effect.flatMap((host) =>
486 host.RuntimeContext.exports.pipe(
487 Effect.flatMap((exports) => exports.program),
488 provideProcessTelemetry(host.RuntimeContext),
489 ),
490 ),
491 Effect.provide(
492 layer.pipe(Layer.provideMerge(Layer.mergeAll(
493 Layer.effect(
494 Stack,
495 Effect.all([
496 Config.String("ALCHEMY_STACK_NAME"),
497 Config.String("ALCHEMY_STAGE")
498 ]).pipe(
499 Effect.map(([name, stage]) => ({
500 name,
501 stage,
502 bindings: {},
503 resources: {}
504 }))
505 )
506 ),
507 // Module-scope declarations shared with the Stack may read the stage.
508 Layer.effect(Stage, Config.String("ALCHEMY_STAGE")),
509 )),
510 Layer.provideMerge(platform),
511 Layer.provideMerge(
512 Layer.succeed(
513 ConfigProvider.ConfigProvider,
514 reifyBoundConfigProvider(ConfigProvider.fromEnv(), process.env)
515 )
516 ),
517 )
518 ),
519 Effect.scoped
520);
521
522console.log("Kubernetes Job bootstrap starting...");
523await Effect.runPromise(program).catch((err) => {
524 console.error("Kubernetes Job bootstrap failed:", err);
525 process.exit(1);
526});
527// Run-to-completion semantics: exit 0 explicitly so lingering handles
528// (sockets, timers) never keep the pod alive after the work is done.
529process.exit(0);
530`;