Boost js file uploads using Web Workers and streams

// Assumes WorkerPool class is defined elsewhere (see below)
const pool = new WorkerPool('upload-worker.js')

const fileInput = document.querySelector('#file-input')

fileInput.addEventListener('change', (evt) => {
  const files = Array.from(evt.target.files)
  files.forEach((file) => {
    console.log(`Queueing ${file.name} for upload...`)
    pool.processFile(file, {
      onProgress: (pct, msg) => updateProgressUI(file.name, pct, msg),
      onComplete: (msg) => showSuccess(file.name, msg),
      onError: (err) => showError(file.name, err),
    })
  })
})

function updateProgressUI(filename, pct, message) {
  // Update your progress bar or UI element here
  console.log(`${filename}: ${pct.toFixed(1)}% – ${message}`)
}

function showSuccess(filename, message) {
  // Update UI to show completion
  console.info(`${filename}: ${message}`)
}

function showError(filename, error) {
  // Update UI to show error state
  console.error(`${filename}: Upload failed - ${error}`)
}

// Example: Clean up the pool when the page unloads
window.addEventListener('unload', () => {
  pool.terminate()
})

self.onmessage = async (event) => {
  const file = event.data
  let reader // Declare reader outside the try block for finally access

  try {
    if (!file || typeof file.stream !== 'function') {
      throw new Error('Invalid file object received.')
    }

    const stream = file.stream()
    reader = stream.getReader()

    const total = file.size
    let uploaded = 0
    let chunkIndex = 0

    while (true) {
      let value, done
      try {
        // Read the next chunk from the stream
        ;({ value, done } = await reader.read())
      } catch (readError) {
        // Handle potential errors during stream reading
        throw new Error(`Error reading file stream: ${readError.message}`)
      }

      if (done) {
        // End of stream reached
        break
      }

      try {
        // Upload the current chunk
        await uploadChunk(value, file.name, chunkIndex++)
        uploaded += value.length

        // Calculate and report progress
        const pct = total > 0 ? (uploaded / total) * 100 : 100
        self.postMessage({
          type: 'progress',
          progress: pct,
          message: `Uploading chunk ${chunkIndex}...`,
        })
      } catch (uploadError) {
        // Handle chunk upload errors
        // Consider implementing retry logic here before throwing
        console.error(
          `Chunk ${chunkIndex - 1} upload failed for ${file.name}: ${uploadError.message}`,
        )
        throw new Error(`Chunk upload failed: ${uploadError.message}`)
      }
    }

    // Signal completion if all chunks were uploaded successfully
    self.postMessage({ type: 'complete', message: 'Upload finished successfully.' })
  } catch (err) {
    // Catch any errors from the process and report them back
    self.postMessage({
      type: 'error',
      message: err.message || 'An unknown error occurred in the worker.',
    })
  } finally {
    // IMPORTANT: Ensure the stream lock is released regardless of success or failure
    if (reader) {
      try {
        await reader.releaseLock()
      } catch (releaseLockError) {
        // Log error if releasing the lock fails, but don't block completion/error reporting
        console.error('Error releasing stream lock:', releaseLockError)
      }
    }
    // Optional: Close the worker if it's designed for single use
    // self.close();
  }
}

async function uploadChunk(chunk, filename, index) {
  const formData = new FormData()
  // Send chunk index for server-side reassembly
  formData.append('chunkIndex', index.toString())
  // Send the actual chunk data as a Blob
  formData.append('fileChunk', new Blob([chunk]), `${filename}.part${index}`)
  // Add any other necessary data (e.g., unique upload ID, total chunks)
  // formData.append('uploadId', 'unique-upload-identifier');
  // formData.append('totalChunks', totalChunks.toString());

  // Replace '/upload' with your actual back-end endpoint
  const res = await fetch('/upload', {
    method: 'POST',
    body: formData,
    // Consider adding an AbortSignal here for cancellation support
    // signal: abortController.signal
  })

  if (!res.ok) {
    // Try to get more details from the server response on failure
    let errorText = `Server responded with status ${res.status}`
    try {
      const serverError = await res.text()
      if (serverError) {
        errorText += `: ${serverError}`
      }
    } catch (e) {
      /* Ignore errors reading response body */
    }
    throw new Error(errorText)
  }

  // Optionally process the server response if needed
  // const result = await res.json();
  // return result;
}

class WorkerPool {
  constructor(script, size = navigator.hardwareConcurrency || 4) {
    this.workers = []
    this.idleWorkers = []
    this.taskQueue = []
    this.taskCallbacks = new Map() // Map task ID to callbacks

    console.log(`Initializing WorkerPool with size ${size}`)
    for (let i = 0; i < size; i++) {
      const worker = new Worker(script)
      worker.id = `worker_${i}`
      // Handle messages from the worker
      worker.onmessage = (e) => this.handleWorkerMessage(worker, e.data)
      // Handle errors occurring within the worker itself
      worker.onerror = (e) => this.handleWorkerError(worker, e)
      this.workers.push(worker)
      this.idleWorkers.push(worker)
    }
  }

  generateTaskId() {
    // Simple unique ID generator for tasks
    return Date.now().toString(36) + Math.random().toString(36).substring(2)
  }

  processFile(file, callbacks) {
    const taskId = this.generateTaskId()
    const task = { id: taskId, file }
    this.taskCallbacks.set(taskId, callbacks)

    const idleWorker = this.idleWorkers.pop()
    if (idleWorker) {
      // An idle worker is available, run the task immediately
      this.runTask(idleWorker, task)
    } else {
      // All workers are busy, add task to the queue
      this.taskQueue.push(task)
      console.log(
        `Worker pool busy. Queued task ${taskId} for ${file.name}. Queue size: ${this.taskQueue.length}`,
      )
    }
  }

  runTask(worker, task) {
    console.log(`Assigning task ${task.id} (${task.file.name}) to ${worker.id}`)
    worker.currentTask = task // Associate task metadata with the worker
    // Send the file object to the worker to start processing
    // For large files, consider if Transferable Objects are applicable/needed
    worker.postMessage(task.file)
  }

  handleWorkerMessage(worker, data) {
    const task = worker.currentTask
    if (!task) {
      console.warn(`Received message from worker ${worker.id} without an assigned task.`)
      return
    }

    const callbacks = this.taskCallbacks.get(task.id)
    if (!callbacks) {
      console.warn(`Received message for unknown or completed task ${task.id}`)
      return // Task might have been cancelled or already completed/failed
    }

    // Process messages based on their type
    switch (data.type) {
      case 'progress':
        if (callbacks.onProgress) callbacks.onProgress(data.progress, data.message)
        break
      case 'complete':
        if (callbacks.onComplete) callbacks.onComplete(data.message)
        this.finishTask(worker, task.id) // Mark task as finished
        break
      case 'error':
        if (callbacks.onError) callbacks.onError(data.message)
        this.finishTask(worker, task.id) // Also finish task on error
        break
      default:
        console.warn(`Received unknown message type from worker ${worker.id}:`, data.type)
    }
  }

  handleWorkerError(worker, errorEvent) {
    console.error(`Error in worker ${worker.id}:`, errorEvent.message, errorEvent)
    const task = worker.currentTask
    if (task) {
      const callbacks = this.taskCallbacks.get(task.id)
      if (callbacks && callbacks.onError) {
        // Report the error back via the task's error callback
        callbacks.onError(`Worker script error: ${errorEvent.message}`)
      }
      // Finish the task as it cannot continue
      this.finishTask(worker, task.id)
    } else {
      console.error(
        `Unhandled error in idle worker ${worker.id}. Consider worker replacement strategy.`,
      )
      // If an idle worker errors, you might want to remove it or try respawning
      // For simplicity, we just log it here.
    }
  }

  finishTask(worker, taskId) {
    const task = worker.currentTask
    if (task && task.id === taskId) {
      console.log(`Task ${taskId} (${task.file.name}) finished on ${worker.id}`)
      worker.currentTask = null // Disassociate task from worker
    }
    this.taskCallbacks.delete(taskId) // Remove callbacks

    // Check the queue for the next task
    if (this.taskQueue.length > 0) {
      const nextTask = this.taskQueue.shift()
      console.log(`Dequeuing task ${nextTask.id}. Queue size: ${this.taskQueue.length}`)
      this.runTask(worker, nextTask) // Assign the next task to this now-free worker
    } else {
      this.idleWorkers.push(worker) // No pending tasks, return worker to the idle pool
      console.log(`Worker ${worker.id} is now idle. Idle workers: ${this.idleWorkers.length}`)
    }
  }

  terminate() {
    console.log('Terminating worker pool...')
    this.workers.forEach((worker) => {
      console.log(`Terminating worker ${worker.id}`)
      worker.terminate()
    })
    // Clear internal state
    this.workers = []
    this.idleWorkers = []
    this.taskQueue = []
    this.taskCallbacks.clear()
  }
}

// Example using try...finally for pool termination in application code
const pool = new WorkerPool('upload-worker.js')
try {
  // ... use pool to process files ...
  // Example: await Promise.all(files.map(file => pool.processFileAsync(file)));
} finally {
  // Ensure pool is terminated regardless of success or failure
  console.log('Cleaning up worker pool...')
  pool.terminate()
}

// Example: Using AbortController for fetch cancellation
const ctrl = new AbortController()
const signal = ctrl.signal

// In uploadChunk function:
try {
  const res = await fetch('/upload', { method: 'POST', body: formData, signal })
  // ... handle response ...
} catch (err) {
  if (err.name === 'AbortError') {
    console.log('Chunk upload fetch aborted')
    // Propagate cancellation signal or specific error
    throw new Error('Upload cancelled by user')
  } else {
    console.error('Chunk upload fetch error:', err)
    throw err // Re-throw other errors
  }
}

// To cancel the upload associated with this controller:
// ctrl.abort();