fix(C225-large): espelha mídia web acima de 2 GiB

This commit is contained in:
Francisco Solla 2026-09-25 23:33:47 -03:00
parent 2c65cf6ef0
commit d6499681dc
9 changed files with 489 additions and 17 deletions

View file

@ -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.

View file

@ -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.

View file

@ -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",

View file

@ -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

View file

@ -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<string, string> = {
'.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<void> => {
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<void> => {
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<string, string | undefined>) => {
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<string, string | undefined>
upload?: (args: S3UploadArgs) => Promise<void>
}): Promise<LargeMediaUpload> => {
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<string, string | undefined>
remove?: (args: S3DeleteArgs) => Promise<void>
}): Promise<void> => {
const storage = requireStorage(env)
await remove({
client: storageClient(storage),
bucket: storage.bucket,
key: filename,
})
}

View file

@ -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<WebSpeechAcquisition>
type WebSpeechMediaMirror = (args: {
payload: Payload
req: PayloadTransactionRequest
alt: string
inputPath: string
}) => Promise<{ id: number; cleanup: () => Promise<void> }>
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<void>
mirrorMedia?: WebSpeechMediaMirror
}
class WebSpeechIngestError extends Error {
@ -275,6 +293,77 @@ const extractTranscript = async ({
return { segments, durationSeconds, asrSeconds }
}
const noopCleanup = async (): Promise<void> => {}
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<void> => {
const alt = `${webSpeechPlatformLabel(finding.platform)} — ${metadata.title ?? speechAt}`
const cleanupRef: { current?: () => Promise<void> } = {}
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) {

View file

@ -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',

View file

@ -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<void>): Promise<void> => {
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'])
})
})

View file

@ -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 })
}
})
})