1import { describe, expect, it } from "@effect/vitest";
2import {
3 AuthStandardClientScopes,
4 EnvironmentId,
5 EnvironmentAuthorizationError,
6 WS_METHODS,
7} from "@t3tools/contracts";
8import * as Effect from "effect/Effect";
9import * as DateTime from "effect/DateTime";
10import * as Layer from "effect/Layer";
11import * as Option from "effect/Option";
12import * as Stream from "effect/Stream";
13import * as SubscriptionRef from "effect/SubscriptionRef";
14import { Atom, AtomRegistry } from "effect/reactivity";
15import { vi } from "vite-plus/test";
16
17import * as EnvironmentRegistry from "../connection/registry.ts";
18import {
19 AVAILABLE_CONNECTION_STATE,
20 BearerConnectionTarget,
21 type PreparedConnection,
22 type SupervisorConnectionState,
23} from "../connection/model.ts";
24import * as EnvironmentSupervisor from "../connection/supervisor.ts";
25import type { RpcSession } from "../rpc/session.ts";
26import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
27import { layerRemoteHttpClient } from "../rpc/http.ts";
28import { createAssetEnvironmentAtoms, resolveAssetUrl } from "./assets.ts";
29import { createAttachmentEnvironmentAtoms, runAttachmentUploadCycle } from "./attachments.ts";
30import { runAtomCommand } from "./runtime.ts";
31
32const environmentId = EnvironmentId.make("laptop");
33const base = "https://hub.example/knife/environments/laptop/";
34
35const harness = (hub: boolean, allowed = true) =>
36 Effect.gen(function* () {
37 const tickets = new Map<string, number>();
38 const calls = { assets: 0, uploads: 0, tickets: 0 };
39 const target = new BearerConnectionTarget({
40 environmentId,
41 label: "Laptop",
42 connectionId: hub ? "hub:home:laptop" : "direct:laptop",
43 });
44 const prepared: PreparedConnection = {
45 environmentId,
46 label: "Laptop",
47 httpBaseUrl: base,
48 socketUrl: `${base.replace(/^http/, "ws")}ws`,
49 httpAuthorization: { _tag: "Bearer", token: "root-session" },
50 target,
51 };
52 const fetcher: typeof fetch = async (input, init) => {
53 const url = new URL(String(input));
54 if (url.pathname.endsWith("/api/auth/session"))
55 return Response.json({
56 authenticated: true,
57 auth: {
58 policy: "remote-reachable",
59 bootstrapMethods: [],
60 sessionMethods: [],
61 sessionCookieName: "test",
62 },
63 scopes: allowed ? AuthStandardClientScopes : [],
64 permissions: allowed ? AuthStandardClientScopes : [],
65 });
66 if (url.pathname.endsWith("/api/auth/websocket-ticket")) {
67 expect(new Headers(init?.headers).get("authorization")).toBe("Bearer root-session");
68 const ticket = `media-${++calls.tickets}`;
69 const expiresAt = DateTime.toEpochMillis(DateTime.nowUnsafe()) + 5 * 60_000;
70 tickets.set(ticket, expiresAt);
71 return Response.json({
72 ticket,
73 expiresAt: DateTime.formatIso(DateTime.makeUnsafe(expiresAt)),
74 });
75 }
76 if (
77 url.pathname.includes("/api/assets/") ||
78 url.pathname.includes("/api/attachments/upload/")
79 ) {
80 const authorized =
81 !hub ||
82 (tickets.get(url.searchParams.get("wsTicket") ?? "") ?? 0) >
83 DateTime.toEpochMillis(DateTime.nowUnsafe());
84 return new Response(authorized ? "file bytes" : "unauthorized", {
85 status: authorized ? 200 : 401,
86 });
87 }
88 throw new Error(`Unexpected URL ${url}`);
89 };
90 const client = {
91 [WS_METHODS.assetsCreateUrl]: () =>
92 Effect.sync(() => {
93 calls.assets++;
94 return {
95 relativeUrl: "/api/assets/leaf-signed/photo.png",
96 expiresAt: DateTime.toEpochMillis(DateTime.nowUnsafe()) + 3_600_000,
97 };
98 }),
99 [WS_METHODS.attachmentsCreateUploadUrl]: () =>
100 !allowed
101 ? Effect.fail(
102 new EnvironmentAuthorizationError({
103 requiredScope: "orchestration:operate",
104 message: "Not allowed",
105 }),
106 )
107 : Effect.sync(() => {
108 calls.uploads++;
109 return {
110 attachmentId: "pending",
111 relativeUrl: "/api/attachments/upload/leaf-signed",
112 expiresAt: DateTime.toEpochMillis(DateTime.nowUnsafe()) + 600_000,
113 };
114 }),
115 } as unknown as WsRpcProtocolClient;
116 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
117 target,
118 state: yield* SubscriptionRef.make<SupervisorConnectionState>({
119 ...AVAILABLE_CONNECTION_STATE,
120 phase: "connected" as const,
121 }),
122 prepared: yield* SubscriptionRef.make(Option.some(prepared)),
123 session: yield* SubscriptionRef.make(Option.some({ client } as RpcSession)),
124 connect: Effect.void,
125 disconnect: Effect.void,
126 retryNow: Effect.void,
127 });
128 const environments = EnvironmentRegistry.EnvironmentRegistry.of({
129 run: (_id, effect) =>
130 Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
131 followStream: (_id, stream) =>
132 Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
133 } as EnvironmentRegistry.EnvironmentRegistry["Service"]);
134 const runtime = Atom.runtime(
135 Layer.merge(
136 layerRemoteHttpClient(fetcher),
137 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environments),
138 ),
139 );
140 const registry = AtomRegistry.make();
141 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
142 const assets = createAssetEnvironmentAtoms(runtime);
143 const attachments = createAttachmentEnvironmentAtoms(runtime);
144 const query = assets.createUrl({
145 environmentId,
146 input: { resource: { _tag: "attachment", attachmentId: "photo" } },
147 });
148 return { registry, assets, attachments, query, fetcher, calls };
149 });
150
151describe("hub signed media capabilities", () => {
152 it.effect("authorizes image reads and guarded byte uploads through the public hub prefix", () =>
153 Effect.scoped(
154 Effect.gen(function* () {
155 const test = yield* harness(true);
156 const asset = yield* AtomRegistry.getResult(test.registry, test.query, {
157 suspendOnWaiting: true,
158 });
159 const url = resolveAssetUrl(base, asset.relativeUrl)!;
160 expect(url).toContain(`${base}api/assets/leaf-signed/photo.png?wsTicket=`);
161 expect(asset.expiresAt).toBeLessThanOrEqual(
162 DateTime.toEpochMillis(DateTime.nowUnsafe()) + 5 * 60_000,
163 );
164 expect((yield* Effect.promise(() => test.fetcher(url))).status).toBe(200);
165 expect((yield* Effect.promise(() => test.fetcher(url.split("?")[0]!))).status).toBe(401);
166 const upload = yield* Effect.promise(() =>
167 runAttachmentUploadCycle({
168 registry: test.registry,
169 createUploadUrl: test.attachments.createUploadUrl,
170 remove: test.attachments.remove,
171 environmentId,
172 upload: { type: "file", name: "notes.txt", mimeType: "text/plain", sizeBytes: 3 },
173 resolveUploadUrl: (relativeUrl) => resolveAssetUrl(base, relativeUrl),
174 transport: (uploadUrl) => ({
175 done: test.fetcher(uploadUrl, { method: "PUT", body: "abc" }).then((response) => {
176 expect(response.status).toBe(200);
177 }),
178 abort: () => {},
179 }),
180 }),
181 );
182 expect(upload).toEqual({ status: "uploaded", attachmentId: "pending" });
183 expect(test.calls).toEqual({ assets: 1, uploads: 1, tickets: 2 });
184 }),
185 ),
186 );
187
188 it.effect("does not mint a hub ticket when the server refuses the upload", () =>
189 Effect.scoped(
190 Effect.gen(function* () {
191 const test = yield* harness(true, false);
192 const result = yield* Effect.promise(() =>
193 runAtomCommand(
194 test.registry,
195 test.attachments.createUploadUrl,
196 {
197 environmentId,
198 input: { type: "file", name: "notes.txt", mimeType: "text/plain", sizeBytes: 3 },
199 },
200 { reportFailure: false },
201 ),
202 );
203 expect(result._tag).toBe("Failure");
204 expect(test.calls).toEqual({ assets: 0, uploads: 0, tickets: 0 });
205 }),
206 ),
207 );
208
209 it.effect.each([true, false])(
210 "refreshes ticket-bound images before expiry without shortening direct asset caching (hub=%s)",
211 (hub) =>
212 Effect.scoped(
213 Effect.gen(function* () {
214 vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] });
215 yield* Effect.addFinalizer(() => Effect.sync(() => vi.useRealTimers()));
216 const test = yield* harness(hub);
217 const unmount = test.registry.mount(test.query);
218 yield* Effect.addFinalizer(() => Effect.sync(unmount));
219 yield* AtomRegistry.getResult(test.registry, test.query, { suspendOnWaiting: true });
220 vi.advanceTimersByTime(4 * 60_000);
221 yield* Effect.yieldNow;
222 const asset = yield* AtomRegistry.getResult(test.registry, test.query, {
223 suspendOnWaiting: true,
224 });
225 expect(test.calls.assets).toBe(hub ? 2 : 1);
226 expect(test.calls.tickets).toBe(hub ? 2 : 0);
227 expect(
228 (yield* Effect.promise(() => test.fetcher(resolveAssetUrl(base, asset.relativeUrl)!)))
229 .status,
230 ).toBe(200);
231 }),
232 ),
233 );
234});