fix: revert back uploads to uint8array and stream

Author: hugomrdias

docs/src/content/docs/developer-guides/storage/storage-context.mdx

@@ -231,9 +231,9 @@ const storageContext = await synapse.storage.createContext({
 const llmModel = "sonnnet-4.5";
 const conversationId = "1234567890";
 
-const data = new Blob(["Deep research on decentralization..."])
+const data = new TextEncoder().encode("Deep research on decentralization...")
 
-const preflight = await storageContext.preflightUpload(data.size);
+const preflight = await storageContext.preflightUpload(data.length);
 
 console.log("Estimated costs:", preflight.estimatedCost);
 console.log("Allowance sufficient:", preflight.allowanceCheck.sufficient);

docs/src/content/docs/developer-guides/storage/storage-operations.mdx

@@ -53,9 +53,9 @@ import { privateKeyToAccount } from 'viem/accounts'
 
 const synapse = Synapse.create({ account: privateKeyToAccount('0x...') });
 // ---cut---
-const file = new Blob([new Uint8Array([1, 2, 3, 4, 5])]);
+const data = new Uint8Array([1, 2, 3, 4, 5]);
 
-const result = await synapse.storage.upload(file);
+const result = await synapse.storage.upload(data);
 const downloaded = await synapse.storage.download(result.pieceCid);
 
 console.log("Uploaded:", result.pieceCid);
@@ -71,7 +71,7 @@ Add metadata to organize uploads and enable faster data set reuse - SDK will reu
 import { Synapse } from "@filoz/synapse-sdk";
 import { privateKeyToAccount } from 'viem/accounts'
 
-const data = new Blob([new Uint8Array([1, 2, 3, 4, 5])]);
+const data = new Uint8Array([1, 2, 3, 4, 5]);
 const synapse = Synapse.create({ account: privateKeyToAccount('0x...') });
 // ---cut---
 const context = await synapse.storage.createContext({

docs/src/content/docs/getting-started/index.mdx

@@ -90,12 +90,12 @@ async function main() {
   console.log(`✅ USDFC deposit and Warm Storage service approval successful!`);
 
   // 3) Upload
-  const file = new Blob([
+  const file = new TextEncoder().encode(
     `🚀 Welcome to decentralized storage on Filecoin Onchain Cloud! 
     Your data is safe here. 
     🌍 You need to make sure to meet the minimum size 
     requirement of 127 bytes per upload.`
-  ]);
+  );
   const { pieceCid, size } = await synapse.storage.upload(file)
   console.log(`✅ Upload complete!`);
   console.log(`PieceCID: ${pieceCid}`);
@@ -179,12 +179,12 @@ import { privateKeyToAccount } from 'viem/accounts'
 const synapse = Synapse.create({ account: privateKeyToAccount('0x...') })
 // ---cut---
 // Upload data - SDK automatically selects provider and creates data set if needed
-const file = new Blob([
+const file = new TextEncoder().encode(
     `🚀 Welcome to decentralized storage on Filecoin Onchain Cloud! 
     Your data is safe here. 
     🌍 You need to make sure to meet the minimum size 
     requirement of 127 bytes per upload.`
-]);
+);
 const { pieceCid } = await synapse.storage.upload(file);
 
 // Download data from any provider that has it
@@ -226,12 +226,12 @@ For more control over provider selection and data set management:
 import { Synapse } from "@filoz/synapse-sdk";
 import { privateKeyToAccount } from 'viem/accounts'
 const synapse = Synapse.create({ account: privateKeyToAccount('0x...') })
-const file = new Blob([
+const file = new TextEncoder().encode(
     `🚀 Welcome to decentralized storage on Filecoin Onchain Cloud! 
     Your data is safe here. 
     🌍 You need to make sure to meet the minimum size 
     requirement of 127 bytes per upload.`
-]);
+);
 // ---cut---
 // Create a storage context with specific provider
 const context = await synapse.storage.createContext({

examples/cli/src/commands/upload.ts

@@ -1,8 +1,8 @@
+import { open } from 'node:fs/promises'
 import path from 'node:path'
 import * as p from '@clack/prompts'
 import { createPieceUrlPDP } from '@filoz/synapse-core/utils'
 import { Synapse } from '@filoz/synapse-sdk'
-import { openLazyFile } from '@remix-run/fs'
 import { type Command, command } from 'cleye'
 import { privateKeyClient } from '../client.ts'
 import { globalFlags } from '../flags.ts'
@@ -41,7 +41,7 @@ export const upload: Command = command(
 
     const filePath = argv._.requiredPath
     const absolutePath = path.resolve(filePath)
-    const file = openLazyFile(absolutePath)
+    const fileHandle = await open(absolutePath)
 
     try {
       const synapse = new Synapse({
@@ -63,7 +63,8 @@ export const upload: Command = command(
         },
       })
 
-      await context.upload(file, {
+      const data = fileHandle.readableWebStream()
+      await context.upload(data, {
         metadata: {
           name: path.basename(absolutePath),
         },

packages/synapse-core/src/mocks/pdp.ts

@@ -25,7 +25,7 @@ export interface PieceMetadataCapture {
 }
 
 export function createAndAddPiecesHandler(txHash: Hex, options: PDPMockOptions = {}) {
-  const baseUrl = options.baseUrl ?? 'http://pdp.local'
+  const baseUrl = options.baseUrl ?? 'https://pdp.example.com'
   return http.post(`${baseUrl}/pdp/data-sets/create-and-add`, () => {
     return new HttpResponse(null, {
       status: 201,
@@ -49,7 +49,7 @@ export function dataSetCreationStatusHandler(
   },
   options: PDPMockOptions = {}
 ) {
-  const baseUrl = options.baseUrl ?? 'http://pdp.local'
+  const baseUrl = options.baseUrl ?? 'https://pdp.example.com'
 
   return http.get(`${baseUrl}/pdp/data-sets/created/:txHash`, ({ params }) => {
     if (params.txHash !== txHash) {
@@ -69,7 +69,7 @@ export function pieceAdditionStatusHandler(
   response: any,
   options: PDPMockOptions = {}
 ) {
-  const baseUrl = options.baseUrl ?? 'http://pdp.local'
+  const baseUrl = options.baseUrl ?? 'https://pdp.example.com'
 
   return http.get(`${baseUrl}/pdp/data-sets/:id/pieces/added/:txHash`, ({ params }) => {
     if (params.id !== dataSetId.toString() || params.txHash !== txHash) {
@@ -84,7 +84,7 @@ export function pieceAdditionStatusHandler(
  * Creates a handler for finding pieces
  */
 export function findPieceHandler(pieceCid: string, found: boolean, options: PDPMockOptions = {}) {
-  const baseUrl = options.baseUrl ?? 'http://pdp.local'
+  const baseUrl = options.baseUrl ?? 'https://pdp.example.com'
 
   return http.get(`${baseUrl}/pdp/piece`, ({ request }) => {
     const url = new URL(request.url)
@@ -102,7 +102,7 @@ export function findPieceHandler(pieceCid: string, found: boolean, options: PDPM
 }
 
 export function findAnyPieceHandler(found: boolean, options: PDPMockOptions = {}) {
-  const baseUrl = options.baseUrl ?? 'http://pdp.local'
+  const baseUrl = options.baseUrl ?? 'https://pdp.example.com'
   return http.get(`${baseUrl}/pdp/piece`, ({ request }) => {
     const url = new URL(request.url)
     const queryCid = url.searchParams.get('pieceCid')
@@ -119,7 +119,7 @@ export function findAnyPieceHandler(found: boolean, options: PDPMockOptions = {}
  * Returns a UUID for 201, or a CID for 200
  */
 export function postPieceHandler(pieceCid: string, uuid?: string, options: PDPMockOptions = {}) {
-  const baseUrl = options.baseUrl ?? 'http://pdp.local'
+  const baseUrl = options.baseUrl ?? 'https://pdp.example.com'
   return http.post<Record<string, never>, { pieceCid: string }>(`${baseUrl}/pdp/piece`, async ({ request }) => {
     const body = await request.json()
     assert(body != null, 'Body should be defined')
@@ -142,7 +142,7 @@ export function postPieceHandler(pieceCid: string, uuid?: string, options: PDPMo
 }
 
 export function uploadPieceHandler(uuid: string, options: PDPMockOptions = {}) {
-  const baseUrl = options.baseUrl ?? 'http://pdp.local'
+  const baseUrl = options.baseUrl ?? 'https://pdp.example.com'
   return http.put(`${baseUrl}/pdp/piece/upload/${uuid}`, async () => {
     return HttpResponse.text('No Content', {
       status: 204,
@@ -156,7 +156,7 @@ export function uploadPieceHandler(uuid: string, options: PDPMockOptions = {}) {
  * Note: This endpoint doesn't require a request body
  */
 export function postPieceUploadsHandler(uuid: string, options: PDPMockOptions = {}) {
-  const baseUrl = options.baseUrl ?? 'http://pdp.local'
+  const baseUrl = options.baseUrl ?? 'https://pdp.example.com'
   return http.post(`${baseUrl}/pdp/piece/uploads`, async () => {
     // Create upload session, return UUID in Location header
     return HttpResponse.text('Created', {
@@ -173,7 +173,7 @@ export function postPieceUploadsHandler(uuid: string, options: PDPMockOptions =
  * PUT /pdp/piece/uploads/:uuid - streams piece data
  */
 export function uploadPieceStreamingHandler(uuid: string, options: PDPMockOptions = {}) {
-  const baseUrl = options.baseUrl ?? 'http://pdp.local'
+  const baseUrl = options.baseUrl ?? 'https://pdp.example.com'
   return http.put(`${baseUrl}/pdp/piece/uploads/${uuid}`, async ({ request }) => {
     await request.arrayBuffer()
     return HttpResponse.text('No Content', {
@@ -187,7 +187,7 @@ export function uploadPieceStreamingHandler(uuid: string, options: PDPMockOption
  * POST /pdp/piece/uploads/:uuid - finalize with PieceCID
  */
 export function finalizePieceUploadHandler(uuid: string, expectedPieceCid?: string, options: PDPMockOptions = {}) {
-  const baseUrl = options.baseUrl ?? 'http://pdp.local'
+  const baseUrl = options.baseUrl ?? 'https://pdp.example.com'
   return http.post<{ uuid: string }, { pieceCid: string }>(
     `${baseUrl}/pdp/piece/uploads/${uuid}`,
     async ({ request }) => {
@@ -211,7 +211,7 @@ export function finalizePieceUploadHandler(uuid: string, expectedPieceCid?: stri
  * Returns array of handlers for: POST /pdp/piece/uploads, PUT /pdp/piece/uploads/:uuid, POST /pdp/piece/uploads/:uuid
  */
 export function streamingUploadHandlers(options: PDPMockOptions = {}) {
-  const baseUrl = options.baseUrl ?? 'http://pdp.local'
+  const baseUrl = options.baseUrl ?? 'https://pdp.example.com'
   let uploadCounter = 0
 
   return [
@@ -288,7 +288,7 @@ export function createDataSetWithMetadataCapture(
   captureCallback: (metadata: MetadataCapture) => void,
   options: PDPMockOptions = {}
 ) {
-  const baseUrl = options.baseUrl ?? 'http://pdp.local'
+  const baseUrl = options.baseUrl ?? 'https://pdp.example.com'
 
   return http.post(`${baseUrl}/pdp/data-sets`, async ({ request }) => {
     const body = (await request.json()) as any
@@ -330,7 +330,7 @@ export function addPiecesWithMetadataCapture(
   captureCallback: (metadata: PieceMetadataCapture) => void,
   options: PDPMockOptions = {}
 ) {
-  const baseUrl = options.baseUrl ?? 'http://pdp.local'
+  const baseUrl = options.baseUrl ?? 'https://pdp.example.com'
 
   return http.post<{ id: string }, addPieces.RequestBody>(
     `${baseUrl}/pdp/data-sets/:id/pieces`,

packages/synapse-core/src/sp/index.ts

@@ -14,7 +14,7 @@ export * from './add-pieces.ts'
 export * from './data-sets.ts'
 export * from './get-data-set.ts'
 export * from './schedule-piece-deletion.ts'
-export type { deletePiece } from './sp.ts'
+export type { deletePiece, UploadPieceStreamingData } from './sp.ts'
 export { downloadPiece, findPiece, ping, uploadPiece, uploadPieceStreaming } from './sp.ts'
 export * from './upload.ts'
 export * from './wait-for-add-pieces.ts'

packages/synapse-core/src/sp/sp.ts

@@ -17,6 +17,7 @@ import * as Piece from '../piece.ts'
 import type * as TypedData from '../typed-data/index.ts'
 import { RETRY_CONSTANTS, SIZE_CONSTANTS } from '../utils/constants.ts'
 import { createPieceUrlPDP } from '../utils/piece-url.ts'
+import { isUint8Array } from '../utils/streams.ts'
 
 export namespace createDataSet {
   /**
@@ -235,12 +236,13 @@ export async function uploadPiece(options: uploadPiece.OptionsType): Promise<voi
   }
 }
 
+export type UploadPieceStreamingData = Uint8Array | ReadableStream | import('node:stream/web').ReadableStream
 export namespace uploadPieceStreaming {
   export type OptionsType = {
     /** The service URL of the PDP API. */
     serviceURL: string
     /** The data to upload. */
-    data: Blob
+    data: UploadPieceStreamingData
     /** The size of the data. If defined, it will be used to set the Content-Length header. */
     size?: number
     /** The progress callback. */
@@ -273,9 +275,6 @@ export namespace uploadPieceStreaming {
 export async function uploadPieceStreaming(
   options: uploadPieceStreaming.OptionsType
 ): Promise<uploadPieceStreaming.OutputType> {
-  if (options.data.size < SIZE_CONSTANTS.MIN_UPLOAD_SIZE || options.data.size > SIZE_CONSTANTS.MAX_UPLOAD_SIZE) {
-    throw new InvalidUploadSizeError(options.data.size)
-  }
   // Create upload session (POST /pdp/piece/uploads)
   const createResponse = await request.post(new URL('pdp/piece/uploads', options.serviceURL), {
     timeout: RETRY_CONSTANTS.MAX_RETRY_TIME,
@@ -316,20 +315,46 @@ export async function uploadPieceStreaming(
     getPieceCID = result.getPieceCID
   }
 
-  const dataStream = options.data.stream()
+  const dataStream = isUint8Array(options.data)
+    ? new Blob([options.data as Uint8Array<ArrayBuffer>]).stream()
+    : (options.data as ReadableStream) // ReadableStream types dont match between browsers and Node.js
+
+  const size = isUint8Array(options.data) ? options.data.length : options.size
 
   // Add size tracking and progress reporting
   let bytesUploaded = 0
-  const trackingStream = new TransformStream<Uint8Array, Uint8Array>({
+  const trackingStream = new TransformStream<unknown, Uint8Array>({
     transform(chunk, controller) {
-      bytesUploaded += chunk.length
+      let bytes: Uint8Array | undefined
+
+      if (isUint8Array(chunk)) {
+        bytes = chunk
+      } else if (ArrayBuffer.isView(chunk)) {
+        bytes = new Uint8Array(chunk.buffer, chunk.byteOffset, chunk.byteLength)
+      } else {
+        controller.error('Invalid chunk type only Uint8Array and TypedArray are supported')
+        return
+      }
+
+      bytesUploaded += bytes.length
+
+      if (bytesUploaded > SIZE_CONSTANTS.MAX_UPLOAD_SIZE) {
+        controller.error(new InvalidUploadSizeError(bytesUploaded))
+        return
+      }
 
       // Report progress if callback provided
       if (options.onProgress) {
         options.onProgress(bytesUploaded)
       }
 
-      controller.enqueue(chunk)
+      controller.enqueue(bytes)
+    },
+    flush(controller) {
+      if (bytesUploaded < SIZE_CONSTANTS.MIN_UPLOAD_SIZE) {
+        controller.error(new InvalidUploadSizeError(bytesUploaded))
+        return
+      }
     },
   })
 
@@ -341,7 +366,7 @@ export async function uploadPieceStreaming(
   // PUT /pdp/piece/uploads/{uuid} with streaming body
   const headers: Record<string, string> = {
     'Content-Type': 'application/octet-stream',
-    'Content-Length': options.data.size.toString(),
+    ...(size != null ? { 'Content-Length': size.toString() } : {}),
   }
 
   const uploadResponse = await request.put(new URL(`pdp/piece/uploads/${uploadUuid}`, options.serviceURL), {

packages/synapse-core/src/utils/streams.ts

@@ -111,3 +111,13 @@ export async function* uint8ArrayToAsyncIterable(
     yield data.subarray(i, i + chunkSize)
   }
 }
+
+/**
+ * Check if value is Uint8Array
+ *
+ * @param value - The value to check
+ * @returns True if it's a Uint8Array
+ */
+export function isUint8Array(value: unknown): value is Uint8Array {
+  return value instanceof Uint8Array || (ArrayBuffer.isView(value) && value.constructor.name === 'Uint8Array')
+}