1import * as Effect from "effect/Effect";
2import * as Option from "effect/Option";
3import { HttpServer, type HttpEffect } from "../../Http.ts";
4import * as Output from "../../Output.ts";
5import { Platform } from "../../Platform.ts";
6import { serveRpc, type Rpc } from "../../Rpc.ts";
7import {
8 packEnvValueKeepRedacted,
9 unpackEnvValue,
10} from "../../RuntimeContext.ts";
11import type { ProcessContext } from "../../Server/Process.ts";
12import type { Fetcher } from "../Fetcher.ts";
13import { fromCloudflareFetcher, toCloudflareFetcher } from "../Fetcher.ts";
14import { DurableObject } from "../Workers/DurableObject.ts";
15import { DurableObjectState } from "../Workers/DurableObjectState.ts";
16import { Worker } from "../Workers/Worker.ts";
17import { ContainerTypeId } from "./Container.ts";
18import type {
19 ContainerApplication,
20 ContainerServices,
21 ContainerShape,
22} from "./ContainerApplication.ts";
23
24const toHttpUrl = (url: string) =>
25 url.startsWith("https:") ? `http:${url.slice("https:".length)}` : url;
26
27/**
28 * workerd container ports reject TLS outright ("Connecting to a container
29 * using HTTPS is not currently supported") — but the common proxy pattern
30 * forwards the incoming Worker request, whose production URL is `https://`,
31 * straight to the port. The hop into the container is already secure, so
32 * downgrade the scheme before forwarding, exactly like Cloudflare's own
33 * `@cloudflare/containers` `containerFetch` does
34 * (`request.url.replace("https:", "http:")`).
35 */
36const httpSchemePort = <
37 P extends {
38 fetch: (...args: any[]) => any;
39 connect: (...args: any[]) => any;
40 },
41>(
42 port: P,
43): P =>
44 ({
45 fetch: (input: RequestInfo | URL, init?: RequestInit) =>
46 input instanceof Request
47 ? port.fetch(toHttpUrl(input.url), input)
48 : port.fetch(toHttpUrl(String(input)), init),
49 connect: (address: any, options?: any) => port.connect(address, options),
50 }) as any as P;
51
52export const ContainerPlatform: Platform<
53 ContainerApplication,
54 ContainerServices,
55 ContainerShape,
56 ProcessContext,
57 Container
58> = Platform(
59 "Cloudflare.Container",
60 {
61 createRuntimeContext: (id: string): ProcessContext => {
62 const runners: Effect.Effect<void, never, any>[] = [];
63 const env: Record<string, any> = {};
64
65 const serve = <Req = never>(
66 handler: HttpEffect<Req>,
67 options?: { shape?: Record<string, unknown> },
68 ) =>
69 Effect.sync(() => {
70 // Containers have no native RPC transport (unlike a Cloudflare
71 // Worker's JSRPC), so expose the impl's non-`fetch` shape methods
72 // over the plain-`fetch` RPC protocol: requests to `/__rpc__/*` are
73 // dispatched to the matching shape method, everything else falls
74 // through to the user's `fetch` handler. The DO side talks to this
75 // via `makeFetchRpcStub` over the container's TCP port.
76 const finalHandler = options?.shape
77 ? serveRpc(options.shape, handler)
78 : handler;
79 runners.push(
80 Effect.gen(function* () {
81 const httpServer = yield* Effect.serviceOption(HttpServer).pipe(
82 Effect.map(Option.getOrUndefined),
83 );
84 if (httpServer) {
85 yield* httpServer.serve(finalHandler);
86 return yield* Effect.never;
87 } else {
88 // this should only happen at plantime, validate?
89 }
90 }).pipe(Effect.orDie),
91 );
92 });
93
94 return {
95 Type: ContainerTypeId,
96 LogicalId: id,
97 id,
98 env,
99 set: (bindingId: string, output: Output.Output) =>
100 Effect.sync(() => {
101 const key = bindingId.replaceAll(/[^a-zA-Z0-9]/g, "_");
102 // `packEnvValueKeepRedacted` keeps the Redacted wrapper on the
103 // outside so the provider can deploy secrets through the
104 // Secrets Store (referenced via `secrets`) instead of leaking
105 // them as plain `environmentVariables`, while the inner marker
106 // lets the runtime `get` accessor rebuild the wrapper after
107 // Cloudflare hands the value back as a plain env-var string.
108 // Mirrors `makeWorkerRuntimeContext`.
109 env[key] = output.pipe(Output.map(packEnvValueKeepRedacted));
110 return key;
111 }),
112 get: <T>(key: string) =>
113 // Read straight from `process.env` — see `unpackEnvValue` for why
114 // this must never resolve through `Config.String`.
115 Effect.sync(() => unpackEnvValue<T>(process.env[key]) as T),
116 run: ((effect: Effect.Effect<void, never, any>) =>
117 Effect.sync(() => {
118 runners.push(effect);
119 })) as unknown as ProcessContext["run"],
120 serve,
121 exports: Effect.sync(() => ({
122 default: Effect.all(
123 runners.map((eff) =>
124 Effect.forever(
125 eff.pipe(
126 // Log and ignore errors (daemon mode, it should just re-run)
127 Effect.tapError((err) => Effect.logError(err)),
128 Effect.ignore,
129 // TODO(sam): ignore cause? for now, let that actually kill the server
130 // Effect.ignoreCause
131 ),
132 ),
133 ),
134 {
135 concurrency: "unbounded",
136 },
137 ),
138 })),
139 } as ProcessContext;
140 },
141 },
142 {
143 bind: Effect.fn(function* <Shape, Req = never>(
144 containerEff:
145 | (ContainerApplication & Rpc<Shape>)
146 | Effect.Effect<ContainerApplication & Rpc<Shape>, never, Req>,
147 ) {
148 const namespace = yield* DurableObject;
149
150 const container = Effect.isEffect(containerEff)
151 ? yield* containerEff as unknown as Effect.Effect<
152 ContainerApplication & Rpc<Shape>
153 >
154 : containerEff;
155
156 yield* container.bind`${namespace}`({
157 durableObjects: {
158 namespaceId: namespace.namespaceId,
159 },
160 });
161
162 const worker = yield* Worker;
163 const className = namespace.name;
164
165 yield* worker.bind`${container.LogicalId}`({
166 containers: [
167 {
168 className,
169 dev: container.dev,
170 hash: container.hash.pipe(Output.map((h) => h?.image)),
171 },
172 ],
173 });
174
175 // TODO(sam): register this in the Container Execution Context
176 // const _httpEffect = yield* init;
177 return Effect.gen(function* () {
178 const state = yield* DurableObjectState;
179 return {
180 id: container.LogicalId,
181 running: Effect.sync(() => state.container!.running ?? false),
182 destroy: (error?: any) =>
183 Effect.promise(() => state.container!.destroy(error)),
184 signal: (signo: number) =>
185 Effect.sync(() => state.container!.signal(signo)),
186 getTcpPort: (port: number) =>
187 Effect.sync(() =>
188 fromCloudflareFetcher(
189 httpSchemePort(state.container!.getTcpPort(port)),
190 ),
191 ),
192 setInactivityTimeout: (durationMs: number | bigint) =>
193 Effect.promise(() =>
194 state.container!.setInactivityTimeout(durationMs),
195 ),
196 interceptOutboundHttp: (addr: string, binding: Fetcher) =>
197 toCloudflareFetcher(binding).pipe(
198 Effect.map((binding) =>
199 state.container!.interceptOutboundHttp(addr, binding),
200 ),
201 ),
202 interceptAllOutboundHttp: (binding: Fetcher) =>
203 toCloudflareFetcher(binding).pipe(
204 Effect.map((binding) =>
205 state.container!.interceptAllOutboundHttp(binding),
206 ),
207 ),
208 monitor: () =>
209 Effect.promise(
210 () => state.container?.monitor() ?? Promise.resolve(),
211 ),
212 start: (options?: ContainerStartupOptions) =>
213 Effect.sync(() => state.container!.start(options)),
214 } as unknown;
215 });
216 }),
217 },
218);