diff --git a/docs/changelog/2026-09-25-c225-large-web-media.md b/docs/changelog/2026-09-25-c225-large-web-media.md new file mode 100644 index 00000000..13b3c3ab --- /dev/null +++ b/docs/changelog/2026-09-25-c225-large-web-media.md @@ -0,0 +1 @@ +**C225-large — ingestion of web media above 2 GiB (2026-09-25):** the web-speech ingestion no longer asks the Payload local API to `fs.readFile` an arbitrarily large artifact. The mirror creates the private `internetSpeechMedia` row through a small placeholder, overwrites the same S3 key with a 64 MiB multipart stream, updates the stored `filesize`/MIME metadata, and removes the object if the transaction fails; the technical ceiling is 5 GiB and anything larger remains an honest `pending` failure. Added `@aws-sdk/client-s3`/`@aws-sdk/lib-storage` as runtime dependencies plus unit coverage for the multipart seam and integration coverage for the row metadata path. No schema or UI change; the 3h51m production video remains pending until this follow-up is deployed and the existing batch is retried. diff --git a/docs/ops/teqo-1313-deploy.md b/docs/ops/teqo-1313-deploy.md index e36ad321..08be35d7 100644 --- a/docs/ops/teqo-1313-deploy.md +++ b/docs/ops/teqo-1313-deploy.md @@ -796,6 +796,10 @@ operar depois do merge. do pipeline. Não use um diretório local como fila de upload. - `internetSpeechMedia` continua privada e servida pelo acervo autenticado. Não copie a URL do bucket para um canal público. +- Mídia acima de 2 GiB não passa pelo `fs.readFile` do Payload: o follow-up + C225-large cria a row com um placeholder e sobrescreve a mesma key por upload + multipart direto no S3. O teto técnico desta esteira é 5 GiB; acima disso a + falha fica em `pending`, sem fingir que a mídia foi espelhada. ### Pré-requisitos @@ -1104,6 +1108,10 @@ próprios. - **Egress do homeserver:** a rota para APIs/CDNs pode ser intermitente. O preflight é um gate: se `--dump-json` ou o Garage falhar, a rodada não começa. Não silencie a falha nem troque o banco de produção por um alvo remoto. +- **Mídia acima de 2 GiB:** o código anterior falhava no `fs.readFile` do Node + antes de chegar ao S3. O follow-up C225-large usa upload multipart e uma row + placeholder; confirme no relatório que a etapa `media` terminou e que o objeto + tem o tamanho real. O teto desta implementação é 5 GiB. - **S3 ausente ou endpoint errado:** o novo guard recusa escrita não-local sem as quatro `S3_*`; um erro de conectividade aparece no relatório. O env file do container não deve ser editado para acomodar o host. diff --git a/package.json b/package.json index f74d769b..27354b16 100644 --- a/package.json +++ b/package.json @@ -104,6 +104,8 @@ "dependencies": { "@ai-sdk/deepseek": "3.0.19", "@ai-sdk/react": "^4.0.50", + "@aws-sdk/client-s3": "3.1112.0", + "@aws-sdk/lib-storage": "3.1112.0", "@base-ui/react": "^1.6.0", "@fullcalendar/react": "7.0.2", "@hookform/resolvers": "^5.2.2", @@ -155,7 +157,6 @@ "zod": "^4.3.6" }, "devDependencies": { - "@aws-sdk/client-s3": "^3.1112.0", "@eslint/eslintrc": "^3.3.6", "@ffmpeg-installer/ffmpeg": "^1.1.0", "@playwright/test": "1.58.2", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 292b39ec..d8375cbe 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -17,6 +17,12 @@ importers: '@ai-sdk/react': specifier: ^4.0.50 version: 4.0.50(react@19.2.4)(zod@4.3.6) + '@aws-sdk/client-s3': + specifier: 3.1112.0 + version: 3.1112.0 + '@aws-sdk/lib-storage': + specifier: 3.1112.0 + version: 3.1112.0(@aws-sdk/client-s3@3.1112.0) '@base-ui/react': specifier: ^1.6.0 version: 1.6.0(@date-fns/tz@1.5.0)(@types/react@19.2.14)(date-fns@4.1.0)(react-dom@19.2.4(react@19.2.4))(react@19.2.4) @@ -165,9 +171,6 @@ importers: specifier: ^4.3.6 version: 4.3.6 devDependencies: - '@aws-sdk/client-s3': - specifier: ^3.1112.0 - version: 3.1112.0 '@eslint/eslintrc': specifier: ^3.3.6 version: 3.3.6 diff --git a/src/utilities/media/largeS3Upload.ts b/src/utilities/media/largeS3Upload.ts new file mode 100644 index 00000000..266609c1 --- /dev/null +++ b/src/utilities/media/largeS3Upload.ts @@ -0,0 +1,172 @@ +import 'server-only' + +import { randomUUID } from 'node:crypto' +import { createReadStream } from 'node:fs' +import { stat } from 'node:fs/promises' +import { extname } from 'node:path' + +import { DeleteObjectCommand, S3Client } from '@aws-sdk/client-s3' +import { Upload } from '@aws-sdk/lib-storage' + +import { resolveS3StorageEnv } from '@/utilities/mediaStorage' + +export const LARGE_MEDIA_THRESHOLD_BYTES = 2 * 1024 * 1024 * 1024 +const MAX_LARGE_MEDIA_BYTES = 5 * 1024 * 1024 * 1024 + +const MIME_BY_EXTENSION: Record = { + '.aac': 'audio/aac', + '.flac': 'audio/flac', + '.m4a': 'audio/mp4', + '.mka': 'audio/x-matroska', + '.mp3': 'audio/mpeg', + '.mp4': 'video/mp4', + '.ogg': 'audio/ogg', + '.opus': 'audio/opus', + '.wav': 'audio/wav', + '.webm': 'video/webm', +} + +export type LargeMediaUpload = { + filename: string + mimeType: string + filesize: number + url: string +} + +type S3UploadArgs = { + client: S3Client + bucket: string + key: string + contentType: string + inputPath: string + filesize: number +} + +type S3DeleteArgs = { + client: S3Client + bucket: string + key: string +} + +const storageClient = (storage: { + endpoint: string + region: string + accessKeyId: string + secretAccessKey: string +}): S3Client => + new S3Client({ + region: storage.region, + endpoint: storage.endpoint, + forcePathStyle: true, + credentials: { + accessKeyId: storage.accessKeyId, + secretAccessKey: storage.secretAccessKey, + }, + }) + +const defaultUpload = async ({ + client, + bucket, + key, + contentType, + inputPath, + filesize, +}: S3UploadArgs): Promise => { + const upload = new Upload({ + client, + params: { + Bucket: bucket, + Key: key, + Body: createReadStream(inputPath), + ContentLength: filesize, + ContentType: contentType, + }, + partSize: 64 * 1024 * 1024, + queueSize: 4, + }) + await upload.done() +} + +const defaultDelete = async ({ client, bucket, key }: S3DeleteArgs): Promise => { + await client.send(new DeleteObjectCommand({ Bucket: bucket, Key: key })) +} + +const extensionFor = (inputPath: string): string => { + const extension = extname(inputPath).toLowerCase() + return /^\.[a-z0-9]{1,5}$/.test(extension) ? extension : '' +} + +export const mimeTypeForPath = (inputPath: string): string => + MIME_BY_EXTENSION[extensionFor(inputPath)] ?? 'application/octet-stream' + +const requireStorage = (env: Record) => { + const storage = resolveS3StorageEnv(env) + if (!storage.enabled) { + throw new Error('Upload grande exige o storage S3 configurado; nenhuma mídia foi enviada.') + } + return storage +} + +export const isLargeMedia = ( + filesize: number, + thresholdBytes = LARGE_MEDIA_THRESHOLD_BYTES, +): boolean => filesize > thresholdBytes + +export const uploadLargeMedia = async ({ + inputPath, + filename: providedFilename, + env = process.env, + upload = defaultUpload, +}: { + inputPath: string + filename?: string + env?: Record + upload?: (args: S3UploadArgs) => Promise +}): Promise => { + const storage = requireStorage(env) + const { size } = await stat(inputPath) + if (size > MAX_LARGE_MEDIA_BYTES) { + throw new Error( + `Mídia de ${(size / 1024 / 1024 / 1024).toFixed(2)} GiB excede o limite de 5 GiB.`, + ) + } + + const filename = providedFilename ?? `web-${randomUUID()}${extensionFor(inputPath)}` + if (!/^[\w.-]+$/.test(filename)) { + throw new Error('Nome de arquivo de mídia grande inválido.') + } + const mimeType = mimeTypeForPath(inputPath) + const client = storageClient(storage) + await upload({ + client, + bucket: storage.bucket, + key: filename, + contentType: mimeType, + inputPath, + filesize: size, + }) + + return { + filename, + mimeType, + filesize: size, + url: `${storage.endpoint.replace(/\/$/, '')}/${storage.bucket}/${encodeURIComponent(filename)}`, + } +} + +export const deleteLargeMedia = async ({ + filename, + env = process.env, + remove = defaultDelete, +}: { + filename: string + env?: Record + remove?: (args: S3DeleteArgs) => Promise +}): Promise => { + const storage = requireStorage(env) + await remove({ + client: storageClient(storage), + bucket: storage.bucket, + key: filename, + }) +} diff --git a/src/utilities/speech/webSpeechIngest.ts b/src/utilities/speech/webSpeechIngest.ts index 362c4b6e..9643b24e 100644 --- a/src/utilities/speech/webSpeechIngest.ts +++ b/src/utilities/speech/webSpeechIngest.ts @@ -1,8 +1,8 @@ import 'server-only' -import { mkdtemp, readdir, readFile, rm } from 'node:fs/promises' +import { mkdtemp, readdir, readFile, rm, stat, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' -import { join } from 'node:path' +import { dirname, extname, join } from 'node:path' import type { Payload } from 'payload' import { @@ -30,6 +30,13 @@ import { } from '@/utilities/ai/deepInfraTranscribe' import { downloadUrlToFile } from '@/utilities/media/downloadToFile' import { ffmpegBinary, messageOf, runFfmpeg } from '@/utilities/media/ffmpeg' +import { + deleteLargeMedia, + isLargeMedia, + LARGE_MEDIA_THRESHOLD_BYTES, + uploadLargeMedia, + type LargeMediaUpload, +} from '@/utilities/media/largeS3Upload' import { downloadWithYtDlp, readYtDlpMetadata, @@ -37,7 +44,10 @@ import { YTDLP_INSTALL_HINT, type YtDlpMetadata, } from '@/utilities/media/ytdlp' -import { withPayloadTransaction } from '@/utilities/payloadTransaction' +import { + withPayloadTransaction, + type PayloadTransactionRequest, +} from '@/utilities/payloadTransaction' import { classifySpeech, type SpeechClassificationResult, @@ -93,6 +103,13 @@ export type WebSpeechAcquirer = ( tempDir: string, ) => Promise +type WebSpeechMediaMirror = (args: { + payload: Payload + req: PayloadTransactionRequest + alt: string + inputPath: string +}) => Promise<{ id: number; cleanup: () => Promise }> + export type WebSpeechTranscriber = ( file: Blob, options?: { filename?: string }, @@ -106,6 +123,7 @@ export type WebSpeechIngestDeps = { classify?: WebSpeechClassifier runFfmpeg?: typeof runFfmpeg downloadThumbnail?: (url: string, destinationPath: string) => Promise + mirrorMedia?: WebSpeechMediaMirror } class WebSpeechIngestError extends Error { @@ -275,6 +293,77 @@ const extractTranscript = async ({ return { segments, durationSeconds, asrSeconds } } +const noopCleanup = async (): Promise => {} + +type WebSpeechMediaMirrorOptions = { + thresholdBytes?: number + uploadLarge?: typeof uploadLargeMedia + removeLarge?: typeof deleteLargeMedia +} + +export const createWebSpeechMediaMirror = + ({ + thresholdBytes = LARGE_MEDIA_THRESHOLD_BYTES, + uploadLarge = uploadLargeMedia, + removeLarge = deleteLargeMedia, + }: WebSpeechMediaMirrorOptions = {}): WebSpeechMediaMirror => + async ({ payload, req, alt, inputPath }) => { + const { size } = await stat(inputPath) + if (!isLargeMedia(size, thresholdBytes)) { + const media = await payload.create({ + collection: INTERNET_SPEECH_MEDIA_SLUG, + data: { alt }, + filePath: inputPath, + req, + // Intentional bypass: the ingestion CLI is a trusted actor with no session. + overrideAccess: true, + }) + return { id: media.id, cleanup: noopCleanup } + } + + const placeholderPath = join(dirname(inputPath), `payload-placeholder${extname(inputPath)}`) + await writeFile(placeholderPath, Buffer.alloc(0)) + let filename: string | null = null + + try { + const media = await payload.create({ + collection: INTERNET_SPEECH_MEDIA_SLUG, + data: { alt }, + filePath: placeholderPath, + req, + // Intentional bypass: the placeholder row is owned by the trusted ingestion CLI. + overrideAccess: true, + }) + filename = media.filename ?? null + if (!filename) throw new Error('A mídia grande não recebeu um filename no S3.') + const uploadedFilename = filename + + const uploaded: LargeMediaUpload = await uploadLarge({ + inputPath, + filename: uploadedFilename, + }) + await payload.update({ + collection: INTERNET_SPEECH_MEDIA_SLUG, + id: media.id, + data: { filesize: uploaded.filesize, mimeType: uploaded.mimeType }, + req, + // Intentional bypass: the trusted ingestion CLI finalizes its own media row. + overrideAccess: true, + }) + return { + id: media.id, + cleanup: async () => { + await removeLarge({ filename: uploadedFilename }).catch(() => undefined) + }, + } + } catch (error) { + if (filename) await removeLarge({ filename }).catch(() => undefined) + throw error + } + } + +const defaultMirrorMedia = createWebSpeechMediaMirror() + /** * The artifact and the speech in ONE transaction (`req` threaded through the * upload and the upsert). Throws `WebSpeechIngestError('media')`; the caller @@ -291,6 +380,7 @@ const persistIngestedSpeech = async ({ durationSeconds, inputPath, thumbnail, + mirrorMedia, }: { payload: Payload finding: WebSpeechFinding @@ -302,19 +392,15 @@ const persistIngestedSpeech = async ({ durationSeconds: number | null inputPath: string thumbnail: number | null + mirrorMedia: WebSpeechMediaMirror }): Promise => { const alt = `${webSpeechPlatformLabel(finding.platform)} — ${metadata.title ?? speechAt}` + const cleanupRef: { current?: () => Promise } = {} try { await withPayloadTransaction(payload, async ({ req }) => { - const media = await payload.create({ - collection: INTERNET_SPEECH_MEDIA_SLUG, - data: { alt }, - filePath: inputPath, - req, - // Intentional bypass: the ingestion CLI is a trusted actor with no session. - overrideAccess: true, - }) + const mirrored = await mirrorMedia({ payload, req, alt, inputPath }) + cleanupRef.current = mirrored.cleanup await upsertSpeechBundle( payload, @@ -326,13 +412,14 @@ const persistIngestedSpeech = async ({ endSeconds: segment.endSeconds, text: segment.text, })), - mirroredMedia: media.id, + mirroredMedia: mirrored.id, thumbnail, }, { req }, ) }) } catch (error) { + await cleanupRef.current?.().catch(() => undefined) throw new WebSpeechIngestError('media', messageOf(error, MEDIA_FAILURE)) } } @@ -354,6 +441,7 @@ export const ingestWebSpeech = async ( const downloadThumbnail = deps.downloadThumbnail ?? ((url, destinationPath) => downloadUrlToFile({ url, destinationPath })) + const mirrorMedia = deps.mirrorMedia ?? defaultMirrorMedia let asrSeconds = 0 let tempDir: string | null = null @@ -457,6 +545,7 @@ export const ingestWebSpeech = async ( durationSeconds: extracted.durationSeconds, inputPath, thumbnail: thumbnailId ?? previousThumbnail, + mirrorMedia, }) } catch (error) { if (thumbnailId !== null) { diff --git a/tests/int/webSpeechIngest.int.spec.ts b/tests/int/webSpeechIngest.int.spec.ts index dd4236c4..7fe344b7 100644 --- a/tests/int/webSpeechIngest.int.spec.ts +++ b/tests/int/webSpeechIngest.int.spec.ts @@ -1,6 +1,7 @@ // @vitest-environment node -import { writeFile } from 'node:fs/promises' +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' import { join } from 'node:path' import type { Payload } from 'payload' import { getPayload } from 'payload' @@ -11,7 +12,9 @@ import { normalizeForSearch } from '@/lib/speechSearch' import { INTERNET_SPEECH_MEDIA_SLUG, type WebSpeechFinding } from '@/lib/webSpeech' import type { Speech } from '@/payload-types' import config from '@/payload.config' +import { withPayloadTransaction } from '@/utilities/payloadTransaction' import { + createWebSpeechMediaMirror, ingestWebSpeech, type WebSpeechAcquirer, type WebSpeechClassifier, @@ -383,6 +386,44 @@ describe('ingestWebSpeech (C215)', () => { expect(await findBySourceKey(sourceKey)).toBeNull() }) + it('updates a large media row through the direct S3 seam', async () => { + const dir = await mkdtemp(join(tmpdir(), 'web-speech-large-test-')) + const inputPath = join(dir, 'source.mp4') + await writeFile(inputPath, MP4_BYTES) + const mirror = createWebSpeechMediaMirror({ + thresholdBytes: 0, + uploadLarge: async ({ filename }) => ({ + filename: filename ?? 'source.mp4', + mimeType: 'video/mp4', + filesize: MP4_BYTES.length, + url: 'http://127.0.0.1:3900/teqo-media/source.mp4', + }), + removeLarge: async () => undefined, + }) + let mediaId = 0 + + try { + await withPayloadTransaction(payload, async ({ req }) => { + const mirrored = await mirror({ payload, req, alt: 'YouTube — fala', inputPath }) + mediaId = mirrored.id + await mirrored.cleanup() + }) + const media = await payload.findByID({ + collection: INTERNET_SPEECH_MEDIA_SLUG, + id: mediaId, + overrideAccess: true, + }) + expect(media).toMatchObject({ filesize: MP4_BYTES.length, mimeType: 'video/mp4' }) + } finally { + if (mediaId) { + await payload + .delete({ collection: INTERNET_SPEECH_MEDIA_SLUG, id: mediaId, overrideAccess: true }) + .catch(() => undefined) + } + await rm(dir, { recursive: true, force: true }) + } + }) + it('accepts a radio finding with a direct mediaUrl', async () => { const current = finding({ platform: 'radio', diff --git a/tests/unit/largeS3Upload.unit.spec.ts b/tests/unit/largeS3Upload.unit.spec.ts new file mode 100644 index 00000000..9c6c14d3 --- /dev/null +++ b/tests/unit/largeS3Upload.unit.spec.ts @@ -0,0 +1,88 @@ +// @vitest-environment node + +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { describe, expect, it } from 'vitest' + +import { + deleteLargeMedia, + isLargeMedia, + LARGE_MEDIA_THRESHOLD_BYTES, + mimeTypeForPath, + uploadLargeMedia, +} from '@/utilities/media/largeS3Upload' + +const s3Env = { + S3_BUCKET: 'teqo-media', + S3_ENDPOINT: 'http://127.0.0.1:3900', + S3_ACCESS_KEY_ID: 'key', + S3_SECRET_ACCESS_KEY: 'secret', +} + +const withTempFile = async (run: (path: string) => Promise): Promise => { + const dir = await mkdtemp(join(tmpdir(), 'large-s3-test-')) + try { + const path = join(dir, 'source.mp4') + await writeFile(path, Buffer.alloc(16)) + await run(path) + } finally { + await rm(dir, { recursive: true, force: true }) + } +} + +describe('large S3 media upload', () => { + it('switches only above the 2 GiB Payload buffer boundary', () => { + expect(isLargeMedia(LARGE_MEDIA_THRESHOLD_BYTES)).toBe(false) + expect(isLargeMedia(LARGE_MEDIA_THRESHOLD_BYTES + 1)).toBe(true) + }) + + it('maps the media extension to a safe content type', () => { + expect(mimeTypeForPath('/tmp/source.mp4')).toBe('video/mp4') + expect(mimeTypeForPath('/tmp/source.unknown')).toBe('application/octet-stream') + }) + + it('streams a large file to S3 and returns row metadata', async () => { + await withTempFile(async (path) => { + const calls: { key: string; contentType: string; filesize: number }[] = [] + const uploaded = await uploadLargeMedia({ + inputPath: path, + env: s3Env, + upload: async ({ key, contentType, filesize }) => { + calls.push({ key, contentType, filesize }) + }, + }) + + expect(uploaded.filename).toMatch(/^web-[0-9a-f-]+\.mp4$/) + expect(uploaded.mimeType).toBe('video/mp4') + expect(uploaded.filesize).toBe(16) + expect(uploaded.url).toContain('/teqo-media/') + expect(calls).toEqual([{ key: uploaded.filename, contentType: 'video/mp4', filesize: 16 }]) + }) + }) + + it('fails closed when S3 is not configured', async () => { + await withTempFile(async (path) => { + await expect( + uploadLargeMedia({ + inputPath: path, + env: {}, + upload: async () => undefined, + }), + ).rejects.toThrow('S3') + }) + }) + + it('deletes the exact object key on cleanup', async () => { + const keys: string[] = [] + await deleteLargeMedia({ + filename: 'web-test.mp4', + env: s3Env, + remove: async ({ key }) => { + keys.push(key) + }, + }) + expect(keys).toEqual(['web-test.mp4']) + }) +}) diff --git a/tests/unit/webSpeechMediaMirror.unit.spec.ts b/tests/unit/webSpeechMediaMirror.unit.spec.ts new file mode 100644 index 00000000..4eab29fc --- /dev/null +++ b/tests/unit/webSpeechMediaMirror.unit.spec.ts @@ -0,0 +1,69 @@ +// @vitest-environment node + +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import type { Payload } from 'payload' +import { describe, expect, it } from 'vitest' + +import type { PayloadTransactionRequest } from '@/utilities/payloadTransaction' +import { createWebSpeechMediaMirror } from '@/utilities/speech/webSpeechIngest' + +describe('web speech media mirror', () => { + it('overwrites the placeholder row through the large-media S3 seam', async () => { + const dir = await mkdtemp(join(tmpdir(), 'web-speech-mirror-test-')) + const inputPath = join(dir, 'source.mp4') + await writeFile(inputPath, Buffer.alloc(16)) + const created: { alt: string; filePath: string }[] = [] + const updated: { id: number; data: { filesize: number; mimeType: string } }[] = [] + const uploaded: { filename?: string; filesize?: number }[] = [] + const removed: string[] = [] + const payload = { + create: async (args: { data: { alt: string }; filePath: string }) => { + created.push({ alt: args.data.alt, filePath: args.filePath }) + return { id: 42, filename: 'source.mp4' } + }, + update: async (args: { id: number; data: { filesize: number; mimeType: string } }) => { + updated.push(args) + return { id: args.id } + }, + } as unknown as Payload + + try { + const mirror = createWebSpeechMediaMirror({ + thresholdBytes: 0, + uploadLarge: async ({ filename }) => { + uploaded.push({ filename, filesize: 16 }) + return { + filename: filename ?? 'source.mp4', + mimeType: 'video/mp4', + filesize: 16, + url: 'http://127.0.0.1:3900/teqo-media/source.mp4', + } + }, + removeLarge: async ({ filename }) => { + removed.push(filename) + }, + }) + const result = await mirror({ + payload, + req: {} as PayloadTransactionRequest, + alt: 'YouTube — fala', + inputPath, + }) + await result.cleanup() + + expect(created).toHaveLength(1) + expect(created[0].filePath).toContain('payload-placeholder.mp4') + expect(uploaded).toEqual([{ filename: 'source.mp4', filesize: 16 }]) + expect(updated[0]).toMatchObject({ + id: 42, + data: { filesize: 16, mimeType: 'video/mp4' }, + }) + expect(removed).toEqual(['source.mp4']) + } finally { + await rm(dir, { recursive: true, force: true }) + } + }) +})