From ef243c98546df0442d8c01dad9e569004a2643da Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Mon, 30 Mar 2026 16:27:29 -0700 Subject: [PATCH] feat: signed upload urls (#2753) * feat: signed upload urls * fix: make sessions write to dynamo * fix: add logs for failures --- .../src/ExtensionController.ts | 70 +- src/backend/src/api/api_error_handler.js | 44 +- src/backend/src/clients/dynamodb/DDBClient.ts | 98 +- .../kvstore/KVStoreInterfaceService.js | 12 + .../DynamoKVStore/DynamoKVStore.test.ts | 16 + .../services/DynamoKVStore/DynamoKVStore.ts | 65 +- .../MeteringService/MeteringService.ts | 6 +- src/backend/src/services/User.d.ts | 4 +- .../ConcurrentRequestLimiter.ts | 2 +- src/backend/src/services/auth/ACLService.js | 14 +- src/backend/src/services/auth/Actor.d.ts | 36 +- .../modules/FileSystem/operations/upload.js | 929 +++++++++++++++++- src/puter-js/src/modules/KV.js | 67 ++ src/puter-js/test/fs.test.js | 160 ++- 14 files changed, 1470 insertions(+), 53 deletions(-) diff --git a/extensions/extensionController/src/ExtensionController.ts b/extensions/extensionController/src/ExtensionController.ts index 153660f2f..5529a9ab6 100644 --- a/extensions/extensionController/src/ExtensionController.ts +++ b/extensions/extensionController/src/ExtensionController.ts @@ -5,6 +5,7 @@ import type { HttpMethod, RouterMethods, } from '../../api.d.ts'; +declare const extension: Partial>; /** * Class decorator to set prefix on prototype and register routes on instantiation * @argument prefix - prefix for all routes under the class @@ -81,11 +82,53 @@ export const Put = createMethodDecorator('put'); export const Delete = createMethodDecorator('delete'); // TODO DS: add others as needed (patch, etc) +interface HttpErrorOptions { + cause?: unknown; + legacyCode?: string; + code?: string; + fields?: Record; +} + +const isHttpErrorOptions = (value: unknown): value is HttpErrorOptions => { + if ( !value || typeof value !== 'object' || Array.isArray(value) ) { + return false; + } + + return ( + Object.prototype.hasOwnProperty.call(value, 'cause') + || Object.prototype.hasOwnProperty.call(value, 'legacyCode') + || Object.prototype.hasOwnProperty.call(value, 'code') + || Object.prototype.hasOwnProperty.call(value, 'fields') + ); +}; + export class HttpError extends Error { statusCode: number; - constructor (statusCode: StatusCodes, message: string, cause?: unknown) { - super(`${statusCode} - ${message}`, { cause }); + legacyCode?: string; + code?: string; + fields?: Record; + constructor ( + statusCode: StatusCodes, + message: string, + causeOrOptions?: unknown, + legacyCode?: string, + ) { + const options = isHttpErrorOptions(causeOrOptions) + ? causeOrOptions + : undefined; + const cause = options + ? options.cause + : causeOrOptions; + const resolvedLegacyCode = legacyCode ?? options?.legacyCode; + const code = options?.code; + super( + `${statusCode} - ${message}`, + cause !== undefined ? { cause } : undefined, + ); this.statusCode = statusCode; + this.legacyCode = resolvedLegacyCode; + this.code = code; + this.fields = options?.fields; } } @@ -154,7 +197,28 @@ export class ExtensionController { return await route.handler.bind(this)(req, res, next); } catch ( error ) { if ( error instanceof HttpError ) { - res.status(error.statusCode).send({ error: error.message }); + const payload: Record = { + error: error.message, + }; + if ( error.legacyCode ) { + payload.code = error.legacyCode; + } + if ( error.code ) { + if ( payload.code === undefined ) { + payload.code = error.code; + } else { + payload.errorCode = error.code; + } + } + if ( error.fields ) { + for ( const [key, value] of Object.entries(error.fields) ) { + if ( payload[key] !== undefined ) { + continue; + } + payload[key] = value; + } + } + res.status(error.statusCode).send(payload); logger.warn('httpError:', error); return; } diff --git a/src/backend/src/api/api_error_handler.js b/src/backend/src/api/api_error_handler.js index c04ecffbd..297715965 100644 --- a/src/backend/src/api/api_error_handler.js +++ b/src/backend/src/api/api_error_handler.js @@ -17,6 +17,46 @@ * along with this program. If not, see . */ const APIError = require('./APIError'); +const REDACTED_BODY_KEYS = new Set(['thumbnail', 'thumbnailData', 'base64']); +const MAX_LOG_STRING_LENGTH = 2048; + +const sanitizeAlarmBody = (value, key, seen = new WeakSet()) => { + if ( value === null || value === undefined ) { + return value; + } + + if ( typeof value === 'string' ) { + const isRedactedKey = typeof key === 'string' && REDACTED_BODY_KEYS.has(key); + const isDataUrl = value.startsWith('data:'); + if ( isRedactedKey || isDataUrl ) { + return `[redacted:${value.length}]`; + } + + if ( value.length > MAX_LOG_STRING_LENGTH ) { + return `${value.slice(0, MAX_LOG_STRING_LENGTH)}...[truncated:${value.length}]`; + } + return value; + } + + if ( typeof value !== 'object' ) { + return value; + } + + if ( seen.has(value) ) { + return '[circular]'; + } + seen.add(value); + + if ( Array.isArray(value) ) { + return value.map((item) => sanitizeAlarmBody(item, key, seen)); + } + + const output = {}; + for ( const [entryKey, entryValue] of Object.entries(value) ) { + output[entryKey] = sanitizeAlarmBody(entryValue, entryKey, seen); + } + return output; +}; /** * api_error_handler() is an express error handler for API errors. @@ -50,7 +90,7 @@ module.exports = function (err, req, res, next) { if ( typeof err === 'object' && !(err instanceof Error) && - err.hasOwnProperty('message') + Object.prototype.hasOwnProperty.call(err, 'message') ) { const apiError = APIError.create(400, err); return apiError.write(res); @@ -65,7 +105,7 @@ module.exports = function (err, req, res, next) { error: err, url: req.url, method: req.method, - body: req.body, + body: sanitizeAlarmBody(req.body, undefined), headers: req.headers, }); } diff --git a/src/backend/src/clients/dynamodb/DDBClient.ts b/src/backend/src/clients/dynamodb/DDBClient.ts index f40f98c91..f8a23148e 100644 --- a/src/backend/src/clients/dynamodb/DDBClient.ts +++ b/src/backend/src/clients/dynamodb/DDBClient.ts @@ -1,5 +1,5 @@ import { CreateTableCommand, CreateTableCommandInput, DynamoDBClient, UpdateTimeToLiveCommand } from '@aws-sdk/client-dynamodb'; -import { BatchGetCommand, BatchGetCommandInput, DeleteCommand, DynamoDBDocumentClient, GetCommand, PutCommand, QueryCommand, UpdateCommand } from '@aws-sdk/lib-dynamodb'; +import { BatchGetCommand, BatchGetCommandInput, BatchWriteCommand, BatchWriteCommandInput, DeleteCommand, DynamoDBDocumentClient, GetCommand, PutCommand, QueryCommand, UpdateCommand } from '@aws-sdk/lib-dynamodb'; import { NodeHttpHandler } from '@smithy/node-http-handler'; import dynalite from 'dynalite'; import { once } from 'node:events'; @@ -17,6 +17,9 @@ interface DBClientConfig { const LOCAL_DYNAMO_PATH_KEY = ':memory:'; const localDynaliteEndpointPromises = new Map>(); +const MAX_BATCH_WRITE_ITEMS = 25; +const MAX_BATCH_WRITE_RETRIES = 8; +const BATCH_WRITE_RETRY_BASE_MS = 25; const getDynalitePathKey = (path?: string) => { if ( path === ':memory:' ) return LOCAL_DYNAMO_PATH_KEY; @@ -52,6 +55,21 @@ const getOrCreateLocalDynaliteEndpoint = async (pathKey: string) => { return endpointPromise; }; +const chunkValues = (values: T[], size: number): T[][] => { + if ( values.length === 0 ) { + return []; + } + const chunks: T[][] = []; + for ( let index = 0; index < values.length; index += size ) { + chunks.push(values.slice(index, index + size)); + } + return chunks; +}; + +const sleep = async (ms: number) => { + await new Promise((resolve) => setTimeout(resolve, ms)); +}; + export class DDBClient { ddbClientPromise: Promise; #documentClient!: DynamoDBDocumentClient; @@ -170,6 +188,84 @@ export class DDBClient { return this.#documentClient.send(command); } + async batchPut (params: { table: string, item: Record }[]) { + const consumedCapacityByTable = new Map(); + if ( params.length === 0 ) { + return { ConsumedCapacity: [] }; + } + + const accumulateConsumedCapacity = ( + consumedCapacityEntries: Array<{ TableName?: string; CapacityUnits?: number }> | undefined, + ) => { + if ( ! consumedCapacityEntries ) { + return; + } + for ( const consumedCapacityEntry of consumedCapacityEntries ) { + const table = consumedCapacityEntry.TableName; + if ( ! table ) { + continue; + } + + const existingUsage = consumedCapacityByTable.get(table) ?? 0; + consumedCapacityByTable.set( + table, + existingUsage + Number(consumedCapacityEntry.CapacityUnits ?? 0), + ); + } + }; + + const chunks = chunkValues(params, MAX_BATCH_WRITE_ITEMS); + for ( const chunk of chunks ) { + let requestItems = chunk.reduce((acc, curr) => { + const tableRequests = acc[curr.table] ?? []; + tableRequests.push({ + PutRequest: { + Item: curr.item, + }, + }); + acc[curr.table] = tableRequests; + return acc; + }, {} as NonNullable); + + for ( let attempt = 0; attempt <= MAX_BATCH_WRITE_RETRIES; attempt++ ) { + if ( Object.keys(requestItems).length === 0 ) { + break; + } + + const response = await this.#documentClient.send(new BatchWriteCommand({ + RequestItems: requestItems, + ReturnConsumedCapacity: 'TOTAL', + })); + accumulateConsumedCapacity( + response.ConsumedCapacity as Array<{ TableName?: string; CapacityUnits?: number }> | undefined, + ); + + const unprocessedItems = response.UnprocessedItems ?? {}; + if ( Object.keys(unprocessedItems).length === 0 ) { + requestItems = {}; + break; + } + + requestItems = unprocessedItems as NonNullable; + if ( attempt < MAX_BATCH_WRITE_RETRIES ) { + const delayMs = Math.min(1000, BATCH_WRITE_RETRY_BASE_MS * (2 ** attempt)); + await sleep(delayMs); + } + } + + if ( Object.keys(requestItems).length > 0 ) { + throw new Error('Failed to batch write all items to DynamoDB'); + } + } + + return { + ConsumedCapacity: Array.from(consumedCapacityByTable.entries()).map(([TableName, CapacityUnits]) => ({ + TableName, + CapacityUnits, + })), + }; + } + async del> (table: string, key: T) { const command = new DeleteCommand({ TableName: table, diff --git a/src/backend/src/modules/kvstore/KVStoreInterfaceService.js b/src/backend/src/modules/kvstore/KVStoreInterfaceService.js index 9a55246ba..f0d9dc38e 100644 --- a/src/backend/src/modules/kvstore/KVStoreInterfaceService.js +++ b/src/backend/src/modules/kvstore/KVStoreInterfaceService.js @@ -23,6 +23,7 @@ const BaseService = require('../../services/BaseService'); * @typedef {Object} KVStoreInterface * @property {function(KVStoreGetParams): Promise} get - Retrieve the value(s) for the given key(s). * @property {function(KVStoreSetParams): Promise} set - Set a value for a key, with optional expiration. + * @property {function(KVStoreBatchPutParams): Promise} batchPut - Set many key-value entries in one call. * @property {function(KVStoreDelParams): Promise} del - Delete a value by key. * @property {function(KVStoreListParams): Promise} list - List key-value pairs, optionally with pagination. * @property {function(): Promise} flush - Delete all key-value pairs in the store. @@ -42,6 +43,9 @@ const BaseService = require('../../services/BaseService'); * @property {*} value - The value to store. * @property {number} [expireAt] - Optional UNIX timestamp (seconds) when the key should expire. * + * @typedef {Object} KVStoreBatchPutParams + * @property {{key: string, value: *, expireAt?: number}[]} items - Key/value pairs to store. + * * @typedef {Object} KVStoreDelParams * @property {string} key - The key to delete. * @@ -114,6 +118,14 @@ class KVStoreInterfaceService extends BaseService { }, result: { type: 'void' }, }, + batchPut: { + description: 'Set many values by key in a single call.', + parameters: { + items: { type: 'json', required: true }, + optConfig: { type: 'json', description: 'additional options for get, e.g. { appUuid: "someId" }' }, + }, + result: { type: 'void' }, + }, del: { description: 'Delete a value by key.', parameters: { diff --git a/src/backend/src/services/DynamoKVStore/DynamoKVStore.test.ts b/src/backend/src/services/DynamoKVStore/DynamoKVStore.test.ts index a6ba42c10..443e182f5 100644 --- a/src/backend/src/services/DynamoKVStore/DynamoKVStore.test.ts +++ b/src/backend/src/services/DynamoKVStore/DynamoKVStore.test.ts @@ -60,6 +60,22 @@ describe('DynamoKVStore', async () => { expect(stored).toEqual(value); }); + it('batchPut writes multiple values and honors expiration timestamps', async () => { + const actor = makeActor(101); + const nowInSeconds = Math.floor(Date.now() / 1000); + + await su.sudo(actor, () => kvStore.batchPut({ + items: [ + { key: 'batch-a', value: 'first' }, + { key: 'batch-b', value: 'expired', expireAt: nowInSeconds - 1 }, + { key: 'batch-a', value: 'overridden' }, + ], + })); + + const values = await su.sudo(actor, () => kvStore.get({ key: ['batch-a', 'batch-b'] })); + expect(values).toEqual(['overridden', null]); + }); + it('scopes data to the app when provided', async () => { const userId = 2; const actorAppOne = makeActor(userId, 'app-one'); diff --git a/src/backend/src/services/DynamoKVStore/DynamoKVStore.ts b/src/backend/src/services/DynamoKVStore/DynamoKVStore.ts index 056157d5b..73ea82b13 100644 --- a/src/backend/src/services/DynamoKVStore/DynamoKVStore.ts +++ b/src/backend/src/services/DynamoKVStore/DynamoKVStore.ts @@ -94,7 +94,7 @@ export class DynamoKVStore { const key_hash = murmurhash.v3(key); const kv_row = await this.#sqlClient.read( 'SELECT * FROM kv WHERE user_id=? AND app=? AND kkey_hash=? LIMIT 1', - [user.id, appUuid ?? DynamoKVStore.LEGACY_GLOBAL_APP_KEY, key_hash], + [user?.id, appUuid ?? DynamoKVStore.LEGACY_GLOBAL_APP_KEY, key_hash], ); if ( kv_row[0]?.value ) { @@ -103,7 +103,7 @@ export class DynamoKVStore { await this.set({ key: kv_row[0].key, value: kv_row[0].value }); await this.#sqlClient.write( 'DELETE FROM kv WHERE user_id=? AND app=? AND kkey_hash=?', - [user.id, appUuid ?? DynamoKVStore.LEGACY_GLOBAL_APP_KEY, key_hash], + [user?.id, appUuid ?? DynamoKVStore.LEGACY_GLOBAL_APP_KEY, key_hash], ); })(); values.push(kv_row[0]?.value); @@ -146,7 +146,6 @@ export class DynamoKVStore { @Span('kv:set') async set ({ key, value, expireAt, optConfig }: { key: string; value: unknown; expireAt?: number; optConfig?: { appUuid?: string } }): Promise { - const context = Context.get(); const actor = context.get('actor'); @@ -178,6 +177,66 @@ export class DynamoKVStore { return true; } + @Span('kv:batchPut') + async batchPut ({ + items, + optConfig, + }: { + items: Array<{ key: string; value: unknown; expireAt?: number }>; + optConfig?: { appUuid?: string }; + }): Promise { + const context = Context.get(); + const actor = context.get('actor'); + + if ( !Array.isArray(items) || items.length === 0 ) { + return true; + } + + const normalizedByKey = new Map(); + for ( const item of items ) { + const normalizedKey = String(item.key); + if ( normalizedKey === '' ) { + throw APIError.create('field_empty', undefined, { + key: 'key', + }); + } + + if ( Buffer.byteLength(normalizedKey, 'utf8') > 1024 ) { + throw new Error(`key is too large. Max size is ${1024}.`); + } + + normalizedByKey.set(normalizedKey, { + key: normalizedKey, + value: item.value, + expireAt: item.expireAt, + }); + } + + if ( this.#enableMigrationFromSQL ) { + for ( const key of normalizedByKey.keys() ) { + this.get({ key }); + } + } + + const namespace = this.#getNameSpace(actor, optConfig?.appUuid); + const putParams = Array.from(normalizedByKey.values()).map((item) => ({ + table: this.#tableName, + item: { + namespace, + key: item.key, + value: item.value, + ttl: item.expireAt, + }, + })); + const response = await this.#ddbClient.batchPut(putParams); + const usage = response.ConsumedCapacity?.reduce((acc, curr) => { + return acc + Number(curr.CapacityUnits ?? 0); + }, 0) ?? normalizedByKey.size; + + this.#meteringService.incrementUsage(actor, 'kv:write', usage || normalizedByKey.size); + return true; + } + @Span('kv:del') async del ({ key, optConfig}: { key: string;optConfig?: { appUuid?: string } }): Promise { const actor = Context.get('actor'); diff --git a/src/backend/src/services/MeteringService/MeteringService.ts b/src/backend/src/services/MeteringService/MeteringService.ts index cb5d1aeb9..a1afcc109 100644 --- a/src/backend/src/services/MeteringService/MeteringService.ts +++ b/src/backend/src/services/MeteringService/MeteringService.ts @@ -118,7 +118,7 @@ export class MeteringService { usageType = usageType.replace(/\./g, PERIOD_ESCAPE) as keyof typeof COST_MAPS; // replace dots with underscores for kvstore paths, TODO DS: map this back when reading const appId = actor.type?.app?.uid || GLOBAL_APP_KEY; - const userId = actor.type?.user.uuid; + const userId = actor.type?.user?.uuid!; const pathAndAmountMap = { 'total': totalCost, [`${usageType}.units`]: usageAmount, @@ -304,7 +304,7 @@ export class MeteringService { } const appId = actor.type?.app?.uid || GLOBAL_APP_KEY; - const userId = actor.type?.user.uuid; + const userId = actor.type?.user?.uuid!; const actorUsageKey = `${METRICS_PREFIX}:actor:${userId}:${currentMonth}`; const actorUsagesPromise = this.#kvStore.incr({ @@ -585,7 +585,7 @@ export class MeteringService { async getActorSubscription (actor: Actor): Promise<(typeof SUB_POLICIES)[number]> { // TODO DS: maybe allow non-user actors to have subscriptions eventually - if ( ! actor.type?.user.uuid ) { + if ( ! actor.type?.user?.uuid ) { throw new Error('Actor must be a user to get policy'); } diff --git a/src/backend/src/services/User.d.ts b/src/backend/src/services/User.d.ts index d57906ea3..d10076398 100644 --- a/src/backend/src/services/User.d.ts +++ b/src/backend/src/services/User.d.ts @@ -5,6 +5,8 @@ export interface IUser { uuid: string; username: string; email?: string; + free_storage?: number | string | null; + actual_free_storage?: number | string | null; subscription?: (typeof SUB_POLICIES)[number]['id'] & { active: boolean; tier: string; @@ -13,4 +15,4 @@ export interface IUser { repscore: number; email_confirmed: 1 | 0; requires_email_confirmation: 1 | 0; -} \ No newline at end of file +} diff --git a/src/backend/src/services/abuse-prevention/concurrentRequestLimiter/ConcurrentRequestLimiter.ts b/src/backend/src/services/abuse-prevention/concurrentRequestLimiter/ConcurrentRequestLimiter.ts index b4f1e5194..63722580a 100644 --- a/src/backend/src/services/abuse-prevention/concurrentRequestLimiter/ConcurrentRequestLimiter.ts +++ b/src/backend/src/services/abuse-prevention/concurrentRequestLimiter/ConcurrentRequestLimiter.ts @@ -123,7 +123,7 @@ export class ConcurrentRequestLimiter { throw new TypeError('leaseMs must be a positive number'); } - const userId = actor?.type?.user.uuid; + const userId = actor?.type?.user?.uuid; if ( ! userId ) { throw new Error('actor user id is required for concurrency checks'); } diff --git a/src/backend/src/services/auth/ACLService.js b/src/backend/src/services/auth/ACLService.js index 8bfc89552..97d5b8000 100644 --- a/src/backend/src/services/auth/ACLService.js +++ b/src/backend/src/services/auth/ACLService.js @@ -67,7 +67,6 @@ class ACLService extends BaseService { * @returns {Promise} True if access is allowed, false otherwise */ async check (actor, resource, mode) { - const ld = (Context.get('logdent') ?? 0) + 1; /** * Checks if an actor has permission for a specific mode on a resource * @@ -76,18 +75,7 @@ class ACLService extends BaseService { * @param {('see'| 'list'| 'read'| 'write' | 'manage')} mode - The permission mode to check ('see', 'list', 'read', 'write', 'manage') * @returns {Promise} True if actor has permission, false otherwise */ - return await Context.get().sub({ logdent: ld }).arun(async () => { - const result = await this._check_fsNode(actor, resource, mode); - if ( this.verbose ) { - console.log('LOGGING ACL CHECK', { - actor, - mode, - // trace: (new Error()).stack, - result, - }); - } - return result; - }); + return await this._check_fsNode(actor, resource, mode); } /** diff --git a/src/backend/src/services/auth/Actor.d.ts b/src/backend/src/services/auth/Actor.d.ts index 6ea36031a..107a52bd1 100644 --- a/src/backend/src/services/auth/Actor.d.ts +++ b/src/backend/src/services/auth/Actor.d.ts @@ -6,30 +6,30 @@ export interface ActorLogFields { } export class SystemActorType { - constructor (o?: Record); + constructor (_o?: Record); get uid (): string; - get_related_type (type_class: unknown): SystemActorType; + get_related_type (_type_class: unknown): SystemActorType; } export class UserActorType { - constructor (params: { user: IUser; session?: { uuid: string }; hasHttpOnlyCookie?: boolean }); + constructor (_params: { user: IUser; session?: { uuid: string }; hasHttpOnlyCookie?: boolean }); user: IUser; /** When true, this actor can access user-protected HTTP endpoints (e.g. change password). GUI tokens set this false. */ hasHttpOnlyCookie: boolean; get uid (): string; - get_related_type (type_class: unknown): UserActorType; + get_related_type (_type_class: unknown): UserActorType; } export class AppUnderUserActorType { - constructor (params: { user: IUser, app: { uid: string } }); + constructor (_params: { user: IUser, app: { id?: number; uid: string } }); user: IUser; - app: { uid: string }; + app: { id?: number; uid: string }; get uid (): string; - get_related_type (type_class: unknown): UserActorType | AppUnderUserActorType; + get_related_type (_type_class: unknown): UserActorType | AppUnderUserActorType; } export class AccessTokenActorType { - constructor (params: { authorizer: Actor, authorized?: Actor, token: string }); + constructor (_params: { authorizer: Actor, authorized?: Actor, token: string }); authorizer: Actor; authorized?: Actor; token: string; @@ -38,7 +38,7 @@ export class AccessTokenActorType { } export class SiteActorType { - constructor (params: { site: { name: string } }); + constructor (_params: { site: { name: string } }); site: { name: string }; get uid (): string; } @@ -55,20 +55,20 @@ export interface ActorInit { } export class Actor { - constructor (init: ActorInit); - type: { - app?: { uid: string, timestamp?: Date } - authorizer?: Actor - user: IUser + constructor (_init: ActorInit); + type: ActorType & { + user?: IUser; + app?: { id?: number; uid: string; timestamp?: Date }; + authorizer?: Actor; }; get uid (): string; get private_uid (): string; toLogFields (): ActorLogFields; clone (): Actor; - get_related_actor (type_class: unknown): Actor; + get_related_actor (_type_class: unknown): Actor; static create ( - type: new (params?: Record) => ActorType, - params?: { + _type: new (_params?: Record) => ActorType, + _params?: { user_uid?: string; app_uid?: string; user?: IUser; @@ -77,5 +77,5 @@ export class Actor { }, ): Promise; static get_system_actor (): Actor; - static adapt (actor?: Actor | { username?: string, uuid?: string }): Actor; + static adapt (_actor?: Actor | { username?: string, uuid?: string }): Actor; } diff --git a/src/puter-js/src/modules/FileSystem/operations/upload.js b/src/puter-js/src/modules/FileSystem/operations/upload.js index fb462de13..ffd7c706e 100644 --- a/src/puter-js/src/modules/FileSystem/operations/upload.js +++ b/src/puter-js/src/modules/FileSystem/operations/upload.js @@ -2,9 +2,26 @@ import path from '../../../lib/path.js'; import * as utils from '../../../lib/utils.js'; import getAbsolutePathForApp from '../utils/getAbsolutePathForApp.js'; +/* eslint-disable */ const MAX_THUMBNAIL_BYTES = 2 * 1024 * 1024; const DEFAULT_THUMBNAIL_DIMENSION = 128; const MIN_THUMBNAIL_DIMENSION = 32; +const SIGNED_BATCH_WRITE_CAPABILITY_KEY = 'signedBatchWriteSupported'; +const SIGNED_BATCH_REQUEST_CHUNK_SIZE = 500; +const SIGNED_BATCH_CHUNK_PIPELINE_CONCURRENCY = 4; +const SIGNED_BATCH_FILE_UPLOAD_CONCURRENCY = 8; +const SIGNED_MULTIPART_PART_UPLOAD_CONCURRENCY = 8; +const SIGNED_BATCH_WRITE_UNAVAILABLE_STATUSES = new Set([404, 405, 501]); + +// TODO: Remove this api.puter.com gate when signed batch-write is supported in all deployments. +const isSignedBatchWriteApiOrigin = (apiOrigin) => { + try { + const hostname = new URL(apiOrigin).hostname.replace(/\.$/, '').toLowerCase(); + return hostname === 'api.puter.com'; + } catch (error) { + return false; + } +}; const isLikelyImageFile = (file) => { if ( ! file ) return false; @@ -20,6 +37,39 @@ const estimateDataUrlSize = (dataUrl) => { return Math.ceil(base64.length * 3 / 4); }; +const isDataUrl = (value) => { + return typeof value === 'string' && value.startsWith('data:'); +}; + +const parseDataUrlContentType = (dataUrl) => { + if ( ! isDataUrl(dataUrl) ) return undefined; + const commaIndex = dataUrl.indexOf(','); + const metadata = commaIndex === -1 + ? dataUrl.slice(5) + : dataUrl.slice(5, commaIndex); + const [rawContentType] = metadata.split(';'); + const contentType = rawContentType ? rawContentType.trim() : ''; + return contentType || 'application/octet-stream'; +}; + +const dataUrlToBlob = async (dataUrl) => { + const response = await fetch(dataUrl); + if ( ! response.ok ) { + throw new Error('Failed to read thumbnail data URL'); + } + return await response.blob(); +}; + +const normalizeThumbnailData = (thumbnailData) => { + if ( typeof thumbnailData !== 'string' || thumbnailData.length === 0 ) { + return undefined; + } + if ( isDataUrl(thumbnailData) && estimateDataUrlSize(thumbnailData) > MAX_THUMBNAIL_BYTES ) { + return undefined; + } + return thumbnailData; +}; + const scaleDimensions = (width, height, maxDim) => { const base = Math.max(width, height) || 1; const scale = Math.min(1, maxDim / base); @@ -93,7 +143,210 @@ const defaultThumbnailGenerator = async (file) => { return undefined; }; -/* eslint-disable */ +const parseFetchResponseBody = async (response) => { + const text = await response.text(); + if ( ! text ) return null; + + try { + return JSON.parse(text); + } catch (e) { + return text; + } +}; + +const createApiHeaders = (authToken) => { + const headers = { + Authorization: `Bearer ${authToken}`, + 'Content-Type': 'application/json', + }; + + if ( ['web', 'app'].includes(puter.env) ) { + headers.Origin = 'https://puter.work'; + } + + return headers; +}; + +const toRequestError = (response, body, fallbackMessage) => { + const bodyRecord = body && typeof body === 'object' ? body : null; + const message = bodyRecord?.message + ?? (typeof bodyRecord?.error === 'string' ? bodyRecord.error : bodyRecord?.error?.message) + ?? (typeof body === 'string' && body.length > 0 ? body : null) + ?? fallbackMessage + ?? `Request failed with status ${response.status}`; + const error = new Error(message); + error.status = response.status; + error.body = body; + if ( typeof bodyRecord?.code === 'string' && bodyRecord.code.length > 0 ) { + error.code = bodyRecord.code; + } else if ( typeof bodyRecord?.errorCode === 'string' && bodyRecord.errorCode.length > 0 ) { + error.code = bodyRecord.errorCode; + } + return error; +}; + +const postJson = async (apiOrigin, authToken, endpoint, payload) => { + const response = await fetch(`${apiOrigin}${endpoint}`, { + method: 'POST', + headers: createApiHeaders(authToken), + credentials: 'include', + body: JSON.stringify(payload), + }); + const body = await parseFetchResponseBody(response); + if ( ! response.ok ) { + throw toRequestError(response, body, `Failed request to ${endpoint}`); + } + return body; +}; + +const isSignedBatchWriteUnavailableError = (error) => { + if ( !error || typeof error !== 'object' ) return false; + if ( error.signedBatchUnavailable === true ) return true; + const errorBody = error.body && typeof error.body === 'object' + ? error.body + : null; + const hasStructuredErrorCode = Boolean( + typeof errorBody?.code === 'string' && errorBody.code.length > 0, + ) || Boolean( + typeof errorBody?.errorCode === 'string' && errorBody.errorCode.length > 0, + ); + if ( hasStructuredErrorCode ) { + return false; + } + return SIGNED_BATCH_WRITE_UNAVAILABLE_STATUSES.has(error.status); +}; + +const toErrorMessage = (error) => { + if ( error && typeof error === 'object' ) { + if ( typeof error.message === 'string' && error.message.length > 0 ) { + return error.message; + } + if ( typeof error.body === 'string' && error.body.length > 0 ) { + return error.body; + } + if ( error.body && typeof error.body === 'object' ) { + if ( typeof error.body.message === 'string' && error.body.message.length > 0 ) { + return error.body.message; + } + if ( + error.body.error && + typeof error.body.error === 'object' && + typeof error.body.error.message === 'string' && + error.body.error.message.length > 0 + ) { + return error.body.error.message; + } + } + } + return String(error); +}; + +const toSignedRequestPath = (baseDirPath, requestItem) => { + if ( !requestItem || typeof requestItem !== 'object' ) { + return undefined; + } + + if ( requestItem.type === 'directory' ) { + return requestItem.directoryPath; + } + + if ( requestItem.type !== 'file' || !requestItem.file ) { + return undefined; + } + + const file = requestItem.file; + return file.puter_full_path + ?? path.join(baseDirPath, file.filepath || file.name || ''); +}; + +const chunkArray = (values, chunkSize) => { + const chunks = []; + if ( !Array.isArray(values) || values.length === 0 ) { + return chunks; + } + + const normalizedChunkSize = Math.max(1, Number(chunkSize) || 1); + for ( let index = 0; index < values.length; index += normalizedChunkSize ) { + chunks.push(values.slice(index, index + normalizedChunkSize)); + } + return chunks; +}; + +const uploadBlobToSignedUrl = async ({ + url, + blob, + contentType, + onProgress, + onRequestCreated, + onRequestCompleted, +}) => { + return await new Promise((resolve, reject) => { + const request = new XMLHttpRequest(); + request.open('PUT', url, true); + request.withCredentials = false; + + if ( contentType ) { + request.setRequestHeader('Content-Type', contentType); + } + + if ( onRequestCreated ) { + onRequestCreated(request); + } + + let previousLoaded = 0; + request.upload.addEventListener('progress', (event) => { + if ( ! onProgress ) return; + if ( ! event.lengthComputable ) return; + + const delta = Math.max(0, event.loaded - previousLoaded); + previousLoaded = event.loaded; + if ( delta > 0 ) { + onProgress(delta); + } + }); + + request.onload = () => { + if ( onRequestCompleted ) { + onRequestCompleted(request); + } + + if ( blob.size > previousLoaded && onProgress ) { + onProgress(blob.size - previousLoaded); + } + + if ( request.status >= 200 && request.status < 300 ) { + const etag = request.getResponseHeader('etag') ?? request.getResponseHeader('ETag'); + resolve({ etag }); + return; + } + + const error = new Error(`Signed upload failed with status ${request.status}`); + error.status = request.status; + reject(error); + }; + + request.onerror = () => { + if ( onRequestCompleted ) { + onRequestCompleted(request); + } + const error = new Error('Network error during signed upload'); + error.status = request.status; + reject(error); + }; + + request.onabort = () => { + if ( onRequestCompleted ) { + onRequestCompleted(request); + } + const error = new Error('Signed upload aborted'); + error.aborted = true; + reject(error); + }; + + request.send(blob); + }); +}; + const upload = async function (items, dirPath, options = {}) { return new Promise(async (resolve, reject) => { const DataTransferItem = globalThis.DataTransfer || (class DataTransferItem { @@ -140,6 +393,7 @@ const upload = async function (items, dirPath, options = {}) { // This will be used to uniquely identify this operation and its progress // across servers and clients const operation_id = utils.uuidv4(); + let start_callback_fired = false; // Call 'init' callback if provided // init is basically a hook that allows the user to get the operation ID and the XMLHttpRequest object @@ -254,18 +508,29 @@ const upload = async function (items, dirPath, options = {}) { //collect dirs if ( entries[i].isDirectory ) { - dirs.push({ path: path.join(dirPath, entries[i].finalPath ? entries[i].finalPath : entries[i].fullPath) }); + const rawDirPath = entries[i].finalPath ? entries[i].finalPath : entries[i].fullPath; + const relativeDirPath = typeof rawDirPath === 'string' + ? rawDirPath.replace(/^\/+/, '') + : ''; + dirs.push({ path: path.join(dirPath, relativeDirPath) }); } // also files else { // Dragged and dropped files do not have a finalPath property and hence the fileItem will go undefined. // In such cases, we need default to creating the files as uploaded by the user. - let fileItem = entries[i].finalPath ? entries[i].finalPath : entries[i].fullPath; + let fileItem = entries[i].finalPath || entries[i].filepath || entries[i].fullPath || entries[i].name; + if ( typeof fileItem === 'string' ) { + fileItem = fileItem.replace(/^\/+/, ''); + } let [dirLevel, fileName] = [fileItem?.slice(0, fileItem?.lastIndexOf('/')), fileItem?.slice(fileItem?.lastIndexOf('/') + 1)]; // If file name is blank then we need to create only an empty directory. // On the other hand if the file name is not blank(could be undefined), we need to create the file. - fileName != '' && files.push(entries[i]); + if ( fileName !== '' ) { + const normalizedFileItem = fileItem || entries[i].name; + entries[i].puter_full_path = path.join(dirPath, normalizedFileItem); + files.push(entries[i]); + } if ( options.createFileParent && fileItem.includes('/') ) { let incrementalDir; dirLevel.split('/').forEach((directory) => { @@ -323,6 +588,655 @@ const upload = async function (items, dirPath, options = {}) { } } + const signedDirectories = dirs.map((dir) => dir.path); + + const signedBatchWriteCapability = this[SIGNED_BATCH_WRITE_CAPABILITY_KEY]; + const signedBatchWriteAllowed = ( + signedBatchWriteCapability === true || + ( + signedBatchWriteCapability !== false && + isSignedBatchWriteApiOrigin(this.APIOrigin) + ) + ); + + const shouldAttemptSignedBatchWrite = ( + !options.shortcutTo && + (files.length > 0 || signedDirectories.length > 0) && + signedBatchWriteAllowed + ); + + if ( shouldAttemptSignedBatchWrite ) { + const overwriteEnabled = options.overwrite ?? false; + const shouldCreateMissingParents = Boolean( + options.createMissingAncestors || + options.createMissingParents || + options.createFileParent || + dirs.length > 0 + ); + let signedTotalSizeForProgress = total_size > 0 ? total_size : 1; + let signedBytesUploaded = 0; + const activeSignedRequests = new Set(); + const pendingUploadIds = new Set(); + let signedUploadAborted = false; + + const emitSignedProgress = () => { + let op_progress = ((signedBytesUploaded / signedTotalSizeForProgress) * 100).toFixed(2); + op_progress = op_progress > 100 ? 100 : op_progress; + if ( options.progress && typeof options.progress === 'function' ) { + options.progress(operation_id, op_progress); + } + }; + + const addSignedProgress = (delta) => { + if ( delta <= 0 ) return; + signedBytesUploaded += delta; + emitSignedProgress(); + }; + + const abortStartedUploads = async () => { + if ( pendingUploadIds.size === 0 ) return; + const uploadIds = Array.from(pendingUploadIds); + await Promise.all(uploadIds.map(async (uploadId) => { + try { + await postJson(this.APIOrigin, this.authToken, '/fs/abortWrite', { uploadId }); + } catch (e) { + // Ignore abort failures during cleanup. + } + })); + pendingUploadIds.clear(); + }; + + const abortSignedUpload = async () => { + if ( signedUploadAborted ) return; + signedUploadAborted = true; + for ( const request of activeSignedRequests ) { + try { + request.abort(); + } catch (e) { + // Ignore individual abort errors. + } + } + await abortStartedUploads(); + if ( options.abort && typeof options.abort === 'function' ) { + options.abort(operation_id); + } + }; + + xhr.abort = () => { + void abortSignedUpload(); + }; + + try { + const startRequestItems = []; + for ( let index = 0; index < signedDirectories.length; index++ ) { + startRequestItems.push({ + type: 'directory', + directoryPath: signedDirectories[index], + itemUploadId: `dir_${index}`, + }); + } + for ( let index = 0; index < files.length; index++ ) { + startRequestItems.push({ + type: 'file', + file: files[index], + fileIndex: index, + thumbnailData: normalizeThumbnailData(thumbnails[index] ?? options.thumbnail ?? undefined), + itemUploadId: String(index), + }); + } + + const signedThumbnailSizeTotal = startRequestItems.reduce((acc, requestItem) => { + if ( requestItem.type !== 'file' ) { + return acc; + } + if ( ! isDataUrl(requestItem.thumbnailData) ) { + return acc; + } + return acc + estimateDataUrlSize(requestItem.thumbnailData); + }, 0); + const signedTotalBytes = total_size + signedThumbnailSizeTotal; + signedTotalSizeForProgress = signedTotalBytes > 0 ? signedTotalBytes : 1; + + const startBatchRequests = startRequestItems.map((requestItem) => { + if ( requestItem.type === 'directory' ) { + return { + fileMetadata: { + path: requestItem.directoryPath, + size: 0, + contentType: 'application/x-puter-directory', + overwrite: overwriteEnabled, + createMissingParents: shouldCreateMissingParents, + }, + directory: true, + guiMetadata: { + operationId: operation_id, + itemUploadId: requestItem.itemUploadId, + socketId: this.socket.id, + originalClientSocketId: this.socket.id, + }, + }; + } + + const file = requestItem.file; + const targetPath = file.puter_full_path + ?? path.join(dirPath, file.filepath || file.name); + const fileMetadata = { + path: targetPath, + size: file.size, + contentType: file.type || 'application/octet-stream', + overwrite: overwriteEnabled, + dedupeName: options.dedupeName ?? true, + createMissingParents: shouldCreateMissingParents, + }; + + return { + fileMetadata, + ...(isDataUrl(requestItem.thumbnailData) ? { + thumbnailMetadata: { + contentType: parseDataUrlContentType(requestItem.thumbnailData), + size: estimateDataUrlSize(requestItem.thumbnailData), + }, + } : {}), + guiMetadata: { + operationId: operation_id, + itemUploadId: requestItem.itemUploadId, + socketId: this.socket.id, + originalClientSocketId: this.socket.id, + }, + }; + }); + + if ( options.start && typeof options.start === 'function' ) { + options.start(); + start_callback_fired = true; + } + + const responseItemsByRequestIndex = new Map(); + const failedUploadItems = []; + const failedCompletionItems = []; + const startBatchRequestChunks = chunkArray(startBatchRequests, SIGNED_BATCH_REQUEST_CHUNK_SIZE); + const startRequestItemChunks = chunkArray(startRequestItems, SIGNED_BATCH_REQUEST_CHUNK_SIZE); + const startRequestIndexChunks = chunkArray( + startRequestItems.map((_item, index) => index), + SIGNED_BATCH_REQUEST_CHUNK_SIZE, + ); + + if ( + startBatchRequestChunks.length !== startRequestItemChunks.length + || startBatchRequestChunks.length !== startRequestIndexChunks.length + ) { + throw new Error('Signed batch request chunk mapping is invalid'); + } + + const uploadSignedFileTask = async (uploadTask) => { + const { requestIndex, requestItem, startResponse } = uploadTask; + const file = requestItem.file; + + const thumbnailData = requestItem.thumbnailData; + let completionThumbnailData; + if ( isDataUrl(thumbnailData) ) { + const thumbnailUploadUrl = startResponse.thumbnailUploadUrl; + const thumbnailUrl = startResponse.thumbnailUrl; + if ( thumbnailUploadUrl && thumbnailUrl ) { + const thumbnailBlob = await dataUrlToBlob(thumbnailData); + if ( thumbnailBlob.size <= MAX_THUMBNAIL_BYTES ) { + await uploadBlobToSignedUrl({ + url: thumbnailUploadUrl, + blob: thumbnailBlob, + contentType: thumbnailBlob.type || parseDataUrlContentType(thumbnailData), + onProgress: addSignedProgress, + onRequestCreated: (request) => { + activeSignedRequests.add(request); + }, + onRequestCompleted: (request) => { + activeSignedRequests.delete(request); + }, + }); + completionThumbnailData = thumbnailUrl; + } + } + } else if ( typeof thumbnailData === 'string' && thumbnailData.length > 0 ) { + completionThumbnailData = thumbnailData; + } + + if ( startResponse.uploadMode === 'multipart' ) { + const partSize = Number(startResponse.multipartPartSize) || Math.max(file.size, 1); + const declaredPartCount = Number(startResponse.multipartPartCount); + const inferredPartCount = Math.max(1, Math.ceil(file.size / partSize)); + const partCount = Number.isInteger(declaredPartCount) && declaredPartCount > 0 + ? declaredPartCount + : inferredPartCount; + const partUrlMap = new Map(); + const providedPartUrls = Array.isArray(startResponse.multipartPartUrls) + ? startResponse.multipartPartUrls + : []; + for ( const partUrl of providedPartUrls ) { + if ( partUrl?.partNumber && partUrl?.url ) { + partUrlMap.set(Number(partUrl.partNumber), partUrl.url); + } + } + + const missingPartNumbers = []; + for ( let partNumber = 1; partNumber <= partCount; partNumber++ ) { + if ( !partUrlMap.has(partNumber) ) { + missingPartNumbers.push(partNumber); + } + } + if ( missingPartNumbers.length > 0 ) { + const signPartsResponse = await postJson( + this.APIOrigin, + this.authToken, + '/fs/signMultipartParts', + { + uploadId: startResponse.sessionId, + partNumbers: missingPartNumbers, + }, + ); + const signedPartUrls = Array.isArray(signPartsResponse?.multipartPartUrls) + ? signPartsResponse.multipartPartUrls + : []; + for ( const partUrl of signedPartUrls ) { + if ( partUrl?.partNumber && partUrl?.url ) { + partUrlMap.set(Number(partUrl.partNumber), partUrl.url); + } + } + } + + const completedParts = []; + const allPartNumbers = []; + for ( let partNumber = 1; partNumber <= partCount; partNumber++ ) { + allPartNumbers.push(partNumber); + } + const partNumberChunks = chunkArray( + allPartNumbers, + SIGNED_MULTIPART_PART_UPLOAD_CONCURRENCY, + ); + + for ( const partNumberChunk of partNumberChunks ) { + const partUploadSettledResults = await Promise.allSettled(partNumberChunk.map(async (partNumber) => { + const partUrl = partUrlMap.get(partNumber); + if ( !partUrl ) { + throw new Error(`Missing signed multipart URL for part ${partNumber}`); + } + + const startByte = (partNumber - 1) * partSize; + const endByte = Math.min(startByte + partSize, file.size); + const partBlob = file.slice(startByte, endByte); + const uploadResult = await uploadBlobToSignedUrl({ + url: partUrl, + blob: partBlob, + contentType: startResponse.contentType || file.type || 'application/octet-stream', + onProgress: addSignedProgress, + onRequestCreated: (request) => { + activeSignedRequests.add(request); + }, + onRequestCompleted: (request) => { + activeSignedRequests.delete(request); + }, + }); + if ( !uploadResult.etag ) { + throw new Error(`Missing ETag for multipart part ${partNumber}`); + } + + return { + partNumber, + etag: uploadResult.etag, + }; + })); + + for ( const partUploadSettledResult of partUploadSettledResults ) { + if ( partUploadSettledResult.status === 'rejected' ) { + throw partUploadSettledResult.reason; + } + completedParts.push(partUploadSettledResult.value); + } + } + completedParts.sort((partA, partB) => partA.partNumber - partB.partNumber); + + return { + requestIndex, + completionItem: { + uploadId: startResponse.sessionId, + parts: completedParts, + ...(completionThumbnailData !== undefined ? { thumbnailData: completionThumbnailData } : {}), + guiMetadata: { + operationId: operation_id, + itemUploadId: requestItem.itemUploadId, + socketId: this.socket.id, + originalClientSocketId: this.socket.id, + }, + }, + }; + } + + if ( !startResponse.url ) { + throw new Error('Signed upload URL is missing'); + } + + await uploadBlobToSignedUrl({ + url: startResponse.url, + blob: file, + contentType: startResponse.contentType || file.type || 'application/octet-stream', + onProgress: addSignedProgress, + onRequestCreated: (request) => { + activeSignedRequests.add(request); + }, + onRequestCompleted: (request) => { + activeSignedRequests.delete(request); + }, + }); + + return { + requestIndex, + completionItem: { + uploadId: startResponse.sessionId, + ...(completionThumbnailData !== undefined ? { thumbnailData: completionThumbnailData } : {}), + guiMetadata: { + operationId: operation_id, + itemUploadId: requestItem.itemUploadId, + socketId: this.socket.id, + originalClientSocketId: this.socket.id, + }, + }, + }; + }; + + const processSignedRequestChunk = async (chunkIndex) => { + if ( signedUploadAborted ) { + const abortError = new Error('Signed upload aborted'); + abortError.aborted = true; + throw abortError; + } + + const startBatchRequestChunk = startBatchRequestChunks[chunkIndex]; + const chunkRequestItems = startRequestItemChunks[chunkIndex]; + const chunkRequestIndexes = startRequestIndexChunks[chunkIndex]; + if ( !startBatchRequestChunk || !chunkRequestItems || !chunkRequestIndexes ) { + throw new Error('Missing signed batch request chunk'); + } + + const chunkResponses = await postJson( + this.APIOrigin, + this.authToken, + '/fs/startBatchWrite', + startBatchRequestChunk, + ); + + if ( !Array.isArray(chunkResponses) ) { + const unsupportedShapeError = new Error('Signed batch start response is invalid'); + unsupportedShapeError.signedBatchUnavailable = true; + throw unsupportedShapeError; + } + if ( chunkResponses.length !== startBatchRequestChunk.length ) { + throw new Error('Signed batch start response count mismatch'); + } + + const fileUploadTasks = []; + for ( let index = 0; index < chunkResponses.length; index++ ) { + const requestIndex = chunkRequestIndexes[index]; + const requestItem = chunkRequestItems[index]; + const startResponse = chunkResponses[index]; + if ( requestIndex === undefined || !requestItem || !startResponse ) { + throw new Error('Missing batch signed upload metadata'); + } + + if ( requestItem.type === 'directory' ) { + responseItemsByRequestIndex.set(requestIndex, startResponse.fsEntry ?? startResponse); + continue; + } + + if ( !startResponse.sessionId ) { + throw new Error('Signed batch response missing sessionId'); + } + + pendingUploadIds.add(startResponse.sessionId); + fileUploadTasks.push({ + requestIndex, + requestItem, + startResponse, + }); + } + + const completionItems = []; + const localFailedUploadItems = []; + const fileUploadTaskChunks = chunkArray(fileUploadTasks, SIGNED_BATCH_FILE_UPLOAD_CONCURRENCY); + for ( const fileUploadTaskChunk of fileUploadTaskChunks ) { + if ( signedUploadAborted ) { + const abortError = new Error('Signed upload aborted'); + abortError.aborted = true; + throw abortError; + } + + const uploadSettledResults = await Promise.allSettled( + fileUploadTaskChunk.map(async (uploadTask) => { + return await uploadSignedFileTask(uploadTask); + }), + ); + + for ( let resultIndex = 0; resultIndex < uploadSettledResults.length; resultIndex++ ) { + const uploadSettledResult = uploadSettledResults[resultIndex]; + if ( uploadSettledResult?.status === 'rejected' ) { + const failedUploadTask = fileUploadTaskChunk[resultIndex]; + if ( failedUploadTask ) { + localFailedUploadItems.push({ + requestIndex: failedUploadTask.requestIndex, + uploadId: failedUploadTask.startResponse.sessionId, + error: uploadSettledResult.reason, + }); + } + continue; + } + + completionItems.push(uploadSettledResult.value); + } + } + + if ( localFailedUploadItems.length > 0 ) { + const failedUploadIds = Array.from(new Set(localFailedUploadItems.map((item) => item.uploadId))); + await Promise.allSettled(failedUploadIds.map(async (uploadId) => { + pendingUploadIds.delete(uploadId); + await postJson(this.APIOrigin, this.authToken, '/fs/abortWrite', { uploadId }); + })); + failedUploadItems.push(...localFailedUploadItems); + } + + if ( completionItems.length === 0 ) { + return; + } + + completionItems.sort((itemA, itemB) => itemA.requestIndex - itemB.requestIndex); + const completionPayload = completionItems.map((item) => item.completionItem); + const completionRequestIndexes = completionItems.map((item) => item.requestIndex); + const completionPayloadChunks = chunkArray(completionPayload, SIGNED_BATCH_REQUEST_CHUNK_SIZE); + const completionRequestIndexChunks = chunkArray( + completionRequestIndexes, + SIGNED_BATCH_REQUEST_CHUNK_SIZE, + ); + if ( completionPayloadChunks.length !== completionRequestIndexChunks.length ) { + throw new Error('Signed batch completion request mapping is invalid'); + } + + const localFailedCompletionItems = []; + for ( let completionChunkIndex = 0; completionChunkIndex < completionPayloadChunks.length; completionChunkIndex++ ) { + if ( signedUploadAborted ) { + const abortError = new Error('Signed upload aborted'); + abortError.aborted = true; + throw abortError; + } + + const completionPayloadChunk = completionPayloadChunks[completionChunkIndex]; + const completionRequestIndexChunk = completionRequestIndexChunks[completionChunkIndex]; + if ( !completionPayloadChunk || !completionRequestIndexChunk ) { + throw new Error('Missing signed batch completion request chunk'); + } + + try { + const completionResponses = await postJson( + this.APIOrigin, + this.authToken, + '/fs/completeBatchWrite', + completionPayloadChunk, + ); + + if ( !Array.isArray(completionResponses) ) { + throw new Error('Signed batch completion response is invalid'); + } + if ( completionResponses.length !== completionPayloadChunk.length ) { + throw new Error('Signed batch completion response count mismatch'); + } + + for ( let index = 0; index < completionResponses.length; index++ ) { + const completionResponse = completionResponses[index]; + const requestIndex = completionRequestIndexChunk[index]; + const completionPayloadItem = completionPayloadChunk[index]; + if ( requestIndex === undefined ) { + throw new Error('Missing request index for completed signed batch response'); + } + if ( completionPayloadItem?.uploadId ) { + pendingUploadIds.delete(completionPayloadItem.uploadId); + } + responseItemsByRequestIndex.set(requestIndex, completionResponse?.fsEntry ?? completionResponse); + } + } catch (completionChunkError) { + const completionItemSettledResults = await Promise.allSettled(completionPayloadChunk.map(async (completionItem) => { + return await postJson( + this.APIOrigin, + this.authToken, + '/fs/completeWrite', + completionItem, + ); + })); + + for ( let index = 0; index < completionItemSettledResults.length; index++ ) { + const completionItemSettledResult = completionItemSettledResults[index]; + const requestIndex = completionRequestIndexChunk[index]; + const completionPayloadItem = completionPayloadChunk[index]; + if ( requestIndex === undefined || !completionPayloadItem ) { + continue; + } + + if ( completionItemSettledResult?.status === 'fulfilled' ) { + pendingUploadIds.delete(completionPayloadItem.uploadId); + responseItemsByRequestIndex.set( + requestIndex, + completionItemSettledResult.value?.fsEntry ?? completionItemSettledResult.value, + ); + continue; + } + + localFailedCompletionItems.push({ + requestIndex, + uploadId: completionPayloadItem.uploadId, + error: completionItemSettledResult?.status === 'rejected' + ? completionItemSettledResult.reason + : completionChunkError, + }); + } + } + } + + if ( localFailedCompletionItems.length > 0 ) { + const failedCompletionUploadIds = Array.from(new Set(localFailedCompletionItems.map((item) => item.uploadId))); + await Promise.allSettled(failedCompletionUploadIds.map(async (uploadId) => { + pendingUploadIds.delete(uploadId); + await postJson(this.APIOrigin, this.authToken, '/fs/abortWrite', { uploadId }); + })); + failedCompletionItems.push(...localFailedCompletionItems); + } + }; + + const startChunkIndexes = startBatchRequestChunks.map((_chunk, index) => index); + const startChunkGroups = chunkArray( + startChunkIndexes, + SIGNED_BATCH_CHUNK_PIPELINE_CONCURRENCY, + ); + for ( const startChunkGroup of startChunkGroups ) { + const startChunkSettledResults = await Promise.allSettled(startChunkGroup.map(async (chunkIndex) => { + await processSignedRequestChunk(chunkIndex); + })); + for ( const startChunkSettledResult of startChunkSettledResults ) { + if ( startChunkSettledResult.status === 'rejected' ) { + throw startChunkSettledResult.reason; + } + } + } + + this[SIGNED_BATCH_WRITE_CAPABILITY_KEY] = true; + + const failedSignedItems = [ + ...failedUploadItems.map((item) => ({ ...item, stage: 'upload' })), + ...failedCompletionItems.map((item) => ({ ...item, stage: 'complete' })), + ]; + if ( failedSignedItems.length > 0 ) { + const partialError = new Error('One or more signed batch file operations failed'); + const mappedFailedSignedItems = failedSignedItems.map((item) => { + const requestItem = startRequestItems[item.requestIndex]; + const itemPath = toSignedRequestPath(dirPath, requestItem); + return { + requestIndex: item.requestIndex, + uploadId: item.uploadId, + stage: item.stage, + path: itemPath, + name: typeof itemPath === 'string' && itemPath.length > 0 + ? path.basename(itemPath) + : undefined, + message: toErrorMessage(item.error), + }; + }); + if ( typeof console !== 'undefined' && typeof console.error === 'function' ) { + console.error('Signed batch write failed items:', mappedFailedSignedItems); + } + partialError.partial = true; + partialError.failedItems = mappedFailedSignedItems; + partialError.failedPaths = mappedFailedSignedItems + .map((item) => item.path) + .filter((itemPath) => typeof itemPath === 'string' && itemPath.length > 0); + partialError.completedItemCount = responseItemsByRequestIndex.size; + partialError.totalItemCount = startRequestItems.length; + throw partialError; + } + + const signedItemsList = []; + for ( let index = 0; index < startRequestItems.length; index++ ) { + if ( !responseItemsByRequestIndex.has(index) ) { + throw new Error(`Missing signed batch response item at index ${index}`); + } + signedItemsList.push(responseItemsByRequestIndex.get(index)); + } + addSignedProgress(Math.max(0, signedTotalSizeForProgress - signedBytesUploaded)); + + let signedItems = signedItemsList; + signedItems = signedItems.length === 1 ? signedItems[0] : signedItems; + + if ( options.success && typeof options.success === 'function' ) { + options.success(signedItems); + } + return resolve(signedItems); + } catch (signedError) { + if ( signedUploadAborted || signedError?.aborted ) { + return reject(signedError); + } + + const shouldFallbackToLegacy = isSignedBatchWriteUnavailableError(signedError); + + if ( isSignedBatchWriteUnavailableError(signedError) ) { + this[SIGNED_BATCH_WRITE_CAPABILITY_KEY] = false; + } + + try { + await abortStartedUploads(); + } catch (e) { + // Ignore cleanup errors. + } + + if ( shouldFallbackToLegacy ) { + delete xhr.abort; + } else { + return error(signedError); + } + } + } + // total size of the upload is doubled because we will be uploading the files to the server // and then the server will upload them to the cloud total_size = total_size * 2; @@ -391,7 +1305,7 @@ const upload = async function (items, dirPath, options = {}) { // Append file metadata to upload request if ( ! options.shortcutTo ) { for ( let i = 0; i < files.length; i++ ) { - const thumbnail = thumbnails[i] ?? options.thumbnail ?? undefined; + const thumbnail = normalizeThumbnailData(thumbnails[i] ?? options.thumbnail ?? undefined); const fileinfo_payload = { name: files[i].name, type: files[i].type, @@ -407,7 +1321,7 @@ const upload = async function (items, dirPath, options = {}) { } // Append write operations for each file for ( let i = 0; i < files.length; i++ ) { - const thumbnail = thumbnails[i] ?? options.thumbnail ?? undefined; + const thumbnail = normalizeThumbnailData(thumbnails[i] ?? options.thumbnail ?? undefined); const operation = { op: options.shortcutTo ? 'shortcut' : 'write', dedupe_name: options.dedupeName ?? true, @@ -575,8 +1489,9 @@ const upload = async function (items, dirPath, options = {}) { }; // Fire off the 'start' event - if ( options.start && typeof options.start === 'function' ) { + if ( !start_callback_fired && options.start && typeof options.start === 'function' ) { options.start(); + start_callback_fired = true; } // send request diff --git a/src/puter-js/src/modules/KV.js b/src/puter-js/src/modules/KV.js index 00b2dd2b5..1d8e846cf 100644 --- a/src/puter-js/src/modules/KV.js +++ b/src/puter-js/src/modules/KV.js @@ -29,6 +29,7 @@ const gui_cache_keys = [ const isObject = (value) => value !== null && typeof value === 'object' && !Array.isArray(value); const isOptConfigShorthand = (value) => isObject(value) && Object.prototype.hasOwnProperty.call(value, 'appUuid'); +const isBatchSetItem = (value) => isObject(value) && Object.prototype.hasOwnProperty.call(value, 'key'); class KV { MAX_KEY_SIZE = 1024; @@ -120,7 +121,36 @@ class KV { /** @type {SetFunction} */ set = async (...args) => { + if ( Array.isArray(args[0]) ) { + const items = args[0]; + const rest = args.slice(1); + + let optConfig; + let success; + let error; + + if ( rest[0] === undefined ) { + rest.shift(); + } + + if ( isObject(rest[0]) ) { + optConfig = rest.shift(); + } + + if ( typeof rest[0] === 'function' ) { + success = rest.shift(); + } + if ( typeof rest[0] === 'function' ) { + error = rest.shift(); + } + + return await this.setBatch_({ items, optConfig, success, error }); + } + if ( args.length === 1 && isObject(args[0]) ) { + if ( Array.isArray(args[0].items) ) { + return await this.setBatch_(args[0]); + } return await this.set_(args[0]); } @@ -182,6 +212,43 @@ class KV { }, }); + setBatch_ = utils.make_driver_method(['items'], 'puter-kvstore', undefined, 'batchPut', { + preprocess: (args) => { + if ( !Array.isArray(args.items) || args.items.length === 0 ) { + throw { message: 'Items are required', code: 'items_required' }; + } + + const normalizedItems = args.items.map((item) => { + if ( ! isBatchSetItem(item) ) { + throw { message: 'Each item must include a key', code: 'invalid_item' }; + } + + const key = String(item.key); + if ( key.length === 0 ) { + throw { message: 'Key cannot be undefined', code: 'key_undefined' }; + } + if ( key.length > this.MAX_KEY_SIZE ) { + throw { message: `Key size cannot be larger than ${this.MAX_KEY_SIZE}`, code: 'key_too_large' }; + } + + if ( item.value && item.value.length > this.MAX_VALUE_SIZE ) { + throw { message: `Value size cannot be larger than ${this.MAX_VALUE_SIZE}`, code: 'value_too_large' }; + } + + return { + key, + value: item.value, + ...(item.expireAt !== undefined ? { expireAt: item.expireAt } : {}), + }; + }); + + return { + ...args, + items: normalizedItems, + }; + }, + }); + /** * Resolves to the value if the key exists, or `undefined` if the key does not exist. Rejects with an error on failure. */ diff --git a/src/puter-js/test/fs.test.js b/src/puter-js/test/fs.test.js index ef1baa9f0..281744d68 100644 --- a/src/puter-js/test/fs.test.js +++ b/src/puter-js/test/fs.test.js @@ -96,6 +96,51 @@ naughtyStrings = [ "vietnameseTiếng Việt.txt", ] +const runWithUploadFlowMode = async (mode, work) => { + const fs = puter.fs; + const hadCapabilityFlag = Object.prototype.hasOwnProperty.call(fs, 'signedBatchWriteSupported'); + const previousCapabilityFlag = fs.signedBatchWriteSupported; + + if ( mode === 'legacy' ) { + fs.signedBatchWriteSupported = false; + } else if ( mode === 'signed' ) { + fs.signedBatchWriteSupported = true; + } + + try { + return await work(); + } finally { + if ( hadCapabilityFlag ) { + fs.signedBatchWriteSupported = previousCapabilityFlag; + } else { + delete fs.signedBatchWriteSupported; + } + } +}; + +const toItemsArray = (value) => { + if ( Array.isArray(value) ) { + return value; + } + if ( value === undefined || value === null ) { + return []; + } + return [value]; +}; + +const isApiPuterComOrigin = () => { + try { + const apiOrigin = puter?.fs?.APIOrigin ?? puter?.APIOrigin; + if ( typeof apiOrigin !== 'string' || apiOrigin.length === 0 ) { + return false; + } + const hostname = new URL(apiOrigin).hostname.replace(/\.$/, '').toLowerCase(); + return hostname === 'api.puter.com'; + } catch (error) { + return false; + } +}; + window.fsTests = [ { name: "testFSWrite", @@ -795,4 +840,117 @@ window.fsTests = [ } } }, -]; \ No newline at end of file + { + name: "testFSWriteParityBetweenSignedAndLegacyFlow", + description: "Test write() parity between default upload flow (signed when available) and forced legacy fallback flow", + test: async function() { + const rootDir = puter.randName(); + const defaultPath = `${rootDir}/default-parity-write.txt`; + const legacyPath = `${rootDir}/legacy-parity-write.txt`; + const defaultContent = `default-flow-content-${Date.now()}`; + const legacyContent = `legacy-flow-content-${Date.now()}`; + const defaultFlowLabel = isApiPuterComOrigin() ? 'signed-preferred' : 'legacy-default'; + + try { + await puter.fs.mkdir(rootDir); + + const defaultResult = await puter.fs.write(defaultPath, defaultContent, { + overwrite: true, + dedupeName: false, + }); + const legacyResult = await runWithUploadFlowMode('legacy', async () => { + return await puter.fs.write(legacyPath, legacyContent, { + overwrite: true, + dedupeName: false, + }); + }); + + assert(defaultResult && defaultResult.uid, "Default flow write failed"); + assert(legacyResult && legacyResult.uid, "Legacy flow write failed"); + + const defaultRead = await (await puter.fs.read(defaultPath)).text(); + const legacyRead = await (await puter.fs.read(legacyPath)).text(); + + assert(defaultRead === defaultContent, "Default flow wrote unexpected content"); + assert(legacyRead === legacyContent, "Legacy flow wrote unexpected content"); + assert(typeof defaultResult.name === 'string' && defaultResult.name.length > 0, "Default flow write returned invalid name"); + assert(typeof legacyResult.name === 'string' && legacyResult.name.length > 0, "Legacy flow write returned invalid name"); + + pass(`testFSWriteParityBetweenSignedAndLegacyFlow passed (${defaultFlowLabel})`); + } catch (error) { + fail("testFSWriteParityBetweenSignedAndLegacyFlow failed:", error); + } finally { + try { + await puter.fs.delete(rootDir, { recursive: true }); + } catch (cleanupError) { + } + } + } + }, + { + name: "testFSBatchUploadParityBetweenSignedAndLegacyFlow", + description: "Test upload() parity for batched files between default upload flow (signed when available) and forced legacy fallback flow", + test: async function() { + const rootDir = puter.randName(); + const defaultDir = `${rootDir}/default-batch`; + const legacyDir = `${rootDir}/legacy-batch`; + const defaultFlowLabel = isApiPuterComOrigin() ? 'signed-preferred' : 'legacy-default'; + const defaultFiles = [ + new File([`alpha-default-${Date.now()}`], 'alpha.txt', { type: 'text/plain' }), + new File([`beta-default-${Date.now()}`], 'beta.txt', { type: 'text/plain' }), + ]; + const legacyFiles = [ + new File([`alpha-legacy-${Date.now()}`], 'alpha.txt', { type: 'text/plain' }), + new File([`beta-legacy-${Date.now()}`], 'beta.txt', { type: 'text/plain' }), + ]; + + try { + await puter.fs.mkdir(rootDir); + await puter.fs.mkdir(defaultDir, { createMissingParents: true }); + await puter.fs.mkdir(legacyDir, { createMissingParents: true }); + + const defaultUploadResult = await puter.fs.upload(defaultFiles, defaultDir, { + overwrite: true, + dedupeName: false, + strict: true, + }); + + const legacyUploadResult = await runWithUploadFlowMode('legacy', async () => { + return await puter.fs.upload(legacyFiles, legacyDir, { + overwrite: true, + dedupeName: false, + strict: true, + }); + }); + + const defaultItems = toItemsArray(defaultUploadResult); + const legacyItems = toItemsArray(legacyUploadResult); + + assert(defaultItems.length === defaultFiles.length, "Default flow batch upload returned unexpected number of items"); + assert(legacyItems.length === legacyFiles.length, "Legacy flow batch upload returned unexpected number of items"); + + for ( let i = 0; i < defaultFiles.length; i++ ) { + const defaultItem = defaultItems[i]; + const legacyItem = legacyItems[i]; + const expectedName = defaultFiles[i].name; + const defaultContent = await (await puter.fs.read(`${defaultDir}/${expectedName}`)).text(); + const legacyContent = await (await puter.fs.read(`${legacyDir}/${expectedName}`)).text(); + + assert(defaultItem && defaultItem.uid, `Default flow item ${i} missing uid`); + assert(legacyItem && legacyItem.uid, `Legacy flow item ${i} missing uid`); + assert(defaultContent === await defaultFiles[i].text(), `Default flow content mismatch for ${expectedName}`); + assert(legacyContent === await legacyFiles[i].text(), `Legacy flow content mismatch for ${expectedName}`); + } + + pass(`testFSBatchUploadParityBetweenSignedAndLegacyFlow passed (${defaultFlowLabel})`); + } catch (error) { + fail("testFSBatchUploadParityBetweenSignedAndLegacyFlow failed:", error); + } finally { + try { + await puter.fs.delete(rootDir, { recursive: true }); + } catch (cleanupError) { + } + } + } + }, +];