Skip to content

fix: stream provider media uploads instead of buffering files in memory #1835

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nevo-david merged 5 commits into from
Aug 4, 2026
+634 −415
Merged

fix: stream provider media uploads instead of buffering files in memory #1835

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
9980825
fix: stream provider media uploads instead of buffering files in memory
nevo-david Aug 4, 2026
1b121d0
fix: address AI review findings on streamed media uploads
nevo-david Aug 4, 2026
c2944e1
fix: address second-round review findings on streamed uploads
nevo-david Aug 4, 2026
b0bb1c8
fix: guard the Reddit S3 upload response parse
nevo-david Aug 4, 2026
4013c1a
fix: harden streamed uploads after a full audit against origin/main
nevo-david Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
158 changes: 158 additions & 0 deletions libraries/nestjs-libraries/src/integrations/social.abstract.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@ import { ApplicationFailure } from '@temporalio/activity';
import { readOrFetch } from '@gitroom/helpers/utils/read.or.fetch';
import { getSsrfSafeDispatcher } from '@gitroom/nestjs-libraries/dtos/webhooks/ssrf.safe.dispatcher';
import sharp from 'sharp';
import { createReadStream, statSync } from 'fs';
import { Readable } from 'stream';

export type ValidityMedia = {
path: string;
Expand Down Expand Up @@ -173,6 +175,162 @@ export abstract class SocialAbstract {
return { width, height };
}

// Resolves the total byte size of the media without loading it into memory:
// a HEAD request for remote URLs, statSync for local files.
protected async mediaSize(path: string, identifier = ''): Promise<number> {
if (path.indexOf('http') === 0) {
// the media path is user-influenced, keep the SSRF-safe dispatcher that
// this.fetch applies to every other outbound request. identity encoding
// so content-length matches the bytes a later GET actually streams
// (fetch transparently decompresses encoded bodies).
const head = await fetch(path, {
method: 'HEAD',
headers: { 'accept-encoding': 'identity' },
dispatcher: getSsrfSafeDispatcher(),
} as any);
const length = Number(head.headers.get('content-length'));
// A failed HEAD can still carry a content-length (of the error body),
// and a zero/NaN size would poison chunk-count math downstream.
if (!head.ok || !Number.isFinite(length) || length <= 0) {
throw new BadBody(
identifier,
'{}',
Buffer.from('{}'),
'Could not determine the media size for upload'
);
}
return length;
}

return statSync(path).size;
}
Comment on lines +178 to +206

@coderabbitai coderabbitai Bot Aug 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Inspect how media paths are produced and whether relative paths reach providers.
set -euo pipefail

fd -t f 'social.abstract.ts' libraries | xargs -r rg -n 'readOrFetch|FRONTEND_URL|statSync|mediaSize|mediaChunk'

# Find the readOrFetch helper and its path handling.
rg -n -C 8 'export (const|function|async function) readOrFetch' libraries

# Find where media paths are assigned before reaching providers.
rg -n -C 5 "indexOf\('http'\)" libraries apps

Repository: gitroomhq/postiz-app

Length of output: 19327

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== social.abstract.ts relevant methods =="
sed -n '155,235p' libraries/nestjs-libraries/src/integrations/social.abstract.ts

echo
echo "== call sites for mediaSize/mediaChunk =="
rg -n -C 8 '\.(mediaSize|mediaChunk|getImageDimensions)\(' libraries/nestjs-libraries/src/integrations/social || true

echo
echo "== provider-specific media path handling snippets =="
for f in libraries/nestjs-libraries/src/integrations/social/tumblr.provider.ts \
         libraries/nestjs-libraries/src/integrations/social/linkedin.provider.ts \
         libraries/nestjs-libraries/src/integrations/social/discord.provider.ts \
         libraries/nestjs-libraries/src/integrations/social/reddit.provider.ts \
         libraries/nestjs-libraries/src/integrations/social/x.provider.ts; do
  if [ -f "$f" ]; then
    echo "--- $f ---"
    rg -n -C 10 'mediaSize|mediaChunk|getImageDimensions|getMediaUrl|media\.path|m\.path|path.*media' "$f" || true
  fi
done

echo
echo "== media file model / assignments =="
rg -n -C 6 'model (Media|media)|path.*String|upload|multipart|fieldValue|UploadService|uploadTo' libraries apps | head -n 240

Repository: gitroomhq/postiz-app

Length of output: 50376

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== candidate uploads/services/controllers =="
git ls-files | rg 'upload|asset|media|post|posts.*service|controller' | rg '\.ts$' | head -n 200

echo
echo "== focus: media upload path implementations =="
for f in $(git ls-files | rg 'upload|assets*|post' | rg 'providers|services|controllers|repository|schema' | grep '\.ts$' | head -n 80); do
  case "$f" in
    *uploads*|*asset*|*upload*|*post*)
      echo "--- $f ---"
      rg -n -C 4 'upload-directory|uploadDirectory|uploadDir|MEDIA_PATH|path:|filename|originalname|mimetype|files\\(' "$f" || true
      ;;
  esac
done

echo
echo "== PostDetails/media DTO definitions =="
rg -n -C 6 "PostDetails|interface .*Media|type .*Media|path: string|files\\(|express-multer|Multer" libraries apps | head -n 260

Repository: gitroomhq/postiz-app

Length of output: 34687

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== local.storage.ts =="
cat -n libraries/nestjs-libraries/src/upload/local.storage.ts | sed -n '1,220p'

echo
echo "== media.controller.ts =="
cat -n apps/backend/src/api/routes/media.controller.ts | sed -n '1,260p'

echo
echo "== posts.service.ts updateMedia paths =="
cat -n libraries/nestjs-libraries/src/database/prisma/posts/posts.service.ts | sed -n '330,430p'

echo
echo "== create media paths in posts.service.ts =="
rg -n -C 4 'create\\(|createMany\\(|create\\(\\{|media' libraries/nestjs-libraries/src/database/prisma/posts/posts.service.ts | sed -n '1,220p'

Repository: gitroomhq/postiz-app

Length of output: 16653

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== media.repository.ts =="
cat -n libraries/nestjs-libraries/src/database/prisma/media/media.repository.ts | sed -n '1,220p'

echo
echo "== media.service.ts =="
cat -n libraries/nestjs-libraries/src/database/prisma/media/media.service.ts | sed -n '1,240p'

echo
echo "== posts.service.ts create/update calls around media =="
cat -n libraries/nestjs-libraries/src/database/prisma/posts/posts.service.ts | sed -n '620,750p'
cat -n libraries/nestjs-libraries/src/database/prisma/posts/posts.service.ts | sed -n '750,830p'

echo
echo "== provider upload functions around raw paths =="
cat -n libraries/nestjs-libraries/src/integrations/social/x.provider.ts | sed -n '490,525p'
cat -n libraries/nestjs-libraries/src/integrations/social/tiktok.provider.ts | sed -n '790,830p'
cat -n libraries/nestjs-libraries/src/integrations/social/reddit.provider.ts | sed -n '160,190p'
cat -n libraries/nestjs-libraries/src/integrations/social/discord.provider.ts | sed -n '145,175p'
cat -n libraries/nestjs-libraries/src/integrations/social/linkedin.provider.ts | sed -n '255,280p'

Repository: gitroomhq/postiz-app

Length of output: 23667

Resolve non-HTTP media paths before using them with the filesystem.

mediaSize and mediaChunk use statSync / createReadStream for any path that does not start with http, but providers pass raw media paths such as item.path or media.path. Local uploads are stored with URLs like /uploads/year/month/day/r…, not absolute filesystem paths, so these calls can resolve relative to the worker cwd and fail. Resolve these cases like getImageDimensions/updateMedia use process.env.UPLOAD_DIRECTORY for non-HTTP paths.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@libraries/nestjs-libraries/src/integrations/social.abstract.ts` around lines
177 - 202, Update mediaSize and the related mediaChunk filesystem access to
resolve non-HTTP media paths through process.env.UPLOAD_DIRECTORY, matching the
existing getImageDimensions and updateMedia behavior. Preserve direct handling
for HTTP URLs and ensure statSync and createReadStream receive the resolved
filesystem path rather than the raw provider path.


// Reads a single [start, end] byte range into memory. Used by providers whose
// chunked-upload APIs require a Buffer per segment: only one small chunk is
// resident at a time, never the whole file.
protected async mediaChunk(
path: string,
start: number,
end: number,
identifier = ''
): Promise<Buffer> {
if (path.indexOf('http') === 0) {
const response = await fetch(path, {
headers: {
Range: `bytes=${start}-${end}`,
'accept-encoding': 'identity',
},
dispatcher: getSsrfSafeDispatcher(),
} as any);
// Anything but 206 means the server ignored the Range header: buffering
// response.body here would silently load the whole file into memory and
// upload corrupted chunks.
if (response.status !== 206) {
throw new BadBody(
identifier,
'{}',
Buffer.from('{}'),
`Media server did not honor the range request (status ${response.status})`
);
}
Comment thread
Copilot marked this conversation as resolved.
return Buffer.from(await response.arrayBuffer());
}

return new Promise((resolve, reject) => {
const chunks: Buffer[] = [];
createReadStream(path, { start, end })
.on('data', (chunk) => chunks.push(chunk as Buffer))
.on('end', () => resolve(Buffer.concat(chunks)))
.on('error', reject);
});
}

// Opens the media as a Node stream. The media path is user-influenced, so
// remote URLs go through the same SSRF-safe dispatcher as every other
// outbound request - never fetch these with plain axios/fetch.
protected async mediaStream(
path: string,
identifier = ''
): Promise<Readable> {
if (path.indexOf('http') !== 0) {
return createReadStream(path);
}

// identity encoding so the streamed byte count matches the size mediaSize
// reported - a decompressed body would overflow any declared length.
const response = await fetch(path, {
headers: { 'accept-encoding': 'identity' },
dispatcher: getSsrfSafeDispatcher(),
} as any);

if (!response.ok || !response.body) {
throw new BadBody(
identifier,
'{}',
Buffer.from('{}'),
'Could not read the media for upload'
);
}

return Readable.fromWeb(response.body as any);
}

// Streamed request bodies can't be replayed by this.fetch's retry, so
// providers wrap streamed uploads in a factory that rebuilds the request
// (re-opening its source streams) on every attempt. Errors carrying a
// `response` (axios errors, or a thrown { response: { status, data } })
// go through the same retry and handleErrors classification rules as
// this.fetch; anything else is rethrown untouched so Temporal keeps its
// usual retry behavior.
protected async runStreamedUpload<T>(
func: () => Promise<T>,
identifier = '',
totalRetries = 0
): Promise<T> {
try {
return await func();
} catch (err: any) {
if (!err?.response) {
throw err;
}
Comment on lines +292 to +295

@sentry sentry Bot Aug 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bug: Errors thrown when fetching media for runStreamedUpload lack a .response property, which bypasses the function's retry logic and prevents retries on transient network failures.
Severity: HIGH

Suggested Fix

When a media fetch fails with a non-ok status, throw an error object that includes the response property, similar to how axios errors are structured. This will allow the catch block in runStreamedUpload to correctly identify it as a retryable error.

Prompt for AI Agent 
Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: libraries/nestjs-libraries/src/integrations/social.abstract.ts#L292-L295

Potential issue: When fetching media from a source URL for providers like Mastodon, a
`new Error(...)` is thrown if the response status is not ok. This error object lacks a
`.response` property. The `runStreamedUpload` function, which wraps this logic, is
designed to only retry on errors that contain an `err?.response` property. As a result,
transient failures during media fetching (e.g., 429, 500, 503 status codes) will not be
retried, causing the upload to fail immediately instead of recovering.


const status = err.response.status || 500;
const data = err.response.data;
// '|| {}' / "|| '{}'" match this.fetch, which never hands handleErrors
// an empty string.
const json =
(typeof data === 'string' ? data : safeStringify(data || {})) || '{}';
const handleError = this.handleErrors(json, status);

if (
totalRetries <= 2 &&
(status === 429 ||
(status === 500 && !handleError) ||
handleError?.type === 'retry' ||
json.includes('rate_limit_exceeded') ||
json.includes('Rate limit'))
) {
await timer(5000);
return this.runStreamedUpload(func, identifier, totalRetries + 1);
Comment on lines +306 to +314

@sentry sentry Bot Aug 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bug: The retry condition in runStreamedUpload (totalRetries <= 2) allows 4 attempts, while this.fetch (totalRetries > 2) allows 3, contradicting the stated goal of matching behavior.
Severity: MEDIUM

Suggested Fix

Change the retry condition in runStreamedUpload at line 306 from totalRetries <= 2 to totalRetries < 2 or a similar condition that results in 3 total attempts, aligning it with the logic in this.fetch. This will make the retry behavior consistent.

Prompt for AI Agent 
Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: libraries/nestjs-libraries/src/integrations/social.abstract.ts#L306-L314

Potential issue: The retry logic in `runStreamedUpload` uses the condition `totalRetries
<= 2`, which results in 4 total attempts (initial call + 3 retries). This is
inconsistent with the `this.fetch` method, which uses `totalRetries > 2` and allows only
3 total attempts. This discrepancy violates the stated intent in the code comments and
PR description to maintain the "same retry... rules as this.fetch". This can lead to
different behavior for transient errors, adding an extra 5-second delay and an
unnecessary retry attempt for providers using `runStreamedUpload`.

}

if (
(status === 401 &&
(handleError?.type === 'refresh-token' || !handleError)) ||
handleError?.type === 'refresh-token'
) {
throw new RefreshToken(identifier, json, '{}', handleError?.value);
}

throw new BadBody(
identifier,
json,
'{}',
handleError?.value || 'Unknown Error'
);
}
}

public async mention(
token: string,
d: { query: string },
Expand Down
48 changes: 32 additions & 16 deletions libraries/nestjs-libraries/src/integrations/social/bluesky.provider.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import dayjs from 'dayjs';
import { Integration } from '@prisma/client';
import { AuthService } from '@gitroom/helpers/auth/auth.service';
import { isSafePublicHttpsUrl } from '@gitroom/nestjs-libraries/dtos/webhooks/webhook.url.validator';
import { getSsrfSafeDispatcher } from '@gitroom/nestjs-libraries/dtos/webhooks/ssrf.safe.dispatcher';
import sharp from 'sharp';
import { Plug } from '@gitroom/helpers/decorators/plug.decorator';
import { timer } from '@gitroom/helpers/utils/timer';
Expand Down Expand Up @@ -74,22 +75,35 @@ async function uploadVideo(
exp: Date.now() / 1000 + 60 * 30, // 30 minutes
});

async function downloadVideo(
url: string
): Promise<{ video: Buffer; size: number }> {
const response = await fetch(url);
if (!response.ok) {
throw new Error(`Failed to fetch video: ${response.statusText}`);
}
const arrayBuffer = await response.arrayBuffer();
const video = Buffer.from(arrayBuffer);
const size = video.length;
return { video, size };
// The video is never buffered in memory: the size comes from a HEAD request
// and the bytes are streamed straight from the source into the upload.
const headResponse = await fetch(videoPath, {
method: 'HEAD',
// identity encoding so content-length matches the bytes the GET streams
headers: { 'accept-encoding': 'identity' },
// @ts-ignore - undici-only option; blocks SSRF to internal IPs
dispatcher: getSsrfSafeDispatcher(),
});
const videoSize = Number(headResponse.headers.get('content-length') || 0);
if (!headResponse.ok || !videoSize) {
throw new BadBody(
'bluesky',
'{}',
{} as any,
'Could not determine the video size for Bluesky upload'
);
}

const video = await downloadVideo(videoPath);
const videoResponse = await fetch(videoPath, {
headers: { 'accept-encoding': 'identity' },
// @ts-ignore - undici-only option; blocks SSRF to internal IPs
dispatcher: getSsrfSafeDispatcher(),
});
if (!videoResponse.ok || !videoResponse.body) {
throw new Error(`Failed to fetch video: ${videoResponse.statusText}`);
}

console.log('Downloaded video', videoPath, video.size);
console.log('Uploading video', videoPath, videoSize);

const uploadUrl = new URL(
'https://video.bsky.app/xrpc/app.bsky.video.uploadVideo'
Expand All @@ -102,10 +116,12 @@ async function uploadVideo(
headers: {
Authorization: `Bearer ${serviceAuth.token}`,
'Content-Type': 'video/mp4',
'Content-Length': video.size.toString(),
'Content-Length': videoSize.toString(),
},
body: video.video,
});
body: videoResponse.body,
// Required by undici when streaming a request body.
duplex: 'half',
} as any);

const jobStatus = (await uploadResponse.json()) as AppBskyVideoDefs.JobStatus;
console.log('JobId:', jobStatus.jobId);
Expand Down
Loading
Loading