fix(websocket): negotiate blob transport for socket.io

Author: danielebriggi

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@sqlitecloud/drivers",
-  "version": "1.0.837",
+  "version": "1.0.872",
   "description": "SQLiteCloud drivers for Typescript/Javascript in edge, web and node clients",
   "main": "./lib/index.js",
   "types": "./lib/index.d.ts",
@@ -46,6 +46,7 @@
     "eventemitter3": "^5.0.1",
     "lz4js": "^0.2.0",
     "socket.io-client": "^4.8.1",
+    "socket.io-parser": "~4.2.4",
     "whatwg-url": "^14.2.0"
   },
   "peerDependencies": {

package.json

@@ -3,10 +3,54 @@
  */
 
 import { io, Socket } from 'socket.io-client'
+import { Decoder as SocketIODecoder, Encoder as SocketIOEncoder } from 'socket.io-parser'
 import { SQLiteCloudConnection } from './connection'
 import { SQLiteCloudRowset } from './rowset'
-import { ErrorCallback, ResultsCallback, SQLiteCloudCommand, SQLiteCloudConfig, SQLiteCloudError } from './types'
-import { decodeBigIntMarkers, encodeBigIntMarkers } from './utilities'
+import {
+  DEFAULT_WEBSOCKET_BLOB_TRANSFER_FORMAT,
+  ErrorCallback,
+  ResultsCallback,
+  SQLiteCloudCommand,
+  SQLiteCloudConfig,
+  SQLiteCloudError,
+  SQLiteCloudWebsocketBlobTransferFormat
+} from './types'
+import { decodeBigIntMarkers, decodeWebsocketRowsetData, encodeBigIntMarkers, parseWebsocketBlobTransferFormat, parseWebsocketMaxAttachments } from './utilities'
+
+const SocketIODecoderBase = SocketIODecoder as unknown as new (...args: any[]) => { opts?: { maxAttachments?: number } }
+
+function createSocketIOParser(maxAttachments: number) {
+  class SQLiteCloudSocketIODecoder extends SocketIODecoderBase {
+    constructor(opts?: any) {
+      const decoderOptions = typeof opts === 'function' ? { reviver: opts } : opts
+      super(decoderOptions?.reviver)
+      this.opts ||= {}
+      this.opts.maxAttachments = Math.max(this.opts.maxAttachments ?? decoderOptions?.maxAttachments ?? 0, maxAttachments)
+    }
+  }
+
+  return {
+    Encoder: SocketIOEncoder,
+    Decoder: SQLiteCloudSocketIODecoder
+  }
+}
+
+function getResponseBlobTransferFormat(response: any): SQLiteCloudWebsocketBlobTransferFormat | undefined {
+  return parseWebsocketBlobTransferFormat(response?.capabilities?.blobTransferFormat || response?.blobTransferFormat, undefined)
+}
+
+function getAttachmentLimitError(description: unknown, limit: number): SQLiteCloudError | undefined {
+  const descriptionMessage = description instanceof Error ? description.message : typeof description === 'string' ? description : ''
+  if (/illegal attachments/i.test(descriptionMessage)) {
+    return new SQLiteCloudError(
+      `WebSocket blob response exceeded the configured Socket.IO attachment limit (${limit}). Use websocketBlobFormat=base64-blobs-v1 or increase websocketMaxAttachments.`,
+      {
+        errorCode: 'ERR_WEBSOCKET_MAX_ATTACHMENTS_EXCEEDED',
+        cause: description as Error | string
+      }
+    )
+  }
+}
 
 /**
  * Implementation of TransportConnection that connects to the database indirectly
@@ -31,20 +75,33 @@ export class SQLiteCloudWebsocketConnection extends SQLiteCloudConnection {
       if (!this.socket) {
         this.config = config
         const connectionstring = this.config.connectionstring as string
+        const websocketMaxAttachments = parseWebsocketMaxAttachments(this.config.websocketMaxAttachments)
         const gatewayUrl = this.config?.gatewayurl || `${this.config.host === 'localhost' ? 'ws' : 'wss'}://${this.config.host as string}:443`
-        this.socket = io(gatewayUrl, { auth: { token: connectionstring } })
+        this.socket = io(gatewayUrl, { auth: { token: connectionstring }, parser: createSocketIOParser(websocketMaxAttachments) })
 
         this.socket.on('connect', () => {
           callback?.call(this, null)
         })
 
-        this.socket.on('disconnect', reason => {
+        this.socket.on('disconnect', (reason, description) => {
           this.close()
-          callback?.call(this, new SQLiteCloudError('Disconnected', { errorCode: 'ERR_CONNECTION_ENDED', cause: reason }))
+          callback?.call(
+            this,
+            (reason === 'parse error' && getAttachmentLimitError(description, websocketMaxAttachments)) ||
+              new SQLiteCloudError('Disconnected', { errorCode: 'ERR_CONNECTION_ENDED', cause: reason })
+          )
         })
 
         this.socket.on('connect_error', (error: any) => {
           this.close()
+          if (error?.message === 'parse error' || error?.cause === 'parse error') {
+            callback?.call(
+              this,
+              getAttachmentLimitError(error?.description || error?.data || error?.cause, websocketMaxAttachments) ||
+                new SQLiteCloudError('Connection error', { errorCode: 'ERR_CONNECTION_ERROR', cause: error })
+            )
+            return
+          }
           let message = error.message || 'Connection error'
           if (typeof error.context == 'object' && error.context.responseText) {
             try {
@@ -87,16 +144,24 @@ export class SQLiteCloudWebsocketConnection extends SQLiteCloudConnection {
         sql: commands.query,
         bind: encodeBigIntMarkers(commands.parameters),
         row: 'array',
-        safe_integer_mode: this.config.safe_integer_mode
+        safe_integer_mode: this.config.safe_integer_mode,
+        capabilities: {
+          blobTransferFormat: this.config.websocketBlobFormat || DEFAULT_WEBSOCKET_BLOB_TRANSFER_FORMAT
+        }
       },
       (response: any) => {
         if (response?.error) {
           const error = new SQLiteCloudError(response.error.detail, { ...response.error })
           callback?.call(this, error)
         } else {
           const { metadata } = response
-          const data = decodeBigIntMarkers(response?.data, this.config.safe_integer_mode)
-          if (data && metadata) {
+          const blobTransferFormat = getResponseBlobTransferFormat(response)
+          const data =
+            metadata && metadata.numberOfRows !== undefined && metadata.numberOfColumns !== undefined && metadata.columns !== undefined
+              ? decodeWebsocketRowsetData(response?.data, metadata, this.config.safe_integer_mode, blobTransferFormat)
+              : decodeBigIntMarkers(response?.data, this.config.safe_integer_mode)
+
+          if (data !== undefined && metadata) {
             if (metadata.numberOfRows !== undefined && metadata.numberOfColumns !== undefined && metadata.columns !== undefined) {
               console.assert(Array.isArray(data), 'SQLiteCloudWebsocketConnection.transportCommands - data is not an array')
               // we can recreate a SQLiteCloudRowset from the response which we know to be an array of arrays

src/drivers/connection-ws.ts

@@ -20,8 +20,11 @@ export const DEFAULT_PORT = 8860
  * mixed - use BigInt and Number types depending on the value size
  */
 export type SQLiteCloudSafeIntegerMode = 'number' | 'bigint' | 'mixed'
+export type SQLiteCloudWebsocketBlobTransferFormat = 'base64-blobs-v1' | 'socketio-blobs-v1'
 
 export let SAFE_INTEGER_MODE: SQLiteCloudSafeIntegerMode = 'number'
+export const DEFAULT_WEBSOCKET_BLOB_TRANSFER_FORMAT: SQLiteCloudWebsocketBlobTransferFormat = 'base64-blobs-v1'
+export const DEFAULT_WEBSOCKET_MAX_ATTACHMENTS = 100000
 if (typeof process !== 'undefined') {
   const mode = process.env['SAFE_INTEGER_MODE']?.toLowerCase()
   if (mode === 'bigint' || mode === 'mixed' || mode === 'number') {
@@ -94,6 +97,10 @@ export interface SQLiteCloudConfig {
   usewebsocket?: boolean
   /** Url where we can connect to a SQLite Cloud Gateway that has a socket.io deamon waiting to connect, eg. wss://host:443 */
   gatewayurl?: string
+  /** Preferred blob transfer format when using websocket transport. Defaults to base64-blobs-v1 for new clients. */
+  websocketBlobFormat?: SQLiteCloudWebsocketBlobTransferFormat
+  /** Maximum number of socket.io binary attachments accepted by the websocket parser. */
+  websocketMaxAttachments?: number
 
   /** Optional identifier used for verbose logging */
   clientid?: string

src/drivers/types.ts

@@ -3,14 +3,19 @@
 //
 
 import {
+  DEFAULT_WEBSOCKET_BLOB_TRANSFER_FORMAT,
+  DEFAULT_WEBSOCKET_MAX_ATTACHMENTS,
   DEFAULT_PORT,
   DEFAULT_TIMEOUT,
+  ErrorCallback,
   SAFE_INTEGER_MODE,
+  SQLCloudRowsetMetadata,
   SQLiteCloudArrayType,
   SQLiteCloudConfig,
   SQLiteCloudDataTypes,
   SQLiteCloudError,
-  SQLiteCloudSafeIntegerMode
+  SQLiteCloudSafeIntegerMode,
+  SQLiteCloudWebsocketBlobTransferFormat
 } from './types'
 import { getSafeURL } from './safe-imports'
 import { Buffer } from 'buffer'
@@ -187,6 +192,8 @@ export function validateConfiguration(config: SQLiteCloudConfig): SQLiteCloudCon
   config.noblob = parseBoolean(config.noblob)
   config.compression = config.compression != undefined && config.compression != null ? parseBoolean(config.compression) : true // default: true
   config.safe_integer_mode = parseSafeIntegerMode(config.safe_integer_mode || SAFE_INTEGER_MODE)
+  config.websocketBlobFormat = parseWebsocketBlobTransferFormat(config.websocketBlobFormat)
+  config.websocketMaxAttachments = parseWebsocketMaxAttachments(config.websocketMaxAttachments)
 
   config.create = parseBoolean(config.create)
   config.non_linearizable = parseBoolean(config.non_linearizable)
@@ -257,6 +264,8 @@ export function parseconnectionstring(connectionstring: string): SQLiteCloudConf
       maxrowset: options.maxrowset ? parseInt(options.maxrowset) : undefined,
       safe_integer_mode: options.safe_integer_mode ? parseSafeIntegerMode(options.safe_integer_mode) : undefined,
       usewebsocket: options.usewebsocket ? parseBoolean(options.usewebsocket) : undefined,
+      websocketBlobFormat: options.websocket_blob_format ? parseWebsocketBlobTransferFormat(options.websocket_blob_format, undefined) : undefined,
+      websocketMaxAttachments: options.websocket_max_attachments ? parseWebsocketMaxAttachments(options.websocket_max_attachments) : undefined,
       verbose: options.verbose ? parseBoolean(options.verbose) : undefined
     }
 
@@ -302,7 +311,29 @@ export function parseSafeIntegerMode(value: string | SQLiteCloudSafeIntegerMode
   return 'number'
 }
 
+/** Parse websocket BLOB transport format, falling back to the driver default for new websocket clients. */
+export function parseWebsocketBlobTransferFormat(
+  value: string | SQLiteCloudWebsocketBlobTransferFormat | null | undefined,
+  fallback: SQLiteCloudWebsocketBlobTransferFormat | undefined = DEFAULT_WEBSOCKET_BLOB_TRANSFER_FORMAT
+): SQLiteCloudWebsocketBlobTransferFormat | undefined {
+  const format = value?.toLowerCase()
+  if (format === 'base64-blobs-v1' || format === 'socketio-blobs-v1') {
+    return format
+  }
+  return fallback
+}
+
+/** Parse the maximum number of socket.io binary attachments allowed for a websocket response. */
+export function parseWebsocketMaxAttachments(value: string | number | null | undefined): number {
+  const parsed = typeof value === 'string' ? Number.parseInt(value, 10) : value
+  if (Number.isSafeInteger(parsed) && (parsed as number) > 0) {
+    return parsed as number
+  }
+  return DEFAULT_WEBSOCKET_MAX_ATTACHMENTS
+}
+
 const BIGINT_MARKER_RE = /^-?\d+n$/
+const BLOB_COLUMN_TYPE_RE = /\bblob\b/i
 
 /** Convert values that JSON cannot represent losslessly into sqlitecloud-js bigint markers. */
 export function encodeBigIntMarkers(value: any): any {
@@ -349,3 +380,31 @@ export function decodeBigIntMarkers(value: any, safeIntegerMode?: SQLiteCloudSaf
 
   return value
 }
+
+/** Decode websocket rowset cells using metadata-aware rules for bigint markers and negotiated BLOB transport. */
+export function decodeWebsocketRowsetData(
+  data: any,
+  metadata: SQLCloudRowsetMetadata,
+  safeIntegerMode?: SQLiteCloudSafeIntegerMode,
+  blobTransferFormat?: SQLiteCloudWebsocketBlobTransferFormat
+): any {
+  if (!Array.isArray(data)) {
+    return decodeBigIntMarkers(data, safeIntegerMode)
+  }
+
+  const blobColumnIndexes = new Set(
+    metadata.columns.flatMap((column, index) => (column.type && BLOB_COLUMN_TYPE_RE.test(column.type) ? [index] : []))
+  )
+  const decodeCell = (value: any, columnIndex: number) => {
+    if (blobTransferFormat === 'base64-blobs-v1' && blobColumnIndexes.has(columnIndex) && typeof value === 'string') {
+      return Buffer.from(value, 'base64')
+    }
+    return decodeBigIntMarkers(value, safeIntegerMode)
+  }
+
+  if (data.every(row => !Array.isArray(row))) {
+    return data.map((value, index) => decodeCell(value, index % metadata.numberOfColumns))
+  }
+
+  return data.map(row => (Array.isArray(row) ? row.map((value, columnIndex) => decodeCell(value, columnIndex)) : decodeBigIntMarkers(row, safeIntegerMode)))
+}

src/drivers/utilities.ts

@@ -0,0 +1,88 @@
+import { describe, expect, it } from '@jest/globals'
+import { Database, SQLiteCloudRowset } from '../src'
+import { GATEWAY_URL, LONG_TIMEOUT, getChinookDatabase } from './shared'
+
+const BLOB_TRANSPORT_TABLE = 'blob_transport_probe'
+const BLOB_ROW_COUNT = 100
+const BLOB_SIZE = 32
+
+async function assertBlobRowsRoundTrip(database: Database) {
+  await database.sql(`DROP TABLE IF EXISTS ${BLOB_TRANSPORT_TABLE};`)
+  await database.sql(`
+    CREATE TABLE ${BLOB_TRANSPORT_TABLE} (
+      id INTEGER PRIMARY KEY,
+      label TEXT NOT NULL,
+      payload BLOB NOT NULL
+    );
+  `)
+  await database.sql(
+    `
+      WITH RECURSIVE seq(n) AS (
+        SELECT 1
+        UNION ALL
+        SELECT n + 1 FROM seq WHERE n < ?
+      )
+      INSERT INTO ${BLOB_TRANSPORT_TABLE} (id, label, payload)
+      SELECT n, 'row-' || n, randomblob(?)
+      FROM seq;
+    `,
+    BLOB_ROW_COUNT,
+    BLOB_SIZE
+  )
+
+  const rows = (await database.sql(`
+    SELECT
+      id,
+      label,
+      length(payload) AS payloadLength,
+      payload
+    FROM ${BLOB_TRANSPORT_TABLE}
+    ORDER BY id
+    LIMIT ${BLOB_ROW_COUNT};
+  `)) as SQLiteCloudRowset
+
+  expect(rows).toHaveLength(BLOB_ROW_COUNT)
+
+  rows.forEach((row, index) => {
+    expect(row.id).toBe(index + 1)
+    expect(row.label).toBe(`row-${index + 1}`)
+    expect(row.payloadLength).toBe(BLOB_SIZE)
+    expect(row.payload).toBeInstanceOf(Buffer)
+    expect((row.payload as Buffer).length).toBe(BLOB_SIZE)
+  })
+}
+
+describe('blob rowsets across transports', () => {
+  it(
+    'should return more than 12 rows with small blobs over tls',
+    async () => {
+      let database: Database | undefined
+
+      try {
+        database = getChinookDatabase()
+        await assertBlobRowsRoundTrip(database)
+      } finally {
+        database?.close()
+      }
+    },
+    LONG_TIMEOUT
+  )
+
+  it(
+    'should return more than 12 rows with small blobs over websocket',
+    async () => {
+      let database: Database | undefined
+
+      try {
+        database = getChinookDatabase(undefined, {
+          usewebsocket: true,
+          gatewayurl: GATEWAY_URL
+        })
+        await assertBlobRowsRoundTrip(database)
+      } finally {
+        database?.close()
+      }
+    },
+    LONG_TIMEOUT
+  )
+})