diff --git a/workers/datalake/src/blob.ts b/workers/datalake/src/blob.ts index 7b79a184a0..9742277b92 100644 --- a/workers/datalake/src/blob.ts +++ b/workers/datalake/src/blob.ts @@ -18,7 +18,7 @@ import postgres from 'postgres' import * as db from './db' import { toUUID } from './encodings' import { selectStorage } from './storage' -import { type UUID } from './types' +import { type BlobRequest, type WorkspaceRequest, type UUID } from './types' import { copyVideo, deleteVideo } from './video' const expires = 86400 @@ -39,13 +39,9 @@ export function getBlobURL (request: Request, workspace: string, name: string): return new URL(path, request.url).toString() } -export async function handleBlobGet ( - request: Request, - env: Env, - ctx: ExecutionContext, - workspace: string, - name: string -): Promise { +export async function handleBlobGet (request: BlobRequest, env: Env, ctx: ExecutionContext): Promise { + const { workspace, name } = request + const sql = postgres(env.HYPERDRIVE.connectionString) const { bucket } = selectStorage(env, workspace) @@ -82,13 +78,9 @@ export async function handleBlobGet ( return response } -export async function handleBlobHead ( - request: Request, - env: Env, - ctx: ExecutionContext, - workspace: string, - name: string -): Promise { +export async function handleBlobHead (request: BlobRequest, env: Env, ctx: ExecutionContext): Promise { + const { workspace, name } = request + const sql = postgres(env.HYPERDRIVE.connectionString) const { bucket } = selectStorage(env, workspace) @@ -106,7 +98,9 @@ export async function handleBlobHead ( return new Response(null, { headers, status: 200 }) } -export async function deleteBlob (env: Env, workspace: string, name: string): Promise { +export async function handleBlobDelete (request: BlobRequest, env: Env, ctx: ExecutionContext): Promise { + const { workspace, name } = request + const sql = postgres(env.HYPERDRIVE.connectionString) try { @@ -120,13 +114,15 @@ export async function deleteBlob (env: Env, workspace: string, name: string): Pr } } -export async function postBlobFormData (request: Request, env: Env, workspace: string): Promise { +export async function handleUploadFormData (request: WorkspaceRequest, env: Env): Promise { const contentType = request.headers.get('Content-Type') if (contentType === null || !contentType.includes('multipart/form-data')) { console.error({ error: 'expected multipart/form-data' }) return error(400, 'expected multipart/form-data') } + const { workspace } = request + const sql = postgres(env.HYPERDRIVE.connectionString) let formData: FormData diff --git a/workers/datalake/src/image.ts b/workers/datalake/src/image.ts index c309d114b2..1293e5c89f 100644 --- a/workers/datalake/src/image.ts +++ b/workers/datalake/src/image.ts @@ -14,15 +14,17 @@ // import { getBlobURL } from './blob' +import { type BlobRequest } from './types' const prefferedImageFormats = ['webp', 'avif', 'jpeg', 'png'] -export async function getImage ( - request: Request, - workspace: string, - name: string, - transform: string -): Promise { +export async function handleImageGet (request: BlobRequest): Promise { + const { + workspace, + name, + params: { transform } + } = request + const Accept = request.headers.get('Accept') ?? 'image/*' const image: Record = {} diff --git a/workers/datalake/src/index.ts b/workers/datalake/src/index.ts index ffd0fed9e1..83940825ac 100644 --- a/workers/datalake/src/index.ts +++ b/workers/datalake/src/index.ts @@ -13,61 +13,90 @@ // limitations under the License. // -import { type IRequest, Router, error, html } from 'itty-router' -import { - deleteBlob as handleBlobDelete, - handleBlobGet, - handleBlobHead, - postBlobFormData as handleUploadFormData -} from './blob' +import { WorkerEntrypoint } from 'cloudflare:workers' +import { type IRequestStrict, type RequestHandler, Router, error, html } from 'itty-router' + +import { handleBlobDelete, handleBlobGet, handleBlobHead, handleUploadFormData } from './blob' import { cors } from './cors' -import { getImage as handleImageGet } from './image' -import { getVideoMeta as handleVideoMetaGet } from './video' +import { handleImageGet } from './image' +import { handleVideoMetaGet } from './video' import { handleSignAbort, handleSignComplete, handleSignCreate } from './sign' +import { type BlobRequest, type WorkspaceRequest } from './types' const { preflight, corsify } = cors({ maxAge: 86400 }) -export default { - async fetch (request, env, ctx): Promise { - const router = Router({ - before: [preflight], - finally: [corsify] - }) +const router = Router({ + before: [preflight], + finally: [corsify] +}) - router - .get('/blob/:workspace/:name', ({ params }) => handleBlobGet(request, env, ctx, params.workspace, params.name)) - .head('/blob/:workspace/:name', ({ params }) => handleBlobHead(request, env, ctx, params.workspace, params.name)) - .delete('/blob/:workspace/:name', ({ params }) => handleBlobDelete(env, params.workspace, params.name)) - // Image - .get('/image/:transform/:workspace/:name', ({ params }) => - handleImageGet(request, params.workspace, params.name, params.transform) - ) - // Video - .get('/video/:workspace/:name/meta', ({ params }) => - handleVideoMetaGet(request, env, ctx, params.workspace, params.name) - ) - // Form Data - .post('/upload/form-data/:workspace', ({ params }) => handleUploadFormData(request, env, params.workspace)) - // Signed URL - .post('/upload/signed-url/:workspace/:name', ({ params }) => - handleSignCreate(request, env, ctx, params.workspace, params.name) - ) - .put('/upload/signed-url/:workspace/:name', ({ params }) => - handleSignComplete(request, env, ctx, params.workspace, params.name) - ) - .delete('/upload/signed-url/:workspace/:name', ({ params }) => - handleSignAbort(request, env, ctx, params.workspace, params.name) - ) - .all('/', () => - html( - `Huly® Datalake™ https://huly.io - © 2024 Huly Labs` - ) - ) - .all('*', () => error(404)) - - return await router.fetch(request).catch(error) +const withWorkspace: RequestHandler = (request: WorkspaceRequest) => { + if (request.params.workspace === undefined || request.params.workspace === '') { + return error(400, 'Missing workspace') } -} satisfies ExportedHandler + request.workspace = decodeURIComponent(request.params.workspace) +} + +const withBlob: RequestHandler = (request: BlobRequest) => { + if (request.params.name === undefined || request.params.name === '') { + return error(400, 'Missing blob name') + } + request.workspace = decodeURIComponent(request.params.name) +} + +router + .get('/blob/:workspace/:name', withBlob, handleBlobGet) + .head('/blob/:workspace/:name', withBlob, handleBlobHead) + .delete('/blob/:workspace/:name', withBlob, handleBlobDelete) + // Image + .get('/image/:transform/:workspace/:name', withBlob, handleImageGet) + // Video + .get('/video/:workspace/:name/meta', withBlob, handleVideoMetaGet) + // Form Data + .post('/upload/form-data/:workspace', withWorkspace, handleUploadFormData) + // Signed URL + .post('/upload/signed-url/:workspace/:name', withBlob, handleSignCreate) + .put('/upload/signed-url/:workspace/:name', withBlob, handleSignComplete) + .delete('/upload/signed-url/:workspace/:name', withBlob, handleSignAbort) + .all('/', () => + html( + `Huly® Datalake™ https://huly.io + © 2024 Huly Labs` + ) + ) + .all('*', () => error(404)) + +export default class DatalakeWorker extends WorkerEntrypoint { + async fetch (request: Request): Promise { + return await router.fetch(request, this.env, this.ctx).catch(error) + } + + async getBlob (workspace: string, name: string): Promise { + const request = new Request(`https://datalake/blob/${workspace}/${name}`) + const response = await router.fetch(request) + + if (!response.ok) { + console.error({ error: 'datalake error: ' + response.statusText, workspace, name }) + throw new Error(`Failed to fetch blob: ${response.statusText}`) + } + + return await response.arrayBuffer() + } + + async putBlob (workspace: string, name: string, data: ArrayBuffer | Blob | string, type: string): Promise { + const request = new Request(`https://datalake/upload/form-data/${workspace}`) + + const body = new FormData() + const blob = new Blob([data], { type }) + body.set('file', blob, name) + + const response = await router.fetch(request, { method: 'POST', body }) + + if (!response.ok) { + console.error({ error: 'datalake error: ' + response.statusText, workspace, name }) + throw new Error(`Failed to fetch blob: ${response.statusText}`) + } + } +} diff --git a/workers/datalake/src/sign.ts b/workers/datalake/src/sign.ts index e1b2ed1aad..28f97f8f52 100644 --- a/workers/datalake/src/sign.ts +++ b/workers/datalake/src/sign.ts @@ -17,7 +17,7 @@ import { AwsClient } from 'aws4fetch' import { error } from 'itty-router' import { handleBlobUploaded } from './blob' -import { type UUID } from './types' +import { type BlobRequest, type UUID } from './types' import { selectStorage, type Storage } from './storage' const S3_SIGNED_LINK_TTL = 3600 @@ -39,13 +39,8 @@ function getS3Client (storage: Storage): AwsClient { }) } -export async function handleSignCreate ( - request: Request, - env: Env, - ctx: ExecutionContext, - workspace: string, - name: string -): Promise { +export async function handleSignCreate (request: BlobRequest, env: Env, ctx: ExecutionContext): Promise { + const { workspace, name } = request const storage = selectStorage(env, workspace) const accountId = env.R2_ACCOUNT_ID @@ -78,13 +73,9 @@ export async function handleSignCreate ( return new Response(signed.url, { status: 200, headers }) } -export async function handleSignComplete ( - request: Request, - env: Env, - ctx: ExecutionContext, - workspace: string, - name: string -): Promise { +export async function handleSignComplete (request: BlobRequest, env: Env, ctx: ExecutionContext): Promise { + const { workspace, name } = request + const { bucket } = selectStorage(env, workspace) const key = signBlobKey(workspace, name) @@ -117,13 +108,9 @@ export async function handleSignComplete ( return new Response(null, { status: 201 }) } -export async function handleSignAbort ( - request: Request, - env: Env, - ctx: ExecutionContext, - workspace: string, - name: string -): Promise { +export async function handleSignAbort (request: BlobRequest, env: Env, ctx: ExecutionContext): Promise { + const { workspace, name } = request + const key = signBlobKey(workspace, name) // Check if the blob has been uploaded diff --git a/workers/datalake/src/types.ts b/workers/datalake/src/types.ts index 5e5baad474..acad59f013 100644 --- a/workers/datalake/src/types.ts +++ b/workers/datalake/src/types.ts @@ -13,10 +13,21 @@ // limitations under the License. // +import { type IRequestStrict } from 'itty-router' + export type Location = 'weur' | 'eeur' | 'wnam' | 'enam' | 'apac' export type UUID = string & { __uuid: true } +export type WorkspaceRequest = { + workspace: string +} & IRequestStrict + +export type BlobRequest = { + workspace: string + name: string +} & IRequestStrict + export interface CloudflareResponse { success: boolean errors: any diff --git a/workers/datalake/src/video.ts b/workers/datalake/src/video.ts index 229490c009..b8effcbffa 100644 --- a/workers/datalake/src/video.ts +++ b/workers/datalake/src/video.ts @@ -15,7 +15,7 @@ import { error, json } from 'itty-router' -import { type CloudflareResponse, type StreamUploadResponse } from './types' +import { type BlobRequest, type CloudflareResponse, type StreamUploadResponse } from './types' export type StreamUploadState = 'ready' | 'error' | 'inprogress' | 'queued' | 'downloading' | 'pendingupload' @@ -42,13 +42,9 @@ function streamBlobKey (workspace: string, name: string): string { return `v/${workspace}/${name}` } -export async function getVideoMeta ( - request: Request, - env: Env, - ctx: ExecutionContext, - workspace: string, - name: string -): Promise { +export async function handleVideoMetaGet (request: BlobRequest, env: Env, ctx: ExecutionContext): Promise { + const { workspace, name } = request + const key = streamBlobKey(workspace, name) const streamInfo = await env.datalake_blobs.get(key, { type: 'json' })