diff --git a/src/features/sync/constant.ts b/src/features/sync/constant.ts index e5020c9..c7e9450 100644 --- a/src/features/sync/constant.ts +++ b/src/features/sync/constant.ts @@ -7,3 +7,9 @@ export const DBX_URL_PATH = { fileUpload: '/files/upload', fileDownload: '/files/download', } + +// Files above this use an upload session, not /files/upload. +export const DROPBOX_SINGLE_UPLOAD_MAX_BYTES = 150 * 1024 * 1024 + +// Bytes per upload-session append (a multiple of 4 MiB, per Dropbox). +export const DROPBOX_UPLOAD_CHUNK_BYTES = 16 * 1024 * 1024 diff --git a/src/features/sync/lib/Sync.service.ts b/src/features/sync/lib/Sync.service.ts index 34242dd..cb549bb 100644 --- a/src/features/sync/lib/Sync.service.ts +++ b/src/features/sync/lib/Sync.service.ts @@ -16,7 +16,11 @@ import { fileFolderSync, } from '@/db/schema/fileFolderSync.schema' import APIError from '@/errors/APIError' -import { DBX_URL_PATH } from '@/features/sync/constant' +import { + DBX_URL_PATH, + DROPBOX_SINGLE_UPLOAD_MAX_BYTES, + DROPBOX_UPLOAD_CHUNK_BYTES, +} from '@/features/sync/constant' import { MapFilesService } from '@/features/sync/lib/MapFiles.service' import type { AssemblyToDropboxSyncFilesPayload, @@ -982,27 +986,62 @@ export class SyncService extends AuthenticatedDropboxService { private async uploadFileInDropbox(file: CopilotFileRetrieve, path: string) { logger.info('SyncService#uploadFileInDropbox :: Uploading file to', path) - if (file.downloadUrl) { - // download file from Assembly - const resp = await fetch(file.downloadUrl) - // upload file to dropbox - const dbxResponse = await this.dbxClient.uploadFile({ - urlPath: DBX_URL_PATH.fileUpload, - filePath: path, - body: resp.body, - rootNamespaceId: z.string().parse(this.connectionToken.rootNamespaceId), - refreshToken: this.connectionToken.refreshToken, - }) - logger.info('SyncService#uploadFileInDropbox :: File uploaded to', path) - return { - dbxFileId: dbxResponse.id, - contentHash: dbxResponse.contentHash, - } + if (!file.downloadUrl) { + logger.error( + `SyncService#uploadFileInDropbox. Assembly file with Id: ${file.id} has no download url. Channel ID: ${file.channelId}`, + ) + throw new Error('File not found') } - logger.error( - `SyncService#uploadFileInDropbox. Assembly file with Id: ${file.id} has no download url. Channel ID: ${file.channelId}`, - ) - throw new Error('File not found') + + const downloadUrl = file.downloadUrl + const rootNamespaceId = z.string().parse(this.connectionToken.rootNamespaceId) + const refreshToken = this.connectionToken.refreshToken + + // Over the single-shot limit (or unknown size) goes through an upload session. + const needsSession = file.size === undefined || file.size > DROPBOX_SINGLE_UPLOAD_MAX_BYTES + + logger.info('SyncService#uploadFileInDropbox :: Upload method chosen', { + path, + method: needsSession ? 'upload session' : 'direct upload', + sizeBytes: file.size ?? null, + }) + + const dbxResponse = needsSession + ? await this.dbxClient.uploadFileSession({ + filePath: path, + // node-fetch's body is a Node stream, an async iterable at runtime. + getBody: async () => + (await this.fetchAssemblyBody(downloadUrl)) as unknown as AsyncIterable, + rootNamespaceId, + refreshToken, + chunkSize: DROPBOX_UPLOAD_CHUNK_BYTES, + expectedSize: file.size, + }) + : await this.dbxClient.uploadFile({ + urlPath: DBX_URL_PATH.fileUpload, + filePath: path, + body: await this.fetchAssemblyBody(downloadUrl), + rootNamespaceId, + refreshToken, + }) + + logger.info('SyncService#uploadFileInDropbox :: File uploaded to', path) + return { + dbxFileId: dbxResponse.id, + contentHash: dbxResponse.contentHash, + } + } + + // Fetches the Assembly body, failing on a non-ok response so an error page is + // never streamed to Dropbox as the file content. + private async fetchAssemblyBody(downloadUrl: string): Promise { + const resp = await fetch(downloadUrl) + if (!resp.ok || !resp.body) { + throw new Error( + `SyncService#uploadFileInDropbox. Assembly download failed with status ${resp.status}`, + ) + } + return resp.body } async removeChannelSyncMapping(channelSyncId: string) { diff --git a/src/features/sync/lib/__tests__/Sync.uploadRouting.test.ts b/src/features/sync/lib/__tests__/Sync.uploadRouting.test.ts new file mode 100644 index 0000000..92ebcde --- /dev/null +++ b/src/features/sync/lib/__tests__/Sync.uploadRouting.test.ts @@ -0,0 +1,116 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + DROPBOX_SINGLE_UPLOAD_MAX_BYTES, + DROPBOX_UPLOAD_CHUNK_BYTES, +} from '@/features/sync/constant' + +// The Assembly file body is fetched via node-fetch — mock it so no real request runs. +vi.mock('node-fetch', () => ({ default: vi.fn() })) + +// SyncService builds a DropboxClient + CopilotAPI and imports a Trigger.dev task at +// module load — stub all three so the service constructs in isolation. +vi.mock('@/lib/dropbox/DropboxClient', () => ({ + DropboxClient: class { + getDropboxClient() { + return {} + } + dbxAuthClient = { refreshAccessToken: vi.fn() } + uploadFile = vi.fn().mockResolvedValue({ id: 'single-id', contentHash: 'single-hash' }) + uploadFileSession = vi.fn().mockResolvedValue({ id: 'session-id', contentHash: 'session-hash' }) + }, +})) + +vi.mock('@/lib/copilot/CopilotAPI', () => ({ + CopilotAPI: class {}, + isCopilotApiError: () => false, +})) + +vi.mock('@/trigger/processFileSync', () => ({ + bidirectionalMasterSync: { trigger: vi.fn() }, +})) + +import fetch from 'node-fetch' +import { SyncService } from '@/features/sync/lib/Sync.service' + +const mockedFetch = vi.mocked(fetch) + +const user = { portalId: 'p', token: 't', copilot: {} } as never +const connectionToken = { accountId: 'a', refreshToken: 'r', rootNamespaceId: 'n' } as never + +type UploadInternals = { + uploadFileInDropbox( + file: unknown, + path: string, + ): Promise<{ dbxFileId: string; contentHash: string }> + dbxClient: { + uploadFile: ReturnType + uploadFileSession: ReturnType + } +} + +const fileWithSize = (size: number | undefined) => ({ + id: 'asm-1', + downloadUrl: 'https://assembly/download', + size, +}) + +let service: SyncService +let internals: UploadInternals + +beforeEach(() => { + service = new SyncService(user, connectionToken) + internals = service as unknown as UploadInternals + mockedFetch.mockResolvedValue({ ok: true, status: 200, body: {} } as never) +}) + +afterEach(() => { + vi.restoreAllMocks() +}) + +describe('uploadFileInDropbox routing', () => { + it('uses the single-shot upload when the file is at or below the limit', async () => { + const result = await internals.uploadFileInDropbox( + fileWithSize(DROPBOX_SINGLE_UPLOAD_MAX_BYTES), + '/root/small.txt', + ) + + expect(internals.dbxClient.uploadFile).toHaveBeenCalledTimes(1) + expect(internals.dbxClient.uploadFileSession).not.toHaveBeenCalled() + expect(result).toEqual({ dbxFileId: 'single-id', contentHash: 'single-hash' }) + }) + + it('uses an upload session when the file is over the limit', async () => { + const result = await internals.uploadFileInDropbox( + fileWithSize(DROPBOX_SINGLE_UPLOAD_MAX_BYTES + 1), + '/root/big.bin', + ) + + expect(internals.dbxClient.uploadFile).not.toHaveBeenCalled() + expect(internals.dbxClient.uploadFileSession).toHaveBeenCalledTimes(1) + expect(internals.dbxClient.uploadFileSession.mock.calls[0][0]).toMatchObject({ + filePath: '/root/big.bin', + chunkSize: DROPBOX_UPLOAD_CHUNK_BYTES, + rootNamespaceId: 'n', + refreshToken: 'r', + }) + expect(typeof internals.dbxClient.uploadFileSession.mock.calls[0][0].getBody).toBe('function') + expect(result).toEqual({ dbxFileId: 'session-id', contentHash: 'session-hash' }) + }) + + it('uses an upload session when the size is unknown', async () => { + await internals.uploadFileInDropbox(fileWithSize(undefined), '/root/unknown.bin') + + expect(internals.dbxClient.uploadFile).not.toHaveBeenCalled() + expect(internals.dbxClient.uploadFileSession).toHaveBeenCalledTimes(1) + }) + + it('throws instead of uploading when the Assembly download responds with a non-ok status', async () => { + mockedFetch.mockResolvedValue({ ok: false, status: 403, body: {} } as never) + + await expect( + internals.uploadFileInDropbox(fileWithSize(1024), '/root/small.txt'), + ).rejects.toThrow(/download/i) + + expect(internals.dbxClient.uploadFile).not.toHaveBeenCalled() + }) +}) diff --git a/src/lib/__tests__/logger.test.ts b/src/lib/__tests__/logger.test.ts index 17b851b..23d7130 100644 --- a/src/lib/__tests__/logger.test.ts +++ b/src/lib/__tests__/logger.test.ts @@ -4,17 +4,34 @@ import { logger } from '@/lib/logger' afterEach(() => { vi.restoreAllMocks() delete process.env.LOG_LEVEL + delete process.env.LOG_DEPTH }) describe('logger', () => { - it('caps object depth so nested payloads are not fully dumped', () => { + it('shows nested payload fields like opts and file inside a batch array', () => { const spy = vi.spyOn(console, 'info').mockImplementation(() => undefined) - logger.info({ a: { b: { c: { secret: 'deep-value' } } } }) + // Mirrors the batchTriggerAndWait item shape: [{ payload: { opts, file } }]. + logger.info([{ payload: { opts: { channelId: 'ch-1' }, file: { id: 'f-1' } } }]) const out = spy.mock.calls[0][0] as string - expect(out).not.toContain('deep-value') + expect(out).toContain("channelId: 'ch-1'") + expect(out).toContain("id: 'f-1'") + }) + + it('still caps extremely deep objects so a log line stays bounded', () => { + const spy = vi.spyOn(console, 'info').mockImplementation(() => undefined) + logger.info({ a: { b: { c: { d: { e: { secret: 'too-deep' } } } } } }) + const out = spy.mock.calls[0][0] as string + expect(out).not.toContain('too-deep') expect(out).toContain('[Object]') }) + it('honours LOG_DEPTH for deeper inspection', () => { + process.env.LOG_DEPTH = '6' + const spy = vi.spyOn(console, 'info').mockImplementation(() => undefined) + logger.info({ a: { b: { c: { d: { e: { secret: 'now-visible' } } } } } }) + expect(spy.mock.calls[0][0] as string).toContain('now-visible') + }) + it('truncates long arrays instead of dumping every element', () => { const spy = vi.spyOn(console, 'info').mockImplementation(() => undefined) logger.info(Array.from({ length: 100 }, (_, i) => i)) diff --git a/src/lib/copilot/types.ts b/src/lib/copilot/types.ts index 577d119..f244782 100644 --- a/src/lib/copilot/types.ts +++ b/src/lib/copilot/types.ts @@ -139,6 +139,7 @@ export const CopilotFileRetrieveSchema = z.object({ path: z.string(), status: z.string().optional(), downloadUrl: z.string().optional(), + size: z.number().optional(), previousAttributes: z .object({ name: z.string().optional(), diff --git a/src/lib/dropbox/DropboxClient.ts b/src/lib/dropbox/DropboxClient.ts index de90176..6bc5a19 100644 --- a/src/lib/dropbox/DropboxClient.ts +++ b/src/lib/dropbox/DropboxClient.ts @@ -10,8 +10,15 @@ import { type DropboxFileMetadata, DropboxFileMetadataSchema } from '@/lib/dropb import logger from '@/lib/logger' import { withRetry } from '@/lib/withRetry' +import { chunkStream } from '@/utils/chunkStream' import { dropboxArgHeader } from '@/utils/header' +const UPLOAD_SESSION_PATH = { + start: '/files/upload_session/start', + append: '/files/upload_session/append_v2', + finish: '/files/upload_session/finish', +} + export class DropboxClient { protected readonly clientInstance: Dropbox readonly dbxAuthClient: DropboxAuthClient @@ -59,7 +66,7 @@ export class DropboxClient { private async manualFetch( url: string, headers?: Record, - body?: NodeJS.ReadableStream | null, + body?: NodeJS.ReadableStream | Buffer | null, otherOptions?: Record, method: 'GET' | 'POST' | 'PUT' | 'DELETE' = 'POST', ) { @@ -259,6 +266,110 @@ export class DropboxClient { return DropboxFileMetadataSchema.parse(camelKeys(await response.json())) } + // Chunked upload session for files too large for /files/upload. getBody yields a + // fresh stream so a retry restarts the whole session. + async _uploadFileSession({ + filePath, + getBody, + rootNamespaceId, + refreshToken, + chunkSize, + expectedSize, + }: { + filePath: string + getBody: () => Promise> + rootNamespaceId: string + refreshToken: string + chunkSize: number + expectedSize?: number + }): Promise { + await this.dbxAuthClient.refreshAccessToken(refreshToken) + + const body = await getBody() + let sessionId: string | null = null + let offset = 0 + + for await (const chunk of chunkStream(body, chunkSize)) { + if (sessionId === null) { + const started = await this.sessionCall( + UPLOAD_SESSION_PATH.start, + { close: false }, + chunk, + rootNamespaceId, + filePath, + ) + sessionId = ((await started.json()) as { session_id: string }).session_id + } else { + await this.sessionCall( + UPLOAD_SESSION_PATH.append, + { cursor: { session_id: sessionId, offset }, close: false }, + chunk, + rootNamespaceId, + filePath, + ) + } + offset += chunk.length + } + + // No chunks: open an empty session so finish has a cursor. + if (sessionId === null) { + const started = await this.sessionCall( + UPLOAD_SESSION_PATH.start, + { close: false }, + Buffer.alloc(0), + rootNamespaceId, + filePath, + ) + sessionId = ((await started.json()) as { session_id: string }).session_id + } + + // Never finish a short stream — a truncated download must not commit a file. + if (expectedSize !== undefined && offset !== expectedSize) { + throw new Error( + `DropboxClient#uploadFileSession. Streamed ${offset} bytes but expected ${expectedSize} for ${filePath}`, + ) + } + + const finished = await this.sessionCall( + UPLOAD_SESSION_PATH.finish, + { + cursor: { session_id: sessionId, offset }, + commit: { path: filePath, mode: 'add', autorename: false }, + }, + Buffer.alloc(0), + rootNamespaceId, + filePath, + ) + return DropboxFileMetadataSchema.parse(camelKeys(await finished.json())) + } + + // One request in an upload session. Throws the shared Dropbox error on non-200. + private async sessionCall( + urlPath: string, + arg: Record, + body: Buffer, + rootNamespaceId: string, + filePath: string, + ): Promise { + const headers = { + Authorization: `Bearer ${this.dbxAuthClient.authInstance.getAccessToken()}`, + 'Dropbox-API-Path-Root': dropboxArgHeader({ + '.tag': 'namespace_id', + namespace_id: rootNamespaceId, + }), + 'Dropbox-API-Arg': dropboxArgHeader(arg), + 'Content-Type': 'application/octet-stream', + } + const response = await this.manualFetch(`${env.DROPBOX_API_URL}${urlPath}`, headers, body) + if (response.status !== httpStatus.OK) { + throw await this.buildDropboxResponseError( + response, + `DropboxClient#uploadFileSession. Failed to upload file: ${filePath}`, + ) + } + return response + } + private wrapWithRetry( fn: (...args: Args) => Promise, ): (...args: Args) => Promise { @@ -273,4 +384,6 @@ export class DropboxClient { getAllFilesFolders = this.wrapWithRetry(this._getAllFilesFolders) downloadFile = this.wrapWithRetry(this._downloadFile) uploadFile = this.wrapWithRetry(this._uploadFile) + // Retry restarts the whole session — `getBody` hands back a fresh stream each attempt. + uploadFileSession = this.wrapWithRetry(this._uploadFileSession) } diff --git a/src/lib/dropbox/__tests__/DropboxClient.uploadSession.test.ts b/src/lib/dropbox/__tests__/DropboxClient.uploadSession.test.ts new file mode 100644 index 0000000..8962cf8 --- /dev/null +++ b/src/lib/dropbox/__tests__/DropboxClient.uploadSession.test.ts @@ -0,0 +1,178 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +// The session upload streams via node-fetch — mock the module so no real request is made. +vi.mock('node-fetch', () => ({ default: vi.fn() })) + +import fetch from 'node-fetch' +import { DropboxClient } from '@/lib/dropbox/DropboxClient' + +const mockedFetch = vi.mocked(fetch) + +type Call = { url: string; arg: Record; bodyLength: number } + +// Reads the captured fetch calls back into a comparable shape. +const readCalls = (): Call[] => + mockedFetch.mock.calls.map(([url, init]) => { + const headers = (init as { headers: Record }).headers + const body = (init as { body?: Buffer }).body + return { + url: url as string, + arg: JSON.parse(headers['Dropbox-API-Arg']), + bodyLength: body ? body.length : 0, + } + }) + +const jsonResponse = (data: unknown, status = 200) => ({ + status, + json: async () => data, + text: async () => JSON.stringify(data), +}) + +const finishMetadata = { + name: 'big.bin', + path_display: '/root/big.bin', + id: 'id:big', + size: 5, + content_hash: 'hash-big', +} + +// biome-ignore lint/suspicious/useAwait: an async generator stands in for the streamed body +async function* source(bytes: number[]): AsyncGenerator { + yield Uint8Array.from(bytes) +} + +const makeClient = () => { + const client = new DropboxClient('refresh-token') + vi.spyOn(client.dbxAuthClient, 'refreshAccessToken').mockResolvedValue(undefined as never) + vi.spyOn(client.dbxAuthClient.authInstance, 'getAccessToken').mockReturnValue('access-token') + return client +} + +const uploadArgs = () => ({ + filePath: '/root/big.bin', + getBody: () => Promise.resolve(source([1, 2, 3, 4, 5])), + rootNamespaceId: 'ns-1', + refreshToken: 'refresh-token', + chunkSize: 2, + expectedSize: 5, +}) + +describe('DropboxClient#_uploadFileSession', () => { + beforeEach(() => { + mockedFetch.mockReset() + }) + + it('starts a session, appends each chunk at the right offset, then finishes', async () => { + mockedFetch + .mockResolvedValueOnce(jsonResponse({ session_id: 'sess-1' }) as never) // start + .mockResolvedValueOnce(jsonResponse({}) as never) // append 1 + .mockResolvedValueOnce(jsonResponse({}) as never) // append 2 + .mockResolvedValueOnce(jsonResponse(finishMetadata) as never) // finish + + const client = makeClient() + const result = await client._uploadFileSession(uploadArgs()) + + const calls = readCalls() + expect(calls.map((c) => c.url)).toEqual([ + expect.stringContaining('/files/upload_session/start'), + expect.stringContaining('/files/upload_session/append_v2'), + expect.stringContaining('/files/upload_session/append_v2'), + expect.stringContaining('/files/upload_session/finish'), + ]) + + // 5 bytes at chunk size 2 → chunks of 2, 2, 1. + expect(calls.map((c) => c.bodyLength)).toEqual([2, 2, 1, 0]) + + // Append cursors advance by the bytes already sent. + expect(calls[1].arg.cursor).toEqual({ session_id: 'sess-1', offset: 2 }) + expect(calls[2].arg.cursor).toEqual({ session_id: 'sess-1', offset: 4 }) + + // Finish commits at the total offset and the target path. + expect(calls[3].arg.cursor).toEqual({ session_id: 'sess-1', offset: 5 }) + expect(calls[3].arg.commit).toMatchObject({ + path: '/root/big.bin', + mode: 'add', + autorename: false, + }) + + expect(result).toEqual({ + name: 'big.bin', + pathDisplay: '/root/big.bin', + id: 'id:big', + size: 5, + contentHash: 'hash-big', + }) + }) + + it('throws without finishing when the streamed bytes do not match the expected size', async () => { + mockedFetch + .mockResolvedValueOnce(jsonResponse({ session_id: 'sess-1' }) as never) // start + .mockResolvedValueOnce(jsonResponse({}) as never) // append + .mockResolvedValueOnce(jsonResponse({}) as never) // append + + const client = makeClient() + // Body is 5 bytes but the caller expected 8 — a truncated/expired download. + await expect(client._uploadFileSession({ ...uploadArgs(), expectedSize: 8 })).rejects.toThrow( + /Streamed 5 bytes but expected 8/, + ) + + // finish must not run — no file is committed from a short body. + expect(readCalls().some((c) => c.url.includes('/finish'))).toBe(false) + }) + + it('throws the Dropbox error when a session call fails', async () => { + mockedFetch.mockResolvedValueOnce( + jsonResponse({ error_summary: 'too_many_requests' }, 429) as never, + ) + + const client = makeClient() + await expect(client._uploadFileSession(uploadArgs())).rejects.toThrow(/too_many_requests/) + }) + + // Each invocation must pull a fresh stream and open its own session, so retries are clean. + it('pulls a fresh stream from getBody and opens a new session on every invocation', async () => { + const getBody = () => Promise.resolve(source([1, 2, 3])) + const getBodySpy = vi.fn(getBody) + const oneCycle = [ + jsonResponse({ session_id: 'sess' }), // start + jsonResponse({}), // append + jsonResponse(finishMetadata), // finish + ] + mockedFetch + .mockResolvedValueOnce(oneCycle[0] as never) + .mockResolvedValueOnce(oneCycle[1] as never) + .mockResolvedValueOnce(oneCycle[2] as never) + .mockResolvedValueOnce(oneCycle[0] as never) + .mockResolvedValueOnce(oneCycle[1] as never) + .mockResolvedValueOnce(oneCycle[2] as never) + + const client = makeClient() + const args = { ...uploadArgs(), getBody: getBodySpy, expectedSize: 3 } + await client._uploadFileSession(args) + await client._uploadFileSession(args) + + expect(getBodySpy).toHaveBeenCalledTimes(2) + expect(readCalls().filter((c) => c.url.includes('/start'))).toHaveLength(2) + }) + + it('opens and finishes an empty session for a zero-byte file', async () => { + mockedFetch + .mockResolvedValueOnce(jsonResponse({ session_id: 'sess-empty' }) as never) // start + .mockResolvedValueOnce(jsonResponse({ ...finishMetadata, size: 0 }) as never) // finish + + const client = makeClient() + const result = await client._uploadFileSession({ + ...uploadArgs(), + getBody: () => Promise.resolve(source([])), + expectedSize: 0, + }) + + const calls = readCalls() + expect(calls.map((c) => c.url)).toEqual([ + expect.stringContaining('/files/upload_session/start'), + expect.stringContaining('/files/upload_session/finish'), + ]) + expect(calls[0].bodyLength).toBe(0) + expect(result.size).toBe(0) + }) +}) diff --git a/src/lib/logger.ts b/src/lib/logger.ts index 86e679e..cb328eb 100644 --- a/src/lib/logger.ts +++ b/src/lib/logger.ts @@ -24,19 +24,26 @@ const parseLevel = (value: string | undefined): LogLevel => { // Read per call so the level can be set without a rebuild (and stays testable). const minRank = (): number => LEVEL_RANK[parseLevel(process.env.LOG_LEVEL)] -// Bound object output so one log line can't dump a whole payload / file list. -// breakLength: Infinity keeps each log on one physical line so line-based log -// collectors don't split a wrapped object across entries. -const inspectOptions: util.InspectOptions = { - depth: 2, - colors: Boolean(process.stdout.isTTY), - maxArrayLength: 10, - maxStringLength: 512, - breakLength: Infinity, +// How many object levels to print. Deep enough to show nested payloads (e.g. a +// batched item's opts/file), still capped so a line can't dump everything. +const DEFAULT_DEPTH = 4 + +const parseDepth = (): number => { + const raw = Number(process.env.LOG_DEPTH) + return Number.isInteger(raw) && raw >= 0 && raw <= 10 ? raw : DEFAULT_DEPTH } +// breakLength: Infinity keeps each log on one physical line so line-based log +// collectors don't split a wrapped object across entries. function formatArg(arg: unknown): string { - return typeof arg === 'string' ? arg : util.inspect(arg, inspectOptions) + if (typeof arg === 'string') return arg + return util.inspect(arg, { + depth: parseDepth(), + colors: Boolean(process.stdout.isTTY), + maxArrayLength: 10, + maxStringLength: 512, + breakLength: Infinity, + }) } function loggerFactory(level: LogLevel): (...args: unknown[]) => void { diff --git a/src/utils/__tests__/chunkStream.test.ts b/src/utils/__tests__/chunkStream.test.ts new file mode 100644 index 0000000..9fa1e72 --- /dev/null +++ b/src/utils/__tests__/chunkStream.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from 'vitest' +import { chunkStream } from '@/utils/chunkStream' + +// Turns a list of byte-arrays into an async iterable, like a streamed body. +// biome-ignore lint/suspicious/useAwait: an async generator is the async iterable under test +async function* source(parts: number[][]): AsyncGenerator { + for (const part of parts) yield Uint8Array.from(part) +} + +async function collect(gen: AsyncGenerator): Promise { + const out: number[][] = [] + for await (const chunk of gen) out.push([...chunk]) + return out +} + +describe('chunkStream', () => { + it('yields nothing for an empty stream', async () => { + expect(await collect(chunkStream(source([]), 4))).toEqual([]) + }) + + it('yields a single short chunk when the stream is smaller than the chunk size', async () => { + expect(await collect(chunkStream(source([[1, 2, 3]]), 4))).toEqual([[1, 2, 3]]) + }) + + it('splits an exact multiple into full chunks with no remainder', async () => { + const chunks = await collect(chunkStream(source([[1, 2, 3, 4, 5, 6]]), 3)) + expect(chunks).toEqual([ + [1, 2, 3], + [4, 5, 6], + ]) + }) + + it('emits full chunks then a smaller final chunk for the remainder', async () => { + const chunks = await collect(chunkStream(source([[1, 2, 3, 4, 5]]), 2)) + expect(chunks).toEqual([[1, 2], [3, 4], [5]]) + }) + + it('coalesces many small source reads into full chunks across boundaries', async () => { + const parts = [[1], [2, 3], [4], [5, 6, 7]] + const chunks = await collect(chunkStream(source(parts), 3)) + expect(chunks).toEqual([[1, 2, 3], [4, 5, 6], [7]]) + }) + + it('rejects a non-positive chunk size', async () => { + await expect(collect(chunkStream(source([[1]]), 0))).rejects.toThrow(/chunk size/i) + }) +}) diff --git a/src/utils/chunkStream.ts b/src/utils/chunkStream.ts new file mode 100644 index 0000000..c1502a4 --- /dev/null +++ b/src/utils/chunkStream.ts @@ -0,0 +1,25 @@ +// Reads a byte stream into fixed-size buffers; the last one may be smaller. +export async function* chunkStream( + source: AsyncIterable, + chunkSize: number, +): AsyncGenerator { + if (chunkSize <= 0) throw new Error('chunkStream: chunk size must be greater than zero') + + let held: Buffer[] = [] + let heldBytes = 0 + + for await (const part of source) { + held.push(Buffer.from(part)) + heldBytes += part.byteLength + + while (heldBytes >= chunkSize) { + const merged = Buffer.concat(held, heldBytes) + yield merged.subarray(0, chunkSize) + const rest = merged.subarray(chunkSize) + held = rest.byteLength ? [rest] : [] + heldBytes = rest.byteLength + } + } + + if (heldBytes > 0) yield Buffer.concat(held, heldBytes) +} diff --git a/test/factories/copilot.ts b/test/factories/copilot.ts index 5824513..81bd740 100644 --- a/test/factories/copilot.ts +++ b/test/factories/copilot.ts @@ -13,6 +13,8 @@ export const copilotFileFactory = Factory.define(({ params name, object: params.object ?? ObjectType.FILE, path: `/${name}`, + // Small so uploads take the single-shot path by default; folders have no size. + size: isFolder ? undefined : 1024, } }) diff --git a/test/flows/assembly-to-dropbox.large-file.integration.test.ts b/test/flows/assembly-to-dropbox.large-file.integration.test.ts new file mode 100644 index 0000000..424bfc5 --- /dev/null +++ b/test/flows/assembly-to-dropbox.large-file.integration.test.ts @@ -0,0 +1,76 @@ +import { eq } from 'drizzle-orm' +import { HttpResponse } from 'msw' +import { describe, expect, it } from 'vitest' +import db from '@/db' +import { fileFolderSync } from '@/db/schema/fileFolderSync.schema' +import User from '@/lib/copilot/models/User.model' +import type { Token } from '@/lib/copilot/types' +import { initiateAssemblyToDropboxSync } from '@/trigger/processFileSync' +import { copilotDownloadableFactory, copilotListPage } from '../factories' +import { + mockAssemblyFileDownload, + mockCopilot, + mockDropboxGetMetadata, + mockDropboxUploadSession, +} from '../msw' +import { channelSeeder, dropboxConnectionSeeder } from '../seeders' + +// A file whose size Assembly doesn't report routes through the upload session (the +// same path a >150 MiB file takes). The single-shot upload is left unmocked, so if +// routing picked it the request would trip msw's onUnhandledRequest: 'error'. Keeping +// the body small lets one chunk exercise the real start -> finish wiring end-to-end; +// the >150 MiB decision and the byte-count guard are covered by unit tests. +describe('initial sync: Assembly -> Dropbox via upload session', () => { + it('uploads a file with no reported size through an upload session', async () => { + const accountId = 'acc-big' + const rootNamespaceId = 'ns-big' + const refreshToken = 'rt-big' + const connection = await dropboxConnectionSeeder.create({ + accountId, + rootNamespaceId, + refreshToken, + }) + const channel = await channelSeeder.create({ + portalId: connection.portalId, + dbxRootPath: '/root', + }) + const channelId = channel.assemblyChannelId + + const bigFile = copilotDownloadableFactory.build({ + path: 'big.txt', + channelId, + size: undefined, // unknown size -> session path + }) + + mockCopilot('/v1/files', () => HttpResponse.json(copilotListPage([bigFile]))) + mockDropboxGetMetadata({}) // path not found yet -> upload + const session = mockDropboxUploadSession() + mockAssemblyFileDownload() // small body, fetched from file.downloadUrl + + const user = new User('test-token', { workspaceId: connection.portalId } as Token) + await initiateAssemblyToDropboxSync.triggerAndWait({ + dbxRootPath: '/root', + assemblyChannelId: channelId, + connectionToken: { refreshToken, accountId, rootNamespaceId }, + user, + }) + + // The session was driven (start + finish), not the single-shot upload. + expect(session.calls.start).toBe(1) + expect(session.calls.finish).toBe(1) + + const [row] = await db + .select() + .from(fileFolderSync) + .where(eq(fileFolderSync.channelSyncId, channel.id)) + expect(row).toMatchObject({ + object: 'file', + itemPath: '/big.txt', + assemblyFileId: bigFile.id, + dbxFileId: 'id:dbx:/root/big.txt', // committed by finish at the target path + contentHash: 'hash-file', + pendingAction: null, + deletedAt: null, + }) + }) +}) diff --git a/test/msw/write.ts b/test/msw/write.ts index a224628..45dbb3c 100644 --- a/test/msw/write.ts +++ b/test/msw/write.ts @@ -93,6 +93,31 @@ export function mockDropboxUpload( ) } +// Upload session: start -> append_v2 (per chunk) -> finish. Returns call counters so a +// test can assert the session path (not single-shot) was taken. finish carries the +// target path under commit.path and its body parses like a normal upload. +export function mockDropboxUploadSession( + resolve: (filePath: string) => DropboxMeta = (path) => + dropboxFileMetadata({ path_display: path, id: `id:dbx:${path}` }), +): { calls: { start: number; append: number; finish: number } } { + const calls = { start: 0, append: 0, finish: 0 } + mockDropboxContent('/files/upload_session/start', () => { + calls.start++ + return HttpResponse.json({ session_id: 'session:test' }) + }) + mockDropboxContent('/files/upload_session/append_v2', () => { + calls.append++ + return new HttpResponse(null, { status: 200 }) + }) + mockDropboxContent('/files/upload_session/finish', ({ request }) => { + calls.finish++ + const arg = request.headers.get('Dropbox-API-Arg') + const commitPath = arg ? (JSON.parse(arg) as { commit: { path: string } }).commit.path : '' + return HttpResponse.json(resolve(commitPath)) + }) + return { calls } +} + // Download: body + Dropbox-API-Result header carrying the size (client reads size, not Content-Length). export function mockDropboxDownload( bodyByPath: Record,