| author | |
| committer | |
| log | a01b227d6f37d1cfed7a2f47aaace2f72ea76ce3 |
| tree | 9d61ff67b5206779d17dc124faab2c5714798dd8 |
| parent | 9fe4d6568b5e4852f6085d3a22dbc5aa3a632408 |
| signature | Signed by PGP key B5690EEEBB952194 |
Co-authored-by: maria-rcks <254055478+maria-rcks@users.noreply.github.com>6 files changed, 248 insertions(+), 11 deletions(-)
apps/server/src/server.test.ts+37| ... | ... | @@ -5473,6 +5473,43 @@ it.layer(NodeServices.layer)("server router seam", (it) => { |
| 5473 | 5473 | }).pipe(Effect.provide(NodeHttpServer.layerTest)), |
| 5474 | 5474 | ); |
| 5475 | 5475 | |
| 5476 | it.effect("serves absolute host media without a local thread and rejects relative media", () => | |
| 5477 | Effect.gen(function* () { | |
| 5478 | yield* buildAppUnderTest(); | |
| 5479 | const fileSystem = yield* FileSystem.FileSystem; | |
| 5480 | const path = yield* Path.Path; | |
| 5481 | const directory = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-host-media-" }); | |
| 5482 | const wsUrl = yield* getWsServerUrl("/ws"); | |
| 5483 | const threadId = ThreadId.make("thread-on-another-environment"); | |
| 5484 | ||
| 5485 | yield* Effect.scoped( | |
| 5486 | withWsRpcClient(wsUrl, (client) => | |
| 5487 | Effect.gen(function* () { | |
| 5488 | for (const [name, mimeType] of [ | |
| 5489 | ["screenshot.png", "image/png"], | |
| 5490 | ["recording.mp4", "video/mp4"], | |
| 5491 | ] as const) { | |
| 5492 | const filePath = path.join(directory, name); | |
| 5493 | yield* fileSystem.writeFileString(filePath, "host media bytes"); | |
| 5494 | const issued = yield* client[WS_METHODS.assetsCreateUrl]({ | |
| 5495 | resource: { _tag: "media-file", threadId, path: filePath }, | |
| 5496 | }); | |
| 5497 | const response = yield* HttpClient.get(issued.relativeUrl); | |
| 5498 | assert.equal(response.status, 200); | |
| 5499 | assert.equal(response.headers["content-type"], mimeType); | |
| 5500 | assert.equal(yield* response.text, "host media bytes"); | |
| 5501 | ||
| 5502 | const error = yield* client[WS_METHODS.assetsCreateUrl]({ | |
| 5503 | resource: { _tag: "media-file", threadId, path: name }, | |
| 5504 | }).pipe(Effect.flip); | |
| 5505 | assert.equal(error._tag, "AssetWorkspaceContextNotFoundError"); | |
| 5506 | } | |
| 5507 | }), | |
| 5508 | ), | |
| 5509 | ); | |
| 5510 | }).pipe(Effect.provide(NodeHttpServer.layerTest)), | |
| 5511 | ); | |
| 5512 | ||
| 5476 | 5513 | it.effect("uploads image bytes through a signed URL issued by websocket rpc", () => |
| 5477 | 5514 | Effect.gen(function* () { |
| 5478 | 5515 | const config = yield* buildAppUnderTest(); |
apps/server/src/ws.ts+5-1| ... | ... | @@ -9,6 +9,7 @@ import * as Duration from "effect/Duration"; |
| 9 | 9 | import * as Effect from "effect/Effect"; |
| 10 | 10 | import * as Layer from "effect/Layer"; |
| 11 | 11 | import * as Option from "effect/Option"; |
| 12 | import * as Path from "effect/Path"; | |
| 12 | 13 | import * as Queue from "effect/Queue"; |
| 13 | 14 | import * as Ref from "effect/Ref"; |
| 14 | 15 | import * as Schema from "effect/Schema"; |
| ... | ... | @@ -2390,9 +2391,12 @@ const makeWsRpcLayer = ( |
| 2390 | 2391 | observeRpcEffect( |
| 2391 | 2392 | WS_METHODS.assetsCreateUrl, |
| 2392 | 2393 | Effect.gen(function* () { |
| 2394 | const path = yield* Path.Path; | |
| 2395 | // An absolute media path can be linked from a thread on another environment. | |
| 2393 | 2396 | if ( |
| 2394 | 2397 | input.resource._tag === "attachment" || |
| 2395 | input.resource._tag === "native-app-icon" | |
| 2398 | input.resource._tag === "native-app-icon" || | |
| 2399 | (input.resource._tag === "media-file" && path.isAbsolute(input.resource.path)) | |
| 2396 | 2400 | ) { |
| 2397 | 2401 | return yield* issueAssetUrl({ resource: input.resource }); |
| 2398 | 2402 | } |
apps/web/src/state/assets.ts+19-1| ... | ... | @@ -2,12 +2,30 @@ import { |
| 2 | 2 | createAssetEnvironmentAtoms, |
| 3 | 3 | createProjectFaviconUrlAtomFamily, |
| 4 | 4 | } from "@t3tools/client-runtime/state/assets"; |
| 5 | import { Atom } from "effect/unstable/reactivity"; | |
| 5 | 6 | |
| 6 | 7 | import { connectionAtomRuntime } from "../connection/runtime"; |
| 7 | 8 | import { projectFaviconCache } from "../assets/projectFaviconCache"; |
| 9 | import { isElectron } from "../env"; | |
| 10 | import { primaryEnvironmentIdAtom } from "./primaryEnvironment"; | |
| 8 | 11 | import { environmentSession } from "./session"; |
| 9 | 12 | |
| 10 | export const assetEnvironment = createAssetEnvironmentAtoms(connectionAtomRuntime); | |
| 13 | const localMediaEnvironment = Atom.make((get) => { | |
| 14 | if (!isElectron) return null; | |
| 15 | const environmentId = get(primaryEnvironmentIdAtom); | |
| 16 | if (environmentId === null) return null; | |
| 17 | const connection = get(environmentSession.preparedConnectionValueAtom(environmentId)); | |
| 18 | // The session's bootstrap config clears on disconnect and refreshes on reconnect. | |
| 19 | const config = get(environmentSession.initialConfigValueAtom(environmentId)); | |
| 20 | return connection._tag === "None" || config === null | |
| 21 | ? null | |
| 22 | : { environmentId, httpBaseUrl: connection.value.httpBaseUrl }; | |
| 23 | }); | |
| 24 | ||
| 25 | export const assetEnvironment = createAssetEnvironmentAtoms( | |
| 26 | connectionAtomRuntime, | |
| 27 | localMediaEnvironment, | |
| 28 | ); | |
| 11 | 29 | |
| 12 | 30 | export const projectFaviconUrlAtom = createProjectFaviconUrlAtomFamily({ |
| 13 | 31 | imageCache: projectFaviconCache, |
packages/client-runtime/src/state/assets.test.ts+133-2| ... | ... | @@ -1,11 +1,32 @@ |
| 1 | 1 | import { describe, expect, it } from "@effect/vitest"; |
| 2 | import { type AssetCreateUrlResult, EnvironmentId } from "@t3tools/contracts"; | |
| 2 | import { | |
| 3 | type AssetCreateUrlResult, | |
| 4 | AssetWorkspaceAssetNotFoundError, | |
| 5 | AssetWorkspaceAssetInspectionError, | |
| 6 | AssetWorkspaceContextNotFoundError, | |
| 7 | EnvironmentAuthorizationError, | |
| 8 | EnvironmentId, | |
| 9 | ThreadId, | |
| 10 | WS_METHODS, | |
| 11 | } from "@t3tools/contracts"; | |
| 3 | 12 | import * as Cause from "effect/Cause"; |
| 13 | import * as Effect from "effect/Effect"; | |
| 14 | import * as Stream from "effect/Stream"; | |
| 15 | import * as SubscriptionRef from "effect/SubscriptionRef"; | |
| 4 | 16 | import * as Option from "effect/Option"; |
| 5 | 17 | import * as Layer from "effect/Layer"; |
| 6 | 18 | import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity"; |
| 7 | 19 | |
| 8 | import type { EnvironmentRegistry } from "../connection/registry.ts"; | |
| 20 | import { EnvironmentRegistry } from "../connection/registry.ts"; | |
| 21 | import { | |
| 22 | AVAILABLE_CONNECTION_STATE, | |
| 23 | PrimaryConnectionTarget, | |
| 24 | type PreparedConnection, | |
| 25 | type SupervisorConnectionState, | |
| 26 | } from "../connection/model.ts"; | |
| 27 | import { EnvironmentSupervisor } from "../connection/supervisor.ts"; | |
| 28 | import type { RpcSession } from "../rpc/session.ts"; | |
| 29 | import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; | |
| 9 | 30 | import { createProjectFaviconCache } from "../projectFaviconCache.ts"; |
| 10 | 31 | import { |
| 11 | 32 | createAssetEnvironmentAtoms, |
| ... | ... | @@ -37,6 +58,116 @@ describe("asset collection keys", () => { |
| 37 | 58 | }); |
| 38 | 59 | |
| 39 | 60 | describe("createAssetEnvironmentAtoms", () => { |
| 61 | for (const scenario of [ | |
| 62 | { name: "missing video", path: "/tmp/clip.mp4", fallback: true }, | |
| 63 | { name: "literal filename characters", path: "/tmp/frame#one?two.png", fallback: true }, | |
| 64 | { name: "windows path", path: "C:\\Users\\demo\\clip.mp4", fallback: true }, | |
| 65 | { name: "inspection failure", path: "/tmp/clip.mp4", error: "inspection", fallback: true }, | |
| 66 | { name: "foreign thread", path: "/tmp/clip.mp4", error: "context", fallback: true }, | |
| 67 | { name: "remote success", path: "/tmp/clip.mp4", success: true }, | |
| 68 | { name: "relative path", path: "clip.mp4" }, | |
| 69 | { name: "no primary", path: "/tmp/clip.mp4", primary: "none" }, | |
| 70 | { name: "primary reconnect", path: "/tmp/frame.png", primary: "reconnecting" }, | |
| 71 | { name: "same environment", path: "/tmp/clip.mp4", primary: "same" }, | |
| 72 | { name: "non-media", path: "/tmp/report.html" }, | |
| 73 | { name: "authorization failure", path: "/tmp/clip.mp4", error: "auth" }, | |
| 74 | ]) { | |
| 75 | it.effect(`uses the correct environment for ${scenario.name}`, () => | |
| 76 | Effect.gen(function* () { | |
| 77 | const remoteId = EnvironmentId.make("remote"); | |
| 78 | const localId = EnvironmentId.make("local"); | |
| 79 | const resource = { | |
| 80 | _tag: "media-file" as const, | |
| 81 | threadId: ThreadId.make("foreign-thread"), | |
| 82 | path: scenario.path, | |
| 83 | }; | |
| 84 | const error = | |
| 85 | scenario.error === "auth" | |
| 86 | ? new EnvironmentAuthorizationError({ | |
| 87 | message: "denied", | |
| 88 | requiredScope: "orchestration:read", | |
| 89 | }) | |
| 90 | : scenario.error === "inspection" | |
| 91 | ? new AssetWorkspaceAssetInspectionError({ resource, cause: new Error("unreadable") }) | |
| 92 | : scenario.error === "context" | |
| 93 | ? new AssetWorkspaceContextNotFoundError({ resource }) | |
| 94 | : new AssetWorkspaceAssetNotFoundError({ resource }); | |
| 95 | const calls: EnvironmentId[] = []; | |
| 96 | const supervisors = new Map<EnvironmentId, EnvironmentSupervisor["Service"]>(); | |
| 97 | for (const environmentId of [remoteId, localId]) { | |
| 98 | const client = { | |
| 99 | [WS_METHODS.assetsCreateUrl]: () => { | |
| 100 | calls.push(environmentId); | |
| 101 | return environmentId === remoteId && !scenario.success | |
| 102 | ? Effect.fail(error) | |
| 103 | : Effect.succeed({ | |
| 104 | relativeUrl: `/api/assets/${environmentId}/media`, | |
| 105 | expiresAt: 999999, | |
| 106 | }); | |
| 107 | }, | |
| 108 | } as unknown as WsRpcProtocolClient; | |
| 109 | const session = { client } as RpcSession; | |
| 110 | supervisors.set( | |
| 111 | environmentId, | |
| 112 | EnvironmentSupervisor.of({ | |
| 113 | target: new PrimaryConnectionTarget({ | |
| 114 | environmentId, | |
| 115 | label: environmentId, | |
| 116 | httpBaseUrl: `https://${environmentId}.test`, | |
| 117 | wsBaseUrl: `wss://${environmentId}.test`, | |
| 118 | }), | |
| 119 | state: yield* SubscriptionRef.make<SupervisorConnectionState>({ | |
| 120 | ...AVAILABLE_CONNECTION_STATE, | |
| 121 | phase: "connected" as const, | |
| 122 | }), | |
| 123 | session: yield* SubscriptionRef.make(Option.some(session)), | |
| 124 | prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()), | |
| 125 | connect: Effect.void, | |
| 126 | disconnect: Effect.void, | |
| 127 | retryNow: Effect.void, | |
| 128 | }), | |
| 129 | ); | |
| 130 | } | |
| 131 | const environments = EnvironmentRegistry.of({ | |
| 132 | run: (id, effect) => | |
| 133 | Effect.provideService(effect, EnvironmentSupervisor, supervisors.get(id)!), | |
| 134 | followStream: (id, stream) => | |
| 135 | Stream.provideService(stream, EnvironmentSupervisor, supervisors.get(id)!), | |
| 136 | } as EnvironmentRegistry["Service"]); | |
| 137 | const registry = AtomRegistry.make(); | |
| 138 | yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose())); | |
| 139 | const localTarget = { | |
| 140 | environmentId: scenario.primary === "same" ? remoteId : localId, | |
| 141 | httpBaseUrl: "https://local.test", | |
| 142 | }; | |
| 143 | const localEnvironment = Atom.make<typeof localTarget | null>( | |
| 144 | scenario.primary === "none" || scenario.primary === "reconnecting" ? null : localTarget, | |
| 145 | ); | |
| 146 | const assets = createAssetEnvironmentAtoms( | |
| 147 | Atom.runtime(Layer.succeed(EnvironmentRegistry, environments)), | |
| 148 | localEnvironment, | |
| 149 | ); | |
| 150 | const query = assets.createUrl({ environmentId: remoteId, input: { resource } }); | |
| 151 | const result = AtomRegistry.getResult(registry, query, { suspendOnWaiting: true }); | |
| 152 | if (scenario.fallback || scenario.success) { | |
| 153 | expect((yield* result).relativeUrl).toBe( | |
| 154 | scenario.fallback | |
| 155 | ? "https://local.test/api/assets/local/media" | |
| 156 | : "/api/assets/remote/media", | |
| 157 | ); | |
| 158 | } else { | |
| 159 | expect(yield* Effect.flip(result)).toEqual(error); | |
| 160 | } | |
| 161 | expect(calls).toEqual(scenario.fallback ? [remoteId, localId] : [remoteId]); | |
| 162 | if (scenario.primary === "reconnecting") { | |
| 163 | registry.set(localEnvironment, localTarget); | |
| 164 | expect((yield* result).relativeUrl).toBe("https://local.test/api/assets/local/media"); | |
| 165 | expect(calls).toEqual([remoteId, remoteId, localId]); | |
| 166 | } | |
| 167 | }).pipe(Effect.scoped), | |
| 168 | ); | |
| 169 | } | |
| 170 | ||
| 40 | 171 | it("keys asset URL queries by environment and resource", () => { |
| 41 | 172 | const runtime = Atom.runtime(Layer.empty) as unknown as Atom.AtomRuntime< |
| 42 | 173 | EnvironmentRegistry, |
packages/client-runtime/src/state/assets.ts+48-6| ... | ... | @@ -1,22 +1,28 @@ |
| 1 | 1 | import { |
| 2 | type AssetCreateUrlInput, | |
| 2 | 3 | type AssetCreateUrlResult, |
| 3 | 4 | type AssetImageDimensions, |
| 4 | 5 | AssetResource, |
| 5 | 6 | EnvironmentId, |
| 6 | 7 | WS_METHODS, |
| 7 | 8 | } from "@t3tools/contracts"; |
| 9 | import { mediaMimeTypeFromExtension } from "@t3tools/shared/filePreview"; | |
| 10 | import { isWindowsAbsolutePath } from "@t3tools/shared/path"; | |
| 8 | 11 | import { |
| 9 | 12 | getProjectFaviconResourceKey, |
| 10 | 13 | isProjectFaviconFallbackUrl, |
| 11 | 14 | } from "@t3tools/shared/projectFavicon"; |
| 12 | 15 | import * as Effect from "effect/Effect"; |
| 13 | 16 | import * as Option from "effect/Option"; |
| 17 | import * as Result from "effect/Result"; | |
| 14 | 18 | import * as Schema from "effect/Schema"; |
| 15 | import { AsyncResult, Atom } from "effect/unstable/reactivity"; | |
| 19 | import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity"; | |
| 16 | 20 | |
| 17 | import type { EnvironmentRegistry } from "../connection/registry.ts"; | |
| 21 | import * as EnvironmentRegistry from "../connection/registry.ts"; | |
| 22 | import * as EnvironmentSupervisor from "../connection/supervisor.ts"; | |
| 23 | import { request } from "../rpc/client.ts"; | |
| 18 | 24 | import type { ProjectFaviconCache, ProjectFaviconTarget } from "../projectFaviconCache.ts"; |
| 19 | import { createEnvironmentRpcQueryAtomFamily } from "./runtime.ts"; | |
| 25 | import { createEnvironmentQueryAtomFamily } from "./runtime.ts"; | |
| 20 | 26 | |
| 21 | 27 | const ASSET_URL_REFRESH_INTERVAL_MS = 30 * 60_000; |
| 22 | 28 | const ASSET_URL_STALE_TIME_MS = 5 * 60_000; |
| ... | ... | @@ -91,14 +97,50 @@ export function assetUrlStateFromResult( |
| 91 | 97 | } |
| 92 | 98 | |
| 93 | 99 | export function createAssetEnvironmentAtoms<R, E>( |
| 94 | runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>, | |
| 100 | runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, E>, | |
| 101 | localMediaEnvironment?: Atom.Atom<{ | |
| 102 | readonly environmentId: EnvironmentId; | |
| 103 | readonly httpBaseUrl: string; | |
| 104 | } | null>, | |
| 95 | 105 | ) { |
| 96 | const createUrl = createEnvironmentRpcQueryAtomFamily(runtime, { | |
| 106 | const execute = Effect.fn("assets.createUrl")(function* (input: AssetCreateUrlInput) { | |
| 107 | const result = yield* request(WS_METHODS.assetsCreateUrl, input).pipe(Effect.result); | |
| 108 | if (Result.isSuccess(result)) return result.success; | |
| 109 | const error = result.failure; | |
| 110 | const resource = input.resource; | |
| 111 | const local = localMediaEnvironment | |
| 112 | ? (yield* AtomRegistry.AtomRegistry).get(localMediaEnvironment) | |
| 113 | : null; | |
| 114 | const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor; | |
| 115 | if ( | |
| 116 | !local || | |
| 117 | local.environmentId === supervisor.target.environmentId || | |
| 118 | !( | |
| 119 | error._tag === "AssetWorkspaceAssetNotFoundError" || | |
| 120 | error._tag === "AssetWorkspaceAssetInspectionError" || | |
| 121 | error._tag === "AssetWorkspaceContextNotFoundError" | |
| 122 | ) || | |
| 123 | resource._tag !== "media-file" || | |
| 124 | !(resource.path.startsWith("/") || isWindowsAbsolutePath(resource.path)) || | |
| 125 | mediaMimeTypeFromExtension(resource.path.slice(resource.path.lastIndexOf("."))) === null | |
| 126 | ) | |
| 127 | return yield* error; | |
| 128 | const registry = yield* EnvironmentRegistry.EnvironmentRegistry; | |
| 129 | const asset = yield* registry.run( | |
| 130 | local.environmentId, | |
| 131 | request(WS_METHODS.assetsCreateUrl, input), | |
| 132 | ); | |
| 133 | // Callers resolve against the thread's server, so preserve the local server's origin. | |
| 134 | return { ...asset, relativeUrl: new URL(asset.relativeUrl, local.httpBaseUrl).href }; | |
| 135 | }); | |
| 136 | const createUrl = createEnvironmentQueryAtomFamily(runtime, { | |
| 97 | 137 | label: "environment-data:assets:create-url", |
| 98 | tag: WS_METHODS.assetsCreateUrl, | |
| 138 | execute, | |
| 99 | 139 | staleTimeMs: ASSET_URL_STALE_TIME_MS, |
| 100 | 140 | idleTtlMs: ASSET_URL_IDLE_TTL_MS, |
| 101 | 141 | refreshIntervalMs: ASSET_URL_REFRESH_INTERVAL_MS, |
| 142 | refreshTrigger: ({ input }) => | |
| 143 | input.resource._tag === "media-file" ? localMediaEnvironment : undefined, | |
| 102 | 144 | }); |
| 103 | 145 | const createUrlsFamily = Atom.family((key: string) => { |
| 104 | 146 | const [environmentId, resources] = parseAssetCollectionKey(key); |
packages/client-runtime/src/state/runtime.ts+6-1| ... | ... | @@ -482,7 +482,12 @@ export function followStreamInEnvironment<A, E, R>( |
| 482 | 482 | |
| 483 | 483 | export function createEnvironmentQueryAtomFamily<R, ER, Input, A, E>( |
| 484 | 484 | runtime: Atom.AtomRuntime<EnvironmentRegistry | R, ER>, |
| 485 | options: EnvironmentQueryAtomOptions<Input, A, E, EnvironmentSupervisor | R>, | |
| 485 | options: EnvironmentQueryAtomOptions< | |
| 486 | Input, | |
| 487 | A, | |
| 488 | E, | |
| 489 | EnvironmentSupervisor | EnvironmentRegistry | AtomRegistry.AtomRegistry | R | |
| 490 | >, | |
| 486 | 491 | ): (target: { |
| 487 | 492 | readonly environmentId: EnvironmentIdType; |
| 488 | 493 | readonly input: Input; |